GH-713: Add test for e2e transactional binder

See https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/713

The problem was already fixed in SIK.

Resolves #714

spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaBinderActuatorTests.java

@@ -48,6 +48,7 @@ import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.listener.AbstractMessageListenerContainer;
 import org.springframework.kafka.test.rule.EmbeddedKafkaRule;
 import org.springframework.messaging.MessageChannel;
+import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.junit4.SpringRunner;
 
 import static org.assertj.core.api.Assertions.assertThat;
@@ -64,6 +65,7 @@ import static org.assertj.core.api.Assertions.assertThat;
 @SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE, properties = "spring.cloud.stream.bindings.input.group="
 		+ KafkaBinderActuatorTests.TEST_CONSUMER_GROUP)
 // @checkstyle:on
+@DirtiesContext
 public class KafkaBinderActuatorTests {
 
 	static final String TEST_CONSUMER_GROUP = "testGroup-actuatorTests";

spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaBinderExtendedPropertiesTest.java

@@ -51,6 +51,7 @@ import org.springframework.kafka.test.rule.EmbeddedKafkaRule;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.SubscribableChannel;
 import org.springframework.messaging.handler.annotation.SendTo;
+import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.junit4.SpringRunner;
 
 import static org.assertj.core.api.Assertions.assertThat;
@@ -72,6 +73,7 @@ import static org.assertj.core.api.Assertions.assertThat;
 				+ "bindingSpecificPropertyShouldWinOverDefault",
 		"spring.cloud.stream.kafka.default.consumer.ackEachRecord=true",
 		"spring.cloud.stream.kafka.bindings.custom-in.consumer.ackEachRecord=false" })
+@DirtiesContext
 public class KafkaBinderExtendedPropertiesTest {
 
 	private static final String KAFKA_BROKERS_PROPERTY = "spring.cloud.stream.kafka.binder.brokers";

spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/ProducerOnlyTransactionTests.java

@@ -46,6 +46,7 @@ import org.springframework.kafka.test.utils.KafkaTestUtils;
 import org.springframework.kafka.transaction.KafkaTransactionManager;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.support.GenericMessage;
+import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.junit4.SpringRunner;
 import org.springframework.transaction.PlatformTransactionManager;
 import org.springframework.transaction.annotation.EnableTransactionManagement;
@@ -65,6 +66,7 @@ import static org.assertj.core.api.Assertions.assertThat;
 		"spring.cloud.stream.kafka.binder.transaction.transaction-id-prefix=tx.",
 		"spring.cloud.stream.kafka.binder.transaction.producer.configuration.retries=99",
 		"spring.cloud.stream.kafka.binder.transaction.producer.configuration.acks=all"})
