diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaBinderActuatorTests.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaBinderActuatorTests.java index 0358fa9dd..3ed861b44 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaBinderActuatorTests.java +++ b/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"; diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaBinderExtendedPropertiesTest.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaBinderExtendedPropertiesTest.java index 92902b0b2..db2064c0d 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaBinderExtendedPropertiesTest.java +++ b/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"; diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/ProducerOnlyTransactionTests.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/ProducerOnlyTransactionTests.java index d43af0486..b48cacbef 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/ProducerOnlyTransactionTests.java +++ b/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"; diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration2/ConsumerProducerTransactionTests.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration2/ConsumerProducerTransactionTests.java new file mode 100644 index 000000000..51f726610 --- /dev/null +++ b/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 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 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> customizer() { + return (container, dest, group) -> container + .setAfterRollbackProcessor(new DefaultAfterRollbackProcessor<>(0)); + } + + } + +} diff --git a/spring-cloud-stream-binder-kafka/src/test/resources/logback.xml b/spring-cloud-stream-binder-kafka/src/test/resources/logback.xml index ba06d2031..77ab1d058 100644 --- a/spring-cloud-stream-binder-kafka/src/test/resources/logback.xml +++ b/spring-cloud-stream-binder-kafka/src/test/resources/logback.xml @@ -4,7 +4,7 @@ %d{ISO8601} %5p %t %c{2}:%L - %m%n - +