Transaction sample changes

Add @Transactional on the runtime code in the lambda expression,
not on the functional bean. If it is set on the function bean, then
that will only be invoked at application initialization, thus losing
the transactional semantics on the actual runtme code.

Resolves https://github.com/spring-cloud/spring-cloud-stream-samples/issues/212

transaction-kafka-samples/transaction-spring-data-processor/src/main/java/sample/processor/ProcessorApplication.java

@@ -30,6 +30,7 @@ import org.springframework.cloud.stream.config.ListenerContainerCustomizer;
 import org.springframework.context.annotation.Bean;
 import org.springframework.kafka.listener.AbstractMessageListenerContainer;
 import org.springframework.kafka.listener.DefaultAfterRollbackProcessor;
+import org.springframework.stereotype.Component;
 import org.springframework.transaction.annotation.EnableTransactionManagement;
 import org.springframework.util.backoff.FixedBackOff;
 
@@ -42,11 +43,10 @@ public class ProcessorApplication {
 
 	private Logger logger = LoggerFactory.getLogger(this.getClass().getName());
 
-    private AtomicBoolean shouldFail= new AtomicBoolean(false);
+	private AtomicBoolean shouldFail = new AtomicBoolean(false);
 	private PersonRepository repository;
 
 	public ProcessorApplication(PersonRepository repository) {
-
 		this.repository = repository;
 	}
 
@@ -54,31 +54,9 @@ public class ProcessorApplication {
 		SpringApplication.run(ProcessorApplication.class, args);
 	}
 
-    @Transactional
 	@Bean
-    public Function<PersonEvent, PersonEvent> process() {
-        return pe -> {
-            logger.info("Received event={}", pe);
-            Person person = new Person();
-            person.setName(pe.getName());
-
-            if (shouldFail.get()) {
-                shouldFail.set(false);
-                throw new RuntimeException("Simulated network error");
-            } else {
-                //We fail every other request as a test
-                shouldFail.set(true);
-            }
-            logger.info("Saving person={}", person);
-
-            Person savedPerson = repository.save(person);
-
-            PersonEvent event = new PersonEvent();
-            event.setName(savedPerson.getName());
-            event.setType("PersonSaved");
-            logger.info("Sent event={}", event);
-            return event;
-        };
+	public Function<PersonEvent, PersonEvent> process(TxCode txCode) {
+		return pe -> txCode.run(pe);
 	}
 
 	@Bean
@@ -119,4 +97,33 @@ public class ProcessorApplication {
 					.toString();
 		}
 	}
+
+	@Component
+	class TxCode {
+
+		@Transactional
+		PersonEvent run(PersonEvent pe) {
+			logger.info("Received event={}", pe);
+			Person person = new Person();
+			person.setName(pe.getName());
+
+			if (shouldFail.get()) {
+				shouldFail.set(false);
+				throw new RuntimeException("Simulated network error");
+			}
+			else {
+				//We fail every other request as a test
+				shouldFail.set(true);
+			}
+			logger.info("Saving person={}", person);
+
+			Person savedPerson = repository.save(person);
+
+			PersonEvent event = new PersonEvent();
+			event.setName(savedPerson.getName());
+			event.setType("PersonSaved");
+			logger.info("Sent event={}", event);
+			return event;
+		}
+	}
 }