+@DirtiesContext
 public class ProducerOnlyTransactionTests {
 
 	private static final String KAFKA_BROKERS_PROPERTY = "spring.cloud.stream.kafka.binder.brokers";

spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration2/ConsumerProducerTransactionTests.java

@@ -0,0 +1,144 @@
+/*
+ * Copyright 2019-2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.binder.kafka.integration2;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import kafka.server.KafkaConfig;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.ApplicationRunner;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.cloud.stream.annotation.EnableBinding;
+import org.springframework.cloud.stream.annotation.StreamListener;
+import org.springframework.cloud.stream.config.ListenerContainerCustomizer;
+import org.springframework.cloud.stream.messaging.Processor;
+import org.springframework.context.annotation.Bean;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.listener.AbstractMessageListenerContainer;
+import org.springframework.kafka.listener.DefaultAfterRollbackProcessor;
+import org.springframework.kafka.test.rule.EmbeddedKafkaRule;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.support.GenericMessage;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.junit4.SpringRunner;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Gary Russell
+ * @since 3.0
+ *
+ */
+@RunWith(SpringRunner.class)
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE, properties = {
+		"spring.kafka.consumer.properties.isolation.level=read_committed",
+		"spring.kafka.consumer.enable-auto-commit=false",
+		"spring.kafka.consumer.auto-offset-reset=earliest",
+		"spring.cloud.stream.bindings.input.destination=consumer.producer.txIn",
+		"spring.cloud.stream.bindings.input.group=consumer.producer.tx",
+		"spring.cloud.stream.bindings.input.consumer.max-attempts=1",
+		"spring.cloud.stream.bindings.output.destination=consumer.producer.txOut",
+		"spring.cloud.stream.kafka.binder.transaction.transaction-id-prefix=tx.",
+		"spring.cloud.stream.kafka.binder.transaction.producer.configuration.retries=99",
+		"spring.cloud.stream.kafka.binder.transaction.producer.configuration.acks=all"})
+@DirtiesContext
+public class ConsumerProducerTransactionTests {
+
+	private static final String KAFKA_BROKERS_PROPERTY = "spring.cloud.stream.kafka.binder.brokers";
+
+	@ClassRule
+	public static EmbeddedKafkaRule embeddedKafka = new EmbeddedKafkaRule(1, true, "consumer.producer.txOut")
+			.brokerProperty(KafkaConfig.TransactionsTopicReplicationFactorProp(), "1")
+			.brokerProperty(KafkaConfig.TransactionsTopicMinISRProp(), "1");
+
+	@Autowired
+	private Config config;
+
+	@BeforeClass
+	public static void setup() {
+		System.setProperty(KAFKA_BROKERS_PROPERTY,
+				embeddedKafka.getEmbeddedKafka().getBrokersAsString());
+		System.setProperty("spring.kafka.bootstrap-servers",
+				embeddedKafka.getEmbeddedKafka().getBrokersAsString());
+	}
+
+	@AfterClass
+	public static void clean() {
+		System.clearProperty(KAFKA_BROKERS_PROPERTY);
+		System.clearProperty("spring.kafka.bootstrap-servers");
+	}
+
+	@Test
+	public void testProducerRunsInConsumerTransaction() throws InterruptedException {
+		assertThat(this.config.latch.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(this.config.outs).containsExactlyInAnyOrder("ONE", "THREE");
+	}
+
+	@EnableBinding(Processor.class)
+	@EnableAutoConfiguration
+	public static class Config {
+
+		final List<String> outs = new ArrayList<>();
+
+		final CountDownLatch latch = new CountDownLatch(2);
+
+		@Autowired
+		private MessageChannel output;
+
+		@KafkaListener(id = "test.cons.prod", topics = "consumer.producer.txOut")
+		public void listenOut(String in) {
+			this.outs.add(in);
+			this.latch.countDown();
+		}
+
+		@StreamListener(Processor.INPUT)
+		public void listenIn(String in) {
+			this.output.send(new GenericMessage<>(in.toUpperCase()));
+			if (in.equals("two")) {
+				throw new RuntimeException("fail");
+			}
+		}
+
+		@Bean
+		public ApplicationRunner runner(KafkaTemplate<byte[], byte[]> template) {
+			return args -> {
+				template.send("consumer.producer.txIn", "one".getBytes());
+				template.send("consumer.producer.txIn", "two".getBytes());
+				template.send("consumer.producer.txIn", "three".getBytes());
+			};
+		}
+
+		@Bean
+		public ListenerContainerCustomizer<AbstractMessageListenerContainer<?, ?>> customizer() {
+			return (container, dest, group) -> container
+					.setAfterRollbackProcessor(new DefaultAfterRollbackProcessor<>(0));
+		}
+
+	}
+
+}

spring-cloud-stream-binder-kafka/src/test/resources/logback.xml

@@ -4,7 +4,7 @@
             <pattern>%d{ISO8601} %5p %t %c{2}:%L - %m%n</pattern>
         </encoder>
     </appender>
-    <logger name="org.apache.kafka" level="DEBUG"/>
+    <logger name="org.apache.kafka" level="INFO"/>
     <logger name="org.springframework.integration.kafka" level="INFO"/>
     <logger name="org.springframework.kafka" level="INFO"/>
     <logger name="org.springframework.cloud.stream" level="INFO" />