diff --git a/spring-cloud-stream-binder-kafka-streams/pom.xml b/spring-cloud-stream-binder-kafka-streams/pom.xml index d3ad4d3cc..4e8a2f9f8 100644 --- a/spring-cloud-stream-binder-kafka-streams/pom.xml +++ b/spring-cloud-stream-binder-kafka-streams/pom.xml @@ -74,13 +74,13 @@ org.apache.kafka - kafka_2.11 + kafka_2.12 ${kafka.version} test org.apache.kafka - kafka_2.11 + kafka_2.12 ${kafka.version} test test diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java index a12dce095..584f720a0 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java @@ -18,6 +18,7 @@ package org.springframework.cloud.stream.binder.kafka.streams; import java.util.Collection; import java.util.HashMap; +import java.util.List; import java.util.Map; import java.util.Properties; import java.util.stream.Collectors; @@ -172,6 +173,13 @@ public class KafkaStreamsBinderSupportAutoConfiguration { configProperties.getKafkaConnectionString()); } } + else if (bootstrapServerConfig instanceof List) { + List bootStrapCollection = (List) bootstrapServerConfig; + if (bootStrapCollection.size() == 1 && bootStrapCollection.get(0).equals("localhost:9092")) { + properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, + configProperties.getKafkaConnectionString()); + } + } } String binderProvidedApplicationId = configProperties.getApplicationId(); diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/bootstrap/KafkaStreamsBinderBootstrapTest.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/bootstrap/KafkaStreamsBinderBootstrapTest.java index 4078fb112..0c1cb267a 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/bootstrap/KafkaStreamsBinderBootstrapTest.java +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/bootstrap/KafkaStreamsBinderBootstrapTest.java @@ -27,7 +27,7 @@ import org.springframework.cloud.stream.annotation.EnableBinding; import org.springframework.cloud.stream.annotation.Input; import org.springframework.cloud.stream.annotation.StreamListener; import org.springframework.context.ConfigurableApplicationContext; -import org.springframework.kafka.test.rule.KafkaEmbedded; +import org.springframework.kafka.test.rule.EmbeddedKafkaRule; /** * @author Soby Chacko @@ -35,7 +35,7 @@ import org.springframework.kafka.test.rule.KafkaEmbedded; public class KafkaStreamsBinderBootstrapTest { @ClassRule - public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, 10); + public static EmbeddedKafkaRule embeddedKafka = new EmbeddedKafkaRule(1, true, 10); @Test public void testKafkaStreamsBinderWithCustomEnvironmentCanStart() { @@ -48,10 +48,7 @@ public class KafkaStreamsBinderBootstrapTest { "--spring.cloud.stream.binders.kBind1.type=kstream", "--spring.cloud.stream.binders.kBind1.environment" + ".spring.cloud.stream.kafka.streams.binder.brokers" - + "=" + embeddedKafka.getBrokersAsString(), - "--spring.cloud.stream.binders.kBind1.environment.spring" - + ".cloud.stream.kafka.streams.binder.zkNodes=" - + embeddedKafka.getZookeeperConnectionString()); + + "=" + embeddedKafka.getEmbeddedKafka().getBrokersAsString()); applicationContext.close(); } @@ -64,9 +61,7 @@ public class KafkaStreamsBinderBootstrapTest { + "=testKafkaStreamsBinderWithStandardConfigurationCanStart", "--spring.cloud.stream.bindings.input.destination=foo", "--spring.cloud.stream.kafka.streams.binder.brokers=" - + embeddedKafka.getBrokersAsString(), - "--spring.cloud.stream.kafka.streams.binder.zkNodes=" - + embeddedKafka.getZookeeperConnectionString()); + + embeddedKafka.getEmbeddedKafka().getBrokersAsString()); applicationContext.close(); } diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionStateStoreTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionStateStoreTests.java index 616c1c7be..6880f6380 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionStateStoreTests.java +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionStateStoreTests.java @@ -145,7 +145,7 @@ public class KafkaStreamsFunctionStateStoreTests { public StoreBuilder otherStore() { return Stores.windowStoreBuilder( Stores.persistentWindowStore("other-store", - 1L, 3, 3L, false), Serdes.Long(), + 3L, 3, 3L, false), Serdes.Long(), Serdes.Long()); } } diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderWordCountIntegrationTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderWordCountIntegrationTests.java index e813fad60..17b13323f 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderWordCountIntegrationTests.java +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderWordCountIntegrationTests.java @@ -117,9 +117,7 @@ public class KafkaStreamsBinderWordCountIntegrationTests { "--spring.cloud.stream.kafka.streams.timeWindow.length=5000", "--spring.cloud.stream.kafka.streams.timeWindow.advanceBy=0", "--spring.cloud.stream.kafka.streams.binder.brokers=" - + embeddedKafka.getBrokersAsString(), - "--spring.cloud.stream.kafka.streams.binder.zkNodes=" - + embeddedKafka.getZookeeperConnectionString())) { + + embeddedKafka.getBrokersAsString())) { receiveAndValidate(context); } } diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsStateStoreIntegrationTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsStateStoreIntegrationTests.java index ccf562430..d7be3249b 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsStateStoreIntegrationTests.java +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsStateStoreIntegrationTests.java @@ -154,7 +154,7 @@ public class KafkaStreamsStateStoreIntegrationTests { boolean processed; @StreamListener("input") - @KafkaStreamsStateStore(name = "mystate", type = KafkaStreamsStateStoreProperties.StoreType.WINDOW, lengthMs = 300000) + @KafkaStreamsStateStore(name = "mystate", type = KafkaStreamsStateStoreProperties.StoreType.WINDOW, lengthMs = 300000, retentionMs = 300000) @SuppressWarnings({ "deprecation", "unchecked" }) public void process(KStream input) { @@ -189,7 +189,7 @@ public class KafkaStreamsStateStoreIntegrationTests { boolean processed; @StreamListener - @KafkaStreamsStateStore(name = "mystate", type = KafkaStreamsStateStoreProperties.StoreType.WINDOW, lengthMs = 300000) + @KafkaStreamsStateStore(name = "mystate", type = KafkaStreamsStateStoreProperties.StoreType.WINDOW, lengthMs = 300000, retentionMs = 300000) @SuppressWarnings({ "deprecation", "unchecked" }) public void process(@Input("input1")KStream input, @Input("input2")KStream input2) { diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/resources/logback.xml b/spring-cloud-stream-binder-kafka-streams/src/test/resources/logback.xml index 98e9018df..414797f87 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/resources/logback.xml +++ b/spring-cloud-stream-binder-kafka-streams/src/test/resources/logback.xml @@ -4,6 +4,7 @@ %d{ISO8601} %5p %t %c{2}:%L - %m%n + diff --git a/spring-cloud-stream-binder-kafka/pom.xml b/spring-cloud-stream-binder-kafka/pom.xml index 149353118..0eb1f645e 100644 --- a/spring-cloud-stream-binder-kafka/pom.xml +++ b/spring-cloud-stream-binder-kafka/pom.xml @@ -69,13 +69,13 @@ org.apache.kafka - kafka_2.11 + kafka_2.12 ${kafka.version} test org.apache.kafka - kafka_2.11 + kafka_2.12 ${kafka.version} test test diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java index 1659064dd..e7d0e1564 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java @@ -348,6 +348,9 @@ public class KafkaMessageChannelBinder extends if (this.producerListener != null) { kafkaTemplate.setProducerListener(this.producerListener); } + if (this.transactionManager != null) { + kafkaTemplate.setTransactionIdPrefix(configurationProperties.getTransaction().getTransactionIdPrefix()); + } ProducerConfigurationMessageHandler handler = new ProducerConfigurationMessageHandler( kafkaTemplate, destination.getName(), producerProperties, producerFB); if (errorChannel != null) { diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java index 475692b5e..6c888fc3f 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java @@ -130,6 +130,7 @@ import org.springframework.messaging.support.ErrorMessage; import org.springframework.messaging.support.GenericMessage; import org.springframework.messaging.support.MessageBuilder; import org.springframework.util.Assert; +import org.springframework.util.MimeType; import org.springframework.util.MimeTypeUtils; import org.springframework.util.concurrent.ListenableFuture; import org.springframework.util.concurrent.SettableListenableFuture; @@ -363,9 +364,9 @@ public class KafkaBinderTests extends .get(MessageHeaders.CONTENT_TYPE)) .isEqualTo(MimeTypeUtils.TEXT_PLAIN); Assertions.assertThat(inboundMessageRef.get().getHeaders().get("foo")) - .isInstanceOf(String.class); - String actual = (String) inboundMessageRef.get().getHeaders().get("foo"); - Assertions.assertThat(actual).isEqualTo(MimeTypeUtils.TEXT_PLAIN.toString()); + .isInstanceOf(MimeType.class); + MimeType actual = (MimeType) inboundMessageRef.get().getHeaders().get("foo"); + Assertions.assertThat(actual).isEqualTo(MimeTypeUtils.TEXT_PLAIN); producerBinding.unbind(); consumerBinding.unbind(); } diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTransactionTests.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTransactionTests.java index 57953f66e..b83b64d1c 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTransactionTests.java +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTransactionTests.java @@ -17,9 +17,12 @@ package org.springframework.cloud.stream.binder.kafka; import java.util.Collections; +import java.util.HashMap; +import java.util.Map; import org.apache.kafka.clients.producer.Callback; import org.apache.kafka.clients.producer.Producer; +import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.TopicPartition; import org.junit.ClassRule; @@ -54,8 +57,15 @@ import static org.mockito.Mockito.spy; */ public class KafkaTransactionTests { + private static Map brokerProperties = new HashMap<>(); + + static { + brokerProperties.put("transaction.state.log.replication.factor", "1"); + brokerProperties.put("transaction.state.log.min.isr", "1"); + } + @ClassRule - public static final EmbeddedKafkaRule embeddedKafka = new EmbeddedKafkaRule(1); + public static final EmbeddedKafkaRule embeddedKafka = new EmbeddedKafkaRule(1).brokerProperties(brokerProperties); @SuppressWarnings({ "rawtypes", "unchecked" }) @Test @@ -71,6 +81,12 @@ public class KafkaTransactionTests { configurationProperties, kafkaProperties); provisioningProvider.setMetadataRetryOperations(new RetryTemplate()); final Producer mockProducer = mock(Producer.class); + + KafkaProducerProperties extension1 = configurationProperties + .getTransaction().getProducer().getExtension(); + extension1.getConfiguration().put(ProducerConfig.RETRIES_CONFIG, "1"); + extension1.getConfiguration().put(ProducerConfig.ACKS_CONFIG, "all"); + willReturn(Collections.singletonList(new TopicPartition("foo", 0))) .given(mockProducer).partitionsFor(anyString()); KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder( @@ -83,7 +99,7 @@ public class KafkaTransactionTests { DefaultKafkaProducerFactory producerFactory = spy( super.getProducerFactory(transactionIdPrefix, producerProperties)); - willReturn(mockProducer).given(producerFactory).createProducer(); + willReturn(mockProducer).given(producerFactory).createProducer("foo-"); return producerFactory; } diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/bootstrap/MultiBinderMeterRegistryTest.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/bootstrap/MultiBinderMeterRegistryTest.java index 88747c011..2a26123e4 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/bootstrap/MultiBinderMeterRegistryTest.java +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/bootstrap/MultiBinderMeterRegistryTest.java @@ -26,7 +26,7 @@ import org.springframework.boot.builder.SpringApplicationBuilder; import org.springframework.cloud.stream.annotation.EnableBinding; import org.springframework.cloud.stream.messaging.Sink; import org.springframework.context.ConfigurableApplicationContext; -import org.springframework.kafka.test.rule.KafkaEmbedded; +import org.springframework.kafka.test.rule.EmbeddedKafkaRule; import static org.assertj.core.api.Assertions.assertThat; @@ -36,7 +36,7 @@ import static org.assertj.core.api.Assertions.assertThat; public class MultiBinderMeterRegistryTest { @ClassRule - public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, 10); + public static EmbeddedKafkaRule embeddedKafka = new EmbeddedKafkaRule(1, true, 10); @Test public void testMetricsWorkWithMultiBinders() { @@ -48,7 +48,7 @@ public class MultiBinderMeterRegistryTest { "--spring.cloud.stream.binders.inbound.type=kafka", "--spring.cloud.stream.binders.inbound.environment" + ".spring.cloud.stream.kafka.binder.brokers" + "=" - + embeddedKafka.getBrokersAsString()); + + embeddedKafka.getEmbeddedKafka().getBrokersAsString()); final MeterRegistry meterRegistry = applicationContext.getBean(MeterRegistry.class);