diff --git a/pom.xml b/pom.xml index ae6272583..3d1218dcd 100644 --- a/pom.xml +++ b/pom.xml @@ -7,7 +7,7 @@ org.springframework.cloud spring-cloud-stream-parent - 1.1.0.M1 + 1.1.0.BUILD-SNAPSHOT @@ -19,8 +19,8 @@ spring-cloud-stream-binder-kafka spring-cloud-starter-stream-kafka - spring-cloud-stream-binder-kafka-test-support spring-cloud-stream-binder-kafka-docs + spring-cloud-stream-binder-kafka-0.10-test diff --git a/spring-cloud-starter-stream-kafka/pom.xml b/spring-cloud-starter-stream-kafka/pom.xml index b1d327de7..ae85746a4 100644 --- a/spring-cloud-starter-stream-kafka/pom.xml +++ b/spring-cloud-starter-stream-kafka/pom.xml @@ -15,12 +15,42 @@ ${basedir}/../.. + 0.9.0.1 + 1.0.3.RELEASE + 2.0.1.RELEASE + org.springframework.cloud spring-cloud-stream-binder-kafka - 1.1.0.BUILD-SNAPSHOT + ${project.version} + + + org.springframework.kafka + spring-kafka + ${spring-kafka.version} + + + org.apache.kafka + kafka_2.11 + ${kafka.version} + + + org.slf4j + slf4j-log4j12 + + + + + org.apache.kafka + kafka-clients + ${kafka.version} + + + org.springframework.integration + spring-integration-kafka + ${spring-integration-kafka.version} diff --git a/spring-cloud-starter-stream-kafka/src/main/resources/META-INF/spring.provides b/spring-cloud-starter-stream-kafka/src/main/resources/META-INF/spring.provides index cc7cb9cc2..d9ca8793e 100644 --- a/spring-cloud-starter-stream-kafka/src/main/resources/META-INF/spring.provides +++ b/spring-cloud-starter-stream-kafka/src/main/resources/META-INF/spring.provides @@ -1 +1 @@ -provides: spring-cloud-starter-stream-kafka \ No newline at end of file +provides: spring-cloud-starter-stream-kafka-0.10 \ No newline at end of file diff --git a/spring-cloud-stream-binder-kafka-0.10-test/pom.xml b/spring-cloud-stream-binder-kafka-0.10-test/pom.xml new file mode 100644 index 000000000..92aa97f58 --- /dev/null +++ b/spring-cloud-stream-binder-kafka-0.10-test/pom.xml @@ -0,0 +1,80 @@ + + + 4.0.0 + + org.springframework.cloud + spring-cloud-stream-binder-kafka-parent + 1.1.0.BUILD-SNAPSHOT + + spring-cloud-stream-binder-kafka-0.10-test + Spring Cloud Stream Kafka Binder 0.10 Tests + http://projects.spring.io/spring-cloud + + Pivotal Software, Inc. + http://www.spring.io + + + ${basedir}/../.. + 0.10.0.0 + 1.1.0.M1 + 2.0.1.RELEASE + + + + + org.springframework.cloud + spring-cloud-stream-binder-kafka + ${project.version} + test + + + org.springframework.kafka + spring-kafka + ${spring-kafka.version} + test + + + org.apache.kafka + kafka_2.11 + ${kafka.version} + test + + + org.slf4j + slf4j-log4j12 + + + + + org.apache.kafka + kafka-clients + ${kafka.version} + test + + + org.springframework.kafka + spring-kafka-test + test + ${spring-kafka.version} + + + org.springframework.integration + spring-integration-kafka + ${spring-integration-kafka.version} + + + org.springframework.cloud + spring-cloud-stream-binder-kafka + ${project.version} + test-jar + test + + + org.springframework.cloud + spring-cloud-stream-binder-test + test + + + + + diff --git a/spring-cloud-stream-binder-kafka-0.10-test/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka10BinderTests.java b/spring-cloud-stream-binder-kafka-0.10-test/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka10BinderTests.java new file mode 100644 index 000000000..26f3533bc --- /dev/null +++ b/spring-cloud-stream-binder-kafka-0.10-test/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka10BinderTests.java @@ -0,0 +1,179 @@ +/* + * Copyright 2014-2016 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 + * + * http://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; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Properties; + +import kafka.utils.ZKStringSerializer$; +import kafka.utils.ZkUtils; +import org.I0Itec.zkclient.ZkClient; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.common.serialization.ByteArrayDeserializer; +import org.apache.kafka.common.serialization.Deserializer; +import org.junit.Before; +import org.junit.ClassRule; + +import org.springframework.cloud.stream.binder.Binder; +import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; +import org.springframework.cloud.stream.binder.ExtendedProducerProperties; +import org.springframework.cloud.stream.binder.Spy; +import org.springframework.cloud.stream.binder.kafka.admin.Kafka10AdminUtilsOperation; +import org.springframework.cloud.stream.binder.kafka.configuration.KafkaBinderConfigurationProperties; +import org.springframework.kafka.core.ConsumerFactory; +import org.springframework.kafka.core.DefaultKafkaConsumerFactory; +import org.springframework.kafka.support.KafkaHeaders; +import org.springframework.kafka.test.core.BrokerAddress; +import org.springframework.kafka.test.rule.KafkaEmbedded; +import org.springframework.retry.RetryOperations; + +/** + * Integration tests for the {@link KafkaMessageChannelBinder}. + * + * @author Eric Bottard + * @author Marius Bogoevici + * @author Mark Fisher + * @author Ilayaperumal Gopinathan + */ +public class Kafka10BinderTests extends KafkaBinderTests { + + private final String CLASS_UNDER_TEST_NAME = KafkaMessageChannelBinder.class.getSimpleName(); + + @ClassRule + public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, 10); + + private Kafka10TestBinder binder; + + private Kafka10AdminUtilsOperation adminUtilsOperation = new Kafka10AdminUtilsOperation(); + + @Override + protected void binderBindUnbindLatency() throws InterruptedException { + Thread.sleep(500); + } + + @Override + protected Kafka10TestBinder getBinder() { + if (binder == null) { + KafkaBinderConfigurationProperties binderConfiguration = createConfigurationProperties(); + binder = new Kafka10TestBinder(binderConfiguration); + } + return binder; + } + + protected KafkaBinderConfigurationProperties createConfigurationProperties() { + KafkaBinderConfigurationProperties binderConfiguration = new KafkaBinderConfigurationProperties(); + BrokerAddress[] brokerAddresses = embeddedKafka.getBrokerAddresses(); + List bAddresses = new ArrayList<>(); + for (BrokerAddress bAddress : brokerAddresses) { + bAddresses.add(bAddress.toString()); + } + String[] foo = new String[bAddresses.size()]; + binderConfiguration.setBrokers(bAddresses.toArray(foo)); + binderConfiguration.setZkNodes(embeddedKafka.getZookeeperConnectionString()); + return binderConfiguration; + } + + @Override + protected int partitionSize(String topic) { + return consumerFactory().createConsumer().partitionsFor(topic).size(); + } + + @Override + @SuppressWarnings("unchecked") + protected void setMetadataRetryOperations(Binder binder, RetryOperations retryOperations) { + ((Kafka10TestBinder) binder).getBinder().setMetadataRetryOperations(retryOperations); + } + + @Override + protected ZkUtils getZkUtils(KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties) { + final ZkClient zkClient = new ZkClient(kafkaBinderConfigurationProperties.getZkConnectionString(), + kafkaBinderConfigurationProperties.getZkSessionTimeout(), kafkaBinderConfigurationProperties.getZkConnectionTimeout(), + ZKStringSerializer$.MODULE$); + + return new ZkUtils(zkClient, null, false); + } + + @Override + protected void invokeCreateTopic(ZkUtils zkUtils, String topic, int partitions, int replicationFactor, Properties topicConfig) { + adminUtilsOperation.invokeCreateTopic(zkUtils, topic, partitions, replicationFactor, new Properties()); + } + + @Override + protected int invokePartitionSize(String topic, ZkUtils zkUtils) { + return adminUtilsOperation.partitionSize(topic, zkUtils); + } + + @Override + protected ExtendedConsumerProperties createConsumerProperties() { + return new ExtendedConsumerProperties<>(new KafkaConsumerProperties()); + } + + @Override + protected ExtendedProducerProperties createProducerProperties() { + return new ExtendedProducerProperties<>(new KafkaProducerProperties()); + } + + @Override + public String getKafkaOffsetHeaderKey() { + return KafkaHeaders.OFFSET; + } + + @Override + protected Binder getBinder(KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties) { + return new Kafka10TestBinder(kafkaBinderConfigurationProperties); + } + + @Before + public void init() { + String multiplier = System.getenv("KAFKA_TIMEOUT_MULTIPLIER"); + if (multiplier != null) { + timeoutMultiplier = Double.parseDouble(multiplier); + } + } + + @Override + protected boolean usesExplicitRouting() { + return false; + } + + @Override + protected String getClassUnderTestName() { + return CLASS_UNDER_TEST_NAME; + } + + @Override + public Spy spyOn(final String name) { + throw new UnsupportedOperationException("'spyOn' is not used by Kafka tests"); + } + + + private ConsumerFactory consumerFactory() { + Map props = new HashMap<>(); + KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); + props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, configurationProperties.getKafkaConnectionString()); + props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); + props.put(ConsumerConfig.GROUP_ID_CONFIG, "TEST-CONSUMER-GROUP"); + Deserializer valueDecoder = new ByteArrayDeserializer(); + Deserializer keyDecoder = new ByteArrayDeserializer(); + + return new DefaultKafkaConsumerFactory<>(props, keyDecoder, valueDecoder); + } + +} diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTestBinder.java b/spring-cloud-stream-binder-kafka-0.10-test/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka10TestBinder.java similarity index 52% rename from spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTestBinder.java rename to spring-cloud-stream-binder-kafka-0.10-test/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka10TestBinder.java index e551bd8b0..82329db49 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTestBinder.java +++ b/spring-cloud-stream-binder-kafka-0.10-test/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka10TestBinder.java @@ -16,20 +16,9 @@ package org.springframework.cloud.stream.binder.kafka; -import java.util.List; - -import com.esotericsoftware.kryo.Kryo; -import com.esotericsoftware.kryo.Registration; - -import org.springframework.cloud.stream.binder.AbstractTestBinder; -import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; -import org.springframework.cloud.stream.binder.ExtendedProducerProperties; -import org.springframework.cloud.stream.binder.kafka.config.KafkaBinderConfigurationProperties; +import org.springframework.cloud.stream.binder.kafka.admin.Kafka10AdminUtilsOperation; +import org.springframework.cloud.stream.binder.kafka.configuration.KafkaBinderConfigurationProperties; import org.springframework.context.support.GenericApplicationContext; -import org.springframework.integration.codec.Codec; -import org.springframework.integration.codec.kryo.KryoRegistrar; -import org.springframework.integration.codec.kryo.PojoCodec; -import org.springframework.integration.tuple.TupleKryoRegistrar; import org.springframework.kafka.support.LoggingProducerListener; import org.springframework.kafka.support.ProducerListener; @@ -41,10 +30,9 @@ import org.springframework.kafka.support.ProducerListener; * @author Gary Russell * @author Soby Chacko */ -public class KafkaTestBinder extends - AbstractTestBinder, ExtendedProducerProperties> { +public class Kafka10TestBinder extends AbstractKafkaTestBinder { - public KafkaTestBinder(KafkaBinderConfigurationProperties binderConfiguration) { + public Kafka10TestBinder(KafkaBinderConfigurationProperties binderConfiguration) { try { KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder(binderConfiguration); binder.setCodec(getCodec()); @@ -53,6 +41,7 @@ public class KafkaTestBinder extends GenericApplicationContext context = new GenericApplicationContext(); context.refresh(); binder.setApplicationContext(context); + binder.setAdminUtilsOperation(new Kafka10AdminUtilsOperation()); binder.afterPropertiesSet(); this.setBinder(binder); } @@ -61,27 +50,4 @@ public class KafkaTestBinder extends } } - @Override - public void cleanup() { - // do nothing - the rule will take care of that - } - - private static Codec getCodec() { - return new PojoCodec(new TupleRegistrar()); - } - - private static class TupleRegistrar implements KryoRegistrar { - private final TupleKryoRegistrar delegate = new TupleKryoRegistrar(); - - @Override - public void registerTypes(Kryo kryo) { - this.delegate.registerTypes(kryo); - } - - @Override - public List getRegistrations() { - return this.delegate.getRegistrations(); - } - } - } diff --git a/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/overview.adoc b/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/overview.adoc index f4ee39838..ef6b8b823 100644 --- a/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/overview.adoc +++ b/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/overview.adoc @@ -120,6 +120,12 @@ Default: `2097152`. The following properties are available for Kafka consumers only and must be prefixed with `spring.cloud.stream.kafka.bindings..consumer.`. +autoRebalanceEnabled:: + When this is enabled, topic partitions will be automatically rebalanced between various consumers by the broker. + If set to `false`, it will trigger partitions to be statically allocated by the binder. +When setting false, in order for static allocation of partitions to take place, it also needs both `spring.cloud.stream.instaceCount` and +'spring.cloud.stream.instanceIndex' properties set appropriately. The property `spring.cloud.stream.instaceCount` must be greater than 1 in this case. + autoCommitOffset:: Whether to autocommit offsets when a message has been processed. If set to `false`, an `Acknowledgment` header will be available in the message headers for late acknowledgment. @@ -226,4 +232,36 @@ Here is an example of launching a Spring Cloud Stream application with SASL and Exercise caution when using the `autoCreateTopics` and `autoAddPartitions` if using Kerberos. Usually applications may use principals that do not have administrative rights in Kafka and Zookeeper, and relying on Spring Cloud Stream to create/modify topics may fail. In secure environments, we strongly recommend creating topics and managing ACLs administratively using Kafka tooling. +==== + +[NOTE] +==== +In addition to supporting 0.9 based clients, Kafka binder can also work with 0.10 libs. In order to support this, when you create the project that contains your application, include `spring-cloud-starter-stream-kafka` as you normally would do for 0.9 based applications. +Then add these dependencies at the top of the dependencies section in the pom.xml file. + +[source,xml] +---- + + org.springframework.kafka + spring-kafka + 1.1.0.M1 + + + org.springframework.integration + spring-integration-kafka + 2.0.1.RELEASE + + + org.apache.kafka + kafka_2.11 + 0.10.0.0 + + + org.slf4j + slf4j-log4j12 + + + +---- + ==== \ No newline at end of file diff --git a/spring-cloud-stream-binder-kafka-test-support/pom.xml b/spring-cloud-stream-binder-kafka-test-support/pom.xml deleted file mode 100644 index 452f44f09..000000000 --- a/spring-cloud-stream-binder-kafka-test-support/pom.xml +++ /dev/null @@ -1,24 +0,0 @@ - - - 4.0.0 - - org.springframework.cloud - spring-cloud-stream-binder-kafka-parent - 1.1.0.BUILD-SNAPSHOT - - spring-cloud-stream-binder-kafka-test-support - Kafka related test classes - - - - junit - junit - compile - - - org.springframework.kafka - spring-kafka-test - ${spring-kafka.version} - - - diff --git a/spring-cloud-stream-binder-kafka/pom.xml b/spring-cloud-stream-binder-kafka/pom.xml index fa07654ff..4d943ae71 100644 --- a/spring-cloud-stream-binder-kafka/pom.xml +++ b/spring-cloud-stream-binder-kafka/pom.xml @@ -13,6 +13,12 @@ 1.1.0.BUILD-SNAPSHOT + + 0.9.0.1 + 1.0.3.RELEASE + 2.0.1.RELEASE + + org.springframework.boot @@ -37,12 +43,6 @@ spring-cloud-stream-binder-test test - - org.springframework.cloud - spring-cloud-stream-binder-kafka-test-support - test - 1.1.0.BUILD-SNAPSHOT - org.springframework.integration spring-integration-kafka @@ -58,6 +58,17 @@ org.springframework.kafka spring-kafka ${spring-kafka.version} + true + + + org.apache.kafka + kafka_2.11 + true + + + org.apache.kafka + kafka-clients + true org.springframework.kafka @@ -65,14 +76,6 @@ test ${spring-kafka.version} - - org.apache.kafka - kafka_2.11 - - - org.apache.kafka - kafka-clients - org.apache.kafka kafka_2.11 @@ -111,4 +114,21 @@ + + + + + org.apache.maven.plugins + maven-jar-plugin + 3.0.2 + + + + test-jar + + + + + + diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicator.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicator.java index 64a09ad1a..8e35d5fe3 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicator.java +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicator.java @@ -29,10 +29,11 @@ import org.apache.kafka.common.serialization.ByteArrayDeserializer; import org.springframework.boot.actuate.health.Health; import org.springframework.boot.actuate.health.HealthIndicator; -import org.springframework.cloud.stream.binder.kafka.config.KafkaBinderConfigurationProperties; +import org.springframework.cloud.stream.binder.kafka.configuration.KafkaBinderConfigurationProperties; /** * Health indicator for Kafka. + * * @author Ilayaperumal Gopinathan * @author Marius Bogoevici */ @@ -43,7 +44,7 @@ public class KafkaBinderHealthIndicator implements HealthIndicator { private final KafkaBinderConfigurationProperties configurationProperties; public KafkaBinderHealthIndicator(KafkaMessageChannelBinder binder, - KafkaBinderConfigurationProperties configurationProperties) { + KafkaBinderConfigurationProperties configurationProperties) { this.binder = binder; this.configurationProperties = configurationProperties; 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 92ac41df1..548a07b7b 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 @@ -25,8 +25,6 @@ import java.util.Map; import java.util.Properties; import java.util.UUID; -import kafka.admin.AdminUtils; -import kafka.api.TopicMetadata; import kafka.common.ErrorMapping; import kafka.utils.ZkUtils; import org.apache.kafka.clients.consumer.ConsumerConfig; @@ -42,7 +40,6 @@ import org.apache.kafka.common.serialization.ByteArrayDeserializer; import org.apache.kafka.common.serialization.ByteArraySerializer; import org.apache.kafka.common.serialization.Deserializer; import org.apache.kafka.common.utils.Utils; -import scala.collection.Seq; import org.springframework.beans.factory.DisposableBean; import org.springframework.cloud.stream.binder.AbstractMessageChannelBinder; @@ -52,7 +49,8 @@ import org.springframework.cloud.stream.binder.BinderHeaders; import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; import org.springframework.cloud.stream.binder.ExtendedProducerProperties; import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder; -import org.springframework.cloud.stream.binder.kafka.config.KafkaBinderConfigurationProperties; +import org.springframework.cloud.stream.binder.kafka.admin.AdminUtilsOperation; +import org.springframework.cloud.stream.binder.kafka.configuration.KafkaBinderConfigurationProperties; import org.springframework.context.Lifecycle; import org.springframework.expression.common.LiteralExpression; import org.springframework.expression.spel.standard.SpelExpressionParser; @@ -83,6 +81,7 @@ import org.springframework.util.StringUtils; /** * A {@link Binder} that uses Kafka as the underlying middleware. + * * @author Eric Bottard * @author Marius Bogoevici * @author Ilayaperumal Gopinathan @@ -109,6 +108,8 @@ public class KafkaMessageChannelBinder extends private KafkaExtendedBindingProperties extendedBindingProperties = new KafkaExtendedBindingProperties(); + private AdminUtilsOperation adminUtilsOperation; + public KafkaMessageChannelBinder(KafkaBinderConfigurationProperties configurationProperties) { super(false, headersToMap(configurationProperties)); this.configurationProperties = configurationProperties; @@ -130,8 +131,13 @@ public class KafkaMessageChannelBinder extends return headersToMap; } + public void setAdminUtilsOperation(AdminUtilsOperation adminUtilsOperation) { + this.adminUtilsOperation = adminUtilsOperation; + } + /** * Retry configuration for operations such as validating topic creation + * * @param metadataRetryOperations the retry configuration */ public void setMetadataRetryOperations(RetryOperations metadataRetryOperations) { @@ -189,7 +195,7 @@ public class KafkaMessageChannelBinder extends @Override protected MessageHandler createProducerMessageHandler(final String destination, - ExtendedProducerProperties producerProperties) throws Exception { + ExtendedProducerProperties producerProperties) throws Exception { KafkaTopicUtils.validateTopicName(destination); @@ -215,11 +221,12 @@ public class KafkaMessageChannelBinder extends @Override protected String createProducerDestinationIfNecessary(String name, - ExtendedProducerProperties properties) { + ExtendedProducerProperties properties) { if (this.logger.isInfoEnabled()) { this.logger.info("Using kafka topic for outbound: " + name); } KafkaTopicUtils.validateTopicName(name); + Collection partitions = ensureTopicCreated(name, properties.getPartitionCount()); if (properties.getPartitionCount() < partitions.size()) { if (this.logger.isInfoEnabled()) { @@ -259,17 +266,19 @@ public class KafkaMessageChannelBinder extends @Override protected Collection createConsumerDestinationIfNecessary(String name, String group, - ExtendedConsumerProperties properties) { + ExtendedConsumerProperties properties) { KafkaTopicUtils.validateTopicName(name); if (properties.getInstanceCount() == 0) { throw new IllegalArgumentException("Instance count cannot be zero"); } + Collection allPartitions = ensureTopicCreated(name, properties.getInstanceCount() * properties.getConcurrency()); Collection listenedPartitions; - if (properties.getInstanceCount() == 1) { + if (properties.getExtension().isAutoRebalanceEnabled() || + properties.getInstanceCount() == 1) { listenedPartitions = allPartitions; } else { @@ -288,7 +297,7 @@ public class KafkaMessageChannelBinder extends @Override @SuppressWarnings("unchecked") protected MessageProducer createConsumerEndpoint(String name, String group, Collection destination, - ExtendedConsumerProperties properties) { + ExtendedConsumerProperties properties) { boolean anonymous = !StringUtils.hasText(group); Assert.isTrue(!anonymous || !properties.getExtension().isEnableDlq(), "DLQ support is not available for anonymous subscriptions"); @@ -429,38 +438,38 @@ public class KafkaMessageChannelBinder extends this.configurationProperties.getZkConnectionTimeout(), JaasUtils.isZkSecurityEnabled()); try { - final Properties topicConfig = new Properties(); - TopicMetadata topicMetadata = AdminUtils.fetchTopicMetadataFromZk(topicName, zkUtils); - if (topicMetadata.errorCode() == ErrorMapping.NoError()) { - // only consider minPartitionCount for resizing if autoAddPartitions is - // true + short errorCode = adminUtilsOperation.errorCodeFromTopicMetadata(topicName, zkUtils); + if (errorCode == ErrorMapping.NoError()) { + // only consider minPartitionCount for resizing if autoAddPartitions is true int effectivePartitionCount = this.configurationProperties.isAutoAddPartitions() ? Math.max(this.configurationProperties.getMinPartitionCount(), partitionCount) : partitionCount; - if (topicMetadata.partitionsMetadata().size() < effectivePartitionCount) { + int partitionSize = adminUtilsOperation.partitionSize(topicName, zkUtils); + + if (partitionSize < effectivePartitionCount) { if (this.configurationProperties.isAutoAddPartitions()) { - AdminUtils.addPartitions(zkUtils, topicName, effectivePartitionCount, null, false); + adminUtilsOperation.invokeAddPartitions(zkUtils, topicName, effectivePartitionCount, null, false); } else { - int topicSize = topicMetadata.partitionsMetadata().size(); throw new BinderException("The number of expected partitions was: " + partitionCount + ", but " - + topicSize + (topicSize > 1 ? " have " : " has ") + "been found instead." + + partitionSize + (partitionSize > 1 ? " have " : " has ") + "been found instead." + "Consider either increasing the partition count of the topic or enabling " + "`autoAddPartitions`"); } } } - else if (topicMetadata.errorCode() == ErrorMapping.UnknownTopicOrPartitionCode()) { + else if (errorCode == ErrorMapping.UnknownTopicOrPartitionCode()) { if (this.configurationProperties.isAutoCreateTopics()) { - Seq brokerList = zkUtils.getSortedBrokerList(); // always consider minPartitionCount for topic creation final int effectivePartitionCount = Math.max(this.configurationProperties.getMinPartitionCount(), partitionCount); + this.metadataRetryOperations.execute(new RetryCallback() { @Override public Object doWithRetry(RetryContext context) throws RuntimeException { - AdminUtils.createTopic(zkUtils, topicName, effectivePartitionCount, + + adminUtilsOperation.invokeCreateTopic(zkUtils, topicName, effectivePartitionCount, configurationProperties.getReplicationFactor(), new Properties()); return null; } @@ -472,7 +481,7 @@ public class KafkaMessageChannelBinder extends } else { throw new BinderException("Error fetching Kafka topic metadata: ", - ErrorMapping.exceptionFor(topicMetadata.errorCode())); + ErrorMapping.exceptionFor(errorCode)); } try { return this.metadataRetryOperations @@ -549,8 +558,8 @@ public class KafkaMessageChannelBinder extends private final DefaultKafkaProducerFactory producerFactory; private ProducerConfigurationMessageHandler(KafkaTemplate kafkaTemplate, String topic, - ExtendedProducerProperties producerProperties, - DefaultKafkaProducerFactory producerFactory) { + ExtendedProducerProperties producerProperties, + DefaultKafkaProducerFactory producerFactory) { super(kafkaTemplate); setTopicExpression(new LiteralExpression(topic)); setBeanFactory(KafkaMessageChannelBinder.this.getBeanFactory()); diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/admin/AdminUtilsOperation.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/admin/AdminUtilsOperation.java new file mode 100644 index 000000000..b36ce166e --- /dev/null +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/admin/AdminUtilsOperation.java @@ -0,0 +1,75 @@ +/* + * Copyright 2002-2016 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 + * + * http://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.admin; + +import java.util.Properties; + +import kafka.utils.ZkUtils; + +/** + * API around {@link kafka.admin.AdminUtils} to support + * various versions of Kafka brokers. + * + * Note: Implementations that support Kafka brokers other than 0.9, need to use + * a possible strategy that involves reflection around {@link kafka.admin.AdminUtils}. + * + * @author Soby Chacko + */ +public interface AdminUtilsOperation { + + /** + * Invoke {@link kafka.admin.AdminUtils#addPartitions} + * + * @param zkUtils Zookeeper utils + * @param topic name of the topic + * @param numPartitions + * @param replicaAssignmentStr + * @param checkBrokerAvailable + */ + void invokeAddPartitions(ZkUtils zkUtils, String topic, int numPartitions, + String replicaAssignmentStr, boolean checkBrokerAvailable); + + /** + * Invoke {@link kafka.admin.AdminUtils#fetchTopicMetadataFromZk} + * + * @param topic name + * @param zkUtils zookeeper utils + * @return error code + */ + short errorCodeFromTopicMetadata(String topic, ZkUtils zkUtils); + + /** + * Find partition size from Kafka broker using {@link kafka.admin.AdminUtils} + * + * @param topic name + * @param zkUtils zookeeper utils + * @return partition size + */ + int partitionSize(String topic, ZkUtils zkUtils); + + /** + * Inovke {@link kafka.admin.AdminUtils#createTopic} + * + * @param zkUtils zookeeper utils + * @param topic name + * @param partitions + * @param replicationFactor + * @param topicConfig + */ + void invokeCreateTopic(ZkUtils zkUtils, String topic, int partitions, + int replicationFactor, Properties topicConfig); +} diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/admin/Kafka09AdminUtilsOperation.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/admin/Kafka09AdminUtilsOperation.java new file mode 100644 index 000000000..bbec0f6b3 --- /dev/null +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/admin/Kafka09AdminUtilsOperation.java @@ -0,0 +1,51 @@ +/* + * Copyright 2002-2016 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 + * + * http://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.admin; + +import java.util.Properties; + +import kafka.admin.AdminUtils; +import kafka.api.TopicMetadata; +import kafka.utils.ZkUtils; + +/** + * @author Soby Chacko + */ +public class Kafka09AdminUtilsOperation implements AdminUtilsOperation { + + public void invokeAddPartitions(ZkUtils zkUtils, String topic, int numPartitions, + String replicaAssignmentStr, boolean checkBrokerAvailable) { + AdminUtils.addPartitions(zkUtils, topic, numPartitions, + replicaAssignmentStr, checkBrokerAvailable); + } + + public short errorCodeFromTopicMetadata(String topic, ZkUtils zkUtils) { + TopicMetadata topicMetadata = AdminUtils.fetchTopicMetadataFromZk(topic, zkUtils); + return topicMetadata.errorCode(); + } + + public int partitionSize(String topic, ZkUtils zkUtils) { + TopicMetadata topicMetadata = AdminUtils.fetchTopicMetadataFromZk(topic, zkUtils); + return topicMetadata.partitionsMetadata().size(); + } + + public void invokeCreateTopic(ZkUtils zkUtils, String topic, int partitions, + int replicationFactor, Properties topicConfig) { + AdminUtils.createTopic(zkUtils, topic, partitions, replicationFactor, + topicConfig); + } +} diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/admin/Kafka10AdminUtilsOperation.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/admin/Kafka10AdminUtilsOperation.java new file mode 100644 index 000000000..8dab2b3f3 --- /dev/null +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/admin/Kafka10AdminUtilsOperation.java @@ -0,0 +1,151 @@ +/* + * Copyright 2002-2016 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 + * + * http://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.admin; + +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; +import java.util.List; +import java.util.Properties; + +import kafka.api.PartitionMetadata; +import kafka.utils.ZkUtils; + +import org.springframework.util.ReflectionUtils; + +/** + * @author Soby Chacko + */ +public class Kafka10AdminUtilsOperation implements AdminUtilsOperation { + + private static ClassLoader CLASS_LOADER = Kafka10AdminUtilsOperation.class.getClassLoader(); + + private static Class ADMIN_UTIL_CLASS; + + static { + try { + ADMIN_UTIL_CLASS = CLASS_LOADER.loadClass("kafka.admin.AdminUtils"); + } + catch (ClassNotFoundException e) { + throw new IllegalStateException("AdminUtils class not found", e); + } + } + + public void invokeAddPartitions(ZkUtils zkUtils, String topic, int numPartitions, + String replicaAssignmentStr, boolean checkBrokerAvailable) { + try { + Method[] declaredMethods = ADMIN_UTIL_CLASS.getDeclaredMethods(); + Method addPartitions = null; + for (Method m : declaredMethods) { + if (m.getName().equals("addPartitions")) { + addPartitions = m; + } + } + + if (addPartitions != null) { + addPartitions.invoke(null, zkUtils, topic, numPartitions, + replicaAssignmentStr, checkBrokerAvailable, null); + } + else { + throw new InvocationTargetException( + new RuntimeException("method not found")); + } + } + catch (InvocationTargetException e) { + ReflectionUtils.handleInvocationTargetException(e); + } + catch (IllegalAccessException e) { + ReflectionUtils.handleReflectionException(e); + } + } + + public short errorCodeFromTopicMetadata(String topic, ZkUtils zkUtils) { + try { + Method fetchTopicMetadataFromZk = ReflectionUtils.findMethod(ADMIN_UTIL_CLASS, "fetchTopicMetadataFromZk", String.class, ZkUtils.class); + + Object result = fetchTopicMetadataFromZk.invoke(null, topic, zkUtils); + Class topicMetadataClass = CLASS_LOADER.loadClass("org.apache.kafka.common.requests.MetadataResponse$TopicMetadata"); + + Method errorCodeMethod = ReflectionUtils.findMethod(topicMetadataClass, "error"); + Object obj = errorCodeMethod.invoke(result); + Method code = ReflectionUtils.findMethod(obj.getClass(), "code"); + + return (short) code.invoke(obj); + } + catch (ClassNotFoundException e) { + throw new IllegalStateException("AdminUtils class not found", e); + } + catch (InvocationTargetException e) { + ReflectionUtils.handleInvocationTargetException(e); + } + catch (IllegalAccessException e) { + ReflectionUtils.handleReflectionException(e); + } + return 0; + + } + + @SuppressWarnings("unchecked") + public int partitionSize(String topic, ZkUtils zkUtils) { + try { + Method fetchTopicMetadataFromZk = ReflectionUtils.findMethod(ADMIN_UTIL_CLASS, "fetchTopicMetadataFromZk", String.class, ZkUtils.class); + Object result = fetchTopicMetadataFromZk.invoke(null, topic, zkUtils); + Class topicMetadataClass = CLASS_LOADER.loadClass("org.apache.kafka.common.requests.MetadataResponse$TopicMetadata"); + + Method partitionsMetadata = ReflectionUtils.findMethod(topicMetadataClass, "partitionMetadata"); + List foo = (List) partitionsMetadata.invoke(result); + return foo.size(); + } + catch (ClassNotFoundException e) { + throw new IllegalStateException("AdminUtils class not found", e); + } + catch (InvocationTargetException e) { + ReflectionUtils.handleInvocationTargetException(e); + } + catch (IllegalAccessException e) { + ReflectionUtils.handleReflectionException(e); + } + return 0; + } + + public void invokeCreateTopic(ZkUtils zkUtils, String topic, int partitions, + int replicationFactor, Properties topicConfig) { + try { + Method[] declaredMethods = ADMIN_UTIL_CLASS.getDeclaredMethods(); + Method createTopic = null; + for (Method m : declaredMethods) { + if (m.getName().equals("createTopic") && m.getParameterTypes()[m.getParameterTypes().length - 1].getName().endsWith("RackAwareMode")) { + createTopic = m; + break; + } + } + if (createTopic != null) { + createTopic.invoke(null, zkUtils, topic, partitions, + replicationFactor, topicConfig, null); + } + else { + throw new InvocationTargetException( + new RuntimeException("method not found")); + } + } + catch (InvocationTargetException e) { + ReflectionUtils.handleInvocationTargetException(e); + } + catch (IllegalAccessException e) { + ReflectionUtils.handleReflectionException(e); + } + } +} diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/configuration/KafkaBinderConfiguration.java similarity index 50% rename from spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java rename to spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/configuration/KafkaBinderConfiguration.java index 430210c27..58353bb23 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/configuration/KafkaBinderConfiguration.java @@ -14,7 +14,12 @@ * limitations under the License. */ -package org.springframework.cloud.stream.binder.kafka.config; +package org.springframework.cloud.stream.binder.kafka.configuration; + +import java.lang.reflect.Method; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.autoconfigure.PropertyPlaceholderAutoConfiguration; @@ -24,10 +29,18 @@ import org.springframework.cloud.stream.binder.Binder; import org.springframework.cloud.stream.binder.kafka.KafkaBinderHealthIndicator; import org.springframework.cloud.stream.binder.kafka.KafkaExtendedBindingProperties; import org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder; +import org.springframework.cloud.stream.binder.kafka.admin.AdminUtilsOperation; +import org.springframework.cloud.stream.binder.kafka.admin.Kafka09AdminUtilsOperation; +import org.springframework.cloud.stream.binder.kafka.admin.Kafka10AdminUtilsOperation; import org.springframework.cloud.stream.config.codec.kryo.KryoCodecAutoConfiguration; +import org.springframework.context.ApplicationContext; import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Condition; +import org.springframework.context.annotation.ConditionContext; +import org.springframework.context.annotation.Conditional; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Import; +import org.springframework.core.type.AnnotatedTypeMetadata; import org.springframework.integration.codec.Codec; import org.springframework.kafka.support.LoggingProducerListener; import org.springframework.kafka.support.ProducerListener; @@ -45,6 +58,8 @@ import org.springframework.kafka.support.ProducerListener; @EnableConfigurationProperties({KafkaBinderConfigurationProperties.class, KafkaExtendedBindingProperties.class}) public class KafkaBinderConfiguration { + protected final Log logger = LogFactory.getLog(getClass()); + @Autowired private Codec codec; @@ -57,6 +72,9 @@ public class KafkaBinderConfiguration { @Autowired private ProducerListener producerListener; + @Autowired + private ApplicationContext context; + @Bean KafkaMessageChannelBinder kafkaMessageChannelBinder() { KafkaMessageChannelBinder kafkaMessageChannelBinder = new KafkaMessageChannelBinder( @@ -64,6 +82,8 @@ public class KafkaBinderConfiguration { kafkaMessageChannelBinder.setCodec(this.codec); //kafkaMessageChannelBinder.setProducerListener(producerListener); kafkaMessageChannelBinder.setExtendedBindingProperties(this.kafkaExtendedBindingProperties); + AdminUtilsOperation adminUtilsOperation = context.getBean(AdminUtilsOperation.class); + kafkaMessageChannelBinder.setAdminUtilsOperation(adminUtilsOperation); return kafkaMessageChannelBinder; } @@ -77,4 +97,69 @@ public class KafkaBinderConfiguration { KafkaBinderHealthIndicator healthIndicator(KafkaMessageChannelBinder kafkaMessageChannelBinder) { return new KafkaBinderHealthIndicator(kafkaMessageChannelBinder, this.configurationProperties); } + + @Bean(name = "adminUtilsOperation") + @Conditional(Kafka09Condition.class) + public AdminUtilsOperation kafka09AdminUtilsOperation() { + logger.info("AdminUtils selected: Kafka 0.9 AdminUtils"); + return new Kafka09AdminUtilsOperation(); + } + + @Bean(name = "adminUtilsOperation") + @Conditional(Kafka10Condition.class) + public AdminUtilsOperation kafka10AdminUtilsOperation() { + logger.info("AdminUtils selected: Kafka 0.10 AdminUtils"); + return new Kafka10AdminUtilsOperation(); + } + + private static Method getMethod(ClassLoader classLoader, String methodName) { + try { + Class adminUtilClass = classLoader.loadClass("kafka.admin.AdminUtils"); + Method[] declaredMethods = adminUtilClass.getDeclaredMethods(); + for (Method m : declaredMethods) { + if (m.getName().equals(methodName)) { + return m; + } + } + } + catch (ClassNotFoundException e) { + throw new IllegalStateException("AdminUtils not found", e); + } + return null; + } + + static class Kafka10Condition implements Condition { + + @Override + public boolean matches(ConditionContext conditionContext, AnnotatedTypeMetadata annotatedTypeMetadata) { + ClassLoader classLoader = Kafka10Condition.class.getClassLoader(); + Method addPartitions = getMethod(classLoader, "addPartitions"); + if (addPartitions != null) { + Class[] parameterTypes = addPartitions.getParameterTypes(); + Class clazz = parameterTypes[parameterTypes.length - 1]; + if (clazz.getName().equals("kafka.admin.RackAwareMode")) { + return true; + } + } + return false; + } + } + + static class Kafka09Condition implements Condition { + + @Override + public boolean matches(ConditionContext conditionContext, AnnotatedTypeMetadata annotatedTypeMetadata) { + + ClassLoader classLoader = Kafka09Condition.class.getClassLoader(); + Method addPartitions = getMethod(classLoader, "addPartitions"); + if (addPartitions != null) { + Class[] parameterTypes = addPartitions.getParameterTypes(); + Class clazz = parameterTypes[parameterTypes.length - 1]; + if (!clazz.getName().equals("kafka.admin.RackAwareMode")) { + return true; + } + } + return false; + } + } } diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfigurationProperties.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/configuration/KafkaBinderConfigurationProperties.java similarity index 98% rename from spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfigurationProperties.java rename to spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/configuration/KafkaBinderConfigurationProperties.java index 6c9e1467f..35ae8cda6 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfigurationProperties.java +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/configuration/KafkaBinderConfigurationProperties.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.cloud.stream.binder.kafka.config; +package org.springframework.cloud.stream.binder.kafka.configuration; import java.util.HashMap; import java.util.Map; diff --git a/spring-cloud-stream-binder-kafka/src/main/resources/META-INF/spring.binders b/spring-cloud-stream-binder-kafka/src/main/resources/META-INF/spring.binders index 063a7400f..c6c1dc579 100644 --- a/spring-cloud-stream-binder-kafka/src/main/resources/META-INF/spring.binders +++ b/spring-cloud-stream-binder-kafka/src/main/resources/META-INF/spring.binders @@ -1,2 +1,2 @@ kafka:\ -org.springframework.cloud.stream.binder.kafka.config.KafkaBinderConfiguration +org.springframework.cloud.stream.binder.kafka.configuration.KafkaBinderConfiguration diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/AbstractKafkaTestBinder.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/AbstractKafkaTestBinder.java new file mode 100644 index 000000000..483ab9775 --- /dev/null +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/AbstractKafkaTestBinder.java @@ -0,0 +1,46 @@ +package org.springframework.cloud.stream.binder.kafka; + +import java.util.List; + +import com.esotericsoftware.kryo.Kryo; +import com.esotericsoftware.kryo.Registration; + +import org.springframework.cloud.stream.binder.AbstractTestBinder; +import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; +import org.springframework.cloud.stream.binder.ExtendedProducerProperties; +import org.springframework.integration.codec.Codec; +import org.springframework.integration.codec.kryo.KryoRegistrar; +import org.springframework.integration.codec.kryo.PojoCodec; +import org.springframework.integration.tuple.TupleKryoRegistrar; + +/** + * @author Soby Chacko + */ +public abstract class AbstractKafkaTestBinder extends + AbstractTestBinder, ExtendedProducerProperties> { + + @Override + public void cleanup() { + // do nothing - the rule will take care of that + } + + protected static Codec getCodec() { + return new PojoCodec(new TupleRegistrar()); + } + + private static class TupleRegistrar implements KryoRegistrar { + private final TupleKryoRegistrar delegate = new TupleKryoRegistrar(); + + @Override + public void registerTypes(Kryo kryo) { + this.delegate.registerTypes(kryo); + } + + @Override + public List getRegistrations() { + return this.delegate.getRegistrations(); + } + } + +} + diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka09BinderTests.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka09BinderTests.java new file mode 100644 index 000000000..7a0f6dabd --- /dev/null +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka09BinderTests.java @@ -0,0 +1,176 @@ +/* + * Copyright 2014-2016 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 + * + * http://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; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Properties; + +import kafka.utils.ZKStringSerializer$; +import kafka.utils.ZkUtils; +import org.I0Itec.zkclient.ZkClient; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.common.serialization.ByteArrayDeserializer; +import org.apache.kafka.common.serialization.Deserializer; +import org.junit.Before; +import org.junit.ClassRule; + +import org.springframework.cloud.stream.binder.Binder; +import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; +import org.springframework.cloud.stream.binder.ExtendedProducerProperties; +import org.springframework.cloud.stream.binder.Spy; +import org.springframework.cloud.stream.binder.kafka.admin.Kafka09AdminUtilsOperation; +import org.springframework.cloud.stream.binder.kafka.configuration.KafkaBinderConfigurationProperties; +import org.springframework.kafka.core.ConsumerFactory; +import org.springframework.kafka.core.DefaultKafkaConsumerFactory; +import org.springframework.kafka.support.KafkaHeaders; +import org.springframework.kafka.test.core.BrokerAddress; +import org.springframework.kafka.test.rule.KafkaEmbedded; +import org.springframework.retry.RetryOperations; + +/** + * Integration tests for the {@link KafkaMessageChannelBinder}. + * + * @author Eric Bottard + * @author Marius Bogoevici + * @author Mark Fisher + * @author Ilayaperumal Gopinathan + */ +public class Kafka09BinderTests extends KafkaBinderTests { + + private final String CLASS_UNDER_TEST_NAME = KafkaMessageChannelBinder.class.getSimpleName(); + + @ClassRule + public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, 10); + + private Kafka09TestBinder binder; + + private Kafka09AdminUtilsOperation adminUtilsOperation = new Kafka09AdminUtilsOperation(); + + @Override + protected void binderBindUnbindLatency() throws InterruptedException { + Thread.sleep(500); + } + + @Override + protected Kafka09TestBinder getBinder() { + if (binder == null) { + KafkaBinderConfigurationProperties binderConfiguration = createConfigurationProperties(); + binder = new Kafka09TestBinder(binderConfiguration); + } + return binder; + } + + protected KafkaBinderConfigurationProperties createConfigurationProperties() { + KafkaBinderConfigurationProperties binderConfiguration = new KafkaBinderConfigurationProperties(); + BrokerAddress[] brokerAddresses = embeddedKafka.getBrokerAddresses(); + List bAddresses = new ArrayList<>(); + for (BrokerAddress bAddress : brokerAddresses) { + bAddresses.add(bAddress.toString()); + } + String[] foo = new String[bAddresses.size()]; + binderConfiguration.setBrokers(bAddresses.toArray(foo)); + binderConfiguration.setZkNodes(embeddedKafka.getZookeeperConnectionString()); + return binderConfiguration; + } + + @Override + protected int partitionSize(String topic) { + return consumerFactory().createConsumer().partitionsFor(topic).size(); + } + + @Override + protected void setMetadataRetryOperations(Binder binder, RetryOperations retryOperations) { + ((Kafka09TestBinder) binder).getBinder().setMetadataRetryOperations(retryOperations); + } + + @Override + protected ZkUtils getZkUtils(KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties) { + final ZkClient zkClient = new ZkClient(kafkaBinderConfigurationProperties.getZkConnectionString(), + kafkaBinderConfigurationProperties.getZkSessionTimeout(), kafkaBinderConfigurationProperties.getZkConnectionTimeout(), + ZKStringSerializer$.MODULE$); + + return new ZkUtils(zkClient, null, false); + } + + @Override + protected void invokeCreateTopic(ZkUtils zkUtils, String topic, int partitions, int replicationFactor, Properties topicConfig) { + adminUtilsOperation.invokeCreateTopic(zkUtils, topic, partitions, replicationFactor, new Properties()); + } + + @Override + protected int invokePartitionSize(String topic, ZkUtils zkUtils) { + return adminUtilsOperation.partitionSize(topic, zkUtils); + } + + @Override + protected ExtendedConsumerProperties createConsumerProperties() { + return new ExtendedConsumerProperties<>(new KafkaConsumerProperties()); + } + + @Override + protected ExtendedProducerProperties createProducerProperties() { + return new ExtendedProducerProperties<>(new KafkaProducerProperties()); + } + + @Override + public String getKafkaOffsetHeaderKey() { + return KafkaHeaders.OFFSET; + } + + @Override + protected Binder getBinder(KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties) { + return new Kafka09TestBinder(kafkaBinderConfigurationProperties); + } + + @Before + public void init() { + String multiplier = System.getenv("KAFKA_TIMEOUT_MULTIPLIER"); + if (multiplier != null) { + timeoutMultiplier = Double.parseDouble(multiplier); + } + } + + @Override + protected boolean usesExplicitRouting() { + return false; + } + + @Override + protected String getClassUnderTestName() { + return CLASS_UNDER_TEST_NAME; + } + + @Override + public Spy spyOn(final String name) { + throw new UnsupportedOperationException("'spyOn' is not used by Kafka tests"); + } + + private ConsumerFactory consumerFactory() { + Map props = new HashMap<>(); + KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); + props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, configurationProperties.getKafkaConnectionString()); + props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); + Deserializer valueDecoder = new ByteArrayDeserializer(); + Deserializer keyDecoder = new ByteArrayDeserializer(); + + return new DefaultKafkaConsumerFactory<>(props, keyDecoder, valueDecoder); + } + +} diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka09TestBinder.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka09TestBinder.java new file mode 100644 index 000000000..c0ca54214 --- /dev/null +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka09TestBinder.java @@ -0,0 +1,54 @@ +/* + * Copyright 2015-2016 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 + * + * http://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; + +import org.springframework.cloud.stream.binder.kafka.admin.Kafka09AdminUtilsOperation; +import org.springframework.cloud.stream.binder.kafka.configuration.KafkaBinderConfigurationProperties; +import org.springframework.context.support.GenericApplicationContext; +import org.springframework.kafka.support.LoggingProducerListener; +import org.springframework.kafka.support.ProducerListener; + +/** + * Test support class for {@link KafkaMessageChannelBinder}. Creates a binder that uses + * an embedded Kafka cluster. + * @author Eric Bottard + * @author Marius Bogoevici + * @author David Turanski + * @author Gary Russell + * @author Soby Chacko + */ +public class Kafka09TestBinder extends AbstractKafkaTestBinder { + + public Kafka09TestBinder(KafkaBinderConfigurationProperties binderConfiguration) { + try { + KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder(binderConfiguration); + binder.setCodec(getCodec()); + ProducerListener producerListener = new LoggingProducerListener(); + binder.setProducerListener(producerListener); + GenericApplicationContext context = new GenericApplicationContext(); + context.refresh(); + binder.setApplicationContext(context); + binder.setAdminUtilsOperation(new Kafka09AdminUtilsOperation()); + binder.afterPropertiesSet(); + this.setBinder(binder); + } + catch (Exception e) { + throw new RuntimeException(e); + } + } + +} 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 31f54b7c4..599a155bc 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 @@ -1,5 +1,5 @@ /* - * Copyright 2014-2016 the original author or authors. + * Copyright 2016 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. @@ -16,30 +16,18 @@ package org.springframework.cloud.stream.binder.kafka; -import java.util.ArrayList; import java.util.Arrays; -import java.util.Collection; -import java.util.HashMap; import java.util.LinkedHashMap; import java.util.List; -import java.util.Map; import java.util.Properties; import java.util.UUID; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; -import kafka.admin.AdminUtils; -import kafka.api.TopicMetadata; import kafka.utils.ZKStringSerializer$; import kafka.utils.ZkUtils; import org.I0Itec.zkclient.ZkClient; -import org.apache.kafka.clients.consumer.ConsumerConfig; -import org.apache.kafka.common.PartitionInfo; -import org.apache.kafka.common.serialization.ByteArrayDeserializer; -import org.apache.kafka.common.serialization.Deserializer; import org.assertj.core.api.Condition; -import org.junit.Before; -import org.junit.ClassRule; import org.junit.Ignore; import org.junit.Test; @@ -51,27 +39,22 @@ import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; import org.springframework.cloud.stream.binder.ExtendedProducerProperties; import org.springframework.cloud.stream.binder.PartitionCapableBinderTests; import org.springframework.cloud.stream.binder.PartitionTestSupport; -import org.springframework.cloud.stream.binder.Spy; import org.springframework.cloud.stream.binder.TestUtils; -import org.springframework.cloud.stream.binder.kafka.config.KafkaBinderConfigurationProperties; +import org.springframework.cloud.stream.binder.kafka.configuration.KafkaBinderConfigurationProperties; import org.springframework.cloud.stream.config.BindingProperties; import org.springframework.context.support.GenericApplicationContext; import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler; -import org.springframework.kafka.core.ConsumerFactory; -import org.springframework.kafka.core.DefaultKafkaConsumerFactory; -import org.springframework.kafka.support.KafkaHeaders; import org.springframework.kafka.support.TopicPartitionInitialOffset; -import org.springframework.kafka.test.core.BrokerAddress; -import org.springframework.kafka.test.rule.KafkaEmbedded; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; import org.springframework.messaging.MessagingException; import org.springframework.messaging.support.GenericMessage; import org.springframework.messaging.support.MessageBuilder; +import org.springframework.retry.RetryOperations; import org.springframework.retry.backoff.FixedBackOffPolicy; import org.springframework.retry.policy.SimpleRetryPolicy; import org.springframework.retry.support.RetryTemplate; @@ -80,101 +63,37 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.junit.Assert.fail; /** - * Integration tests for the {@link KafkaMessageChannelBinder}. - * @author Eric Bottard - * @author Marius Bogoevici - * @author Mark Fisher - * @author Ilayaperumal Gopinathan + * @author Soby Chacko */ -public class KafkaBinderTests - extends - PartitionCapableBinderTests, - ExtendedProducerProperties> { +public abstract class KafkaBinderTests extends PartitionCapableBinderTests, + ExtendedProducerProperties> { - private final String CLASS_UNDER_TEST_NAME = KafkaMessageChannelBinder.class.getSimpleName(); + protected abstract ExtendedConsumerProperties createConsumerProperties(); - @ClassRule - public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, 10); + protected abstract ExtendedProducerProperties createProducerProperties(); - private KafkaTestBinder binder; + public abstract String getKafkaOffsetHeaderKey(); - @Override - protected void binderBindUnbindLatency() throws InterruptedException { - Thread.sleep(500); - } + protected abstract Binder getBinder(KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties); - @Override - protected KafkaTestBinder getBinder() { - if (binder == null) { - KafkaBinderConfigurationProperties binderConfiguration = createConfigurationProperties(); - binder = new KafkaTestBinder(binderConfiguration); - } - return binder; - } + protected abstract KafkaBinderConfigurationProperties createConfigurationProperties(); - protected KafkaBinderConfigurationProperties createConfigurationProperties() { - KafkaBinderConfigurationProperties binderConfiguration = new KafkaBinderConfigurationProperties(); - BrokerAddress[] brokerAddresses = embeddedKafka.getBrokerAddresses(); - List bAddresses = new ArrayList<>(); - for (BrokerAddress bAddress : brokerAddresses) { - bAddresses.add(bAddress.toString()); - } - String[] foo = new String[bAddresses.size()]; - binderConfiguration.setBrokers(bAddresses.toArray(foo)); - binderConfiguration.setZkNodes(embeddedKafka.getZookeeperConnectionString()); - return binderConfiguration; - } + protected abstract int partitionSize(String topic); - @Override - protected ExtendedConsumerProperties createConsumerProperties() { - return new ExtendedConsumerProperties<>(new KafkaConsumerProperties()); - } + protected abstract void setMetadataRetryOperations(Binder binder, RetryOperations retryOperations); - @Override - protected ExtendedProducerProperties createProducerProperties() { - return new ExtendedProducerProperties<>(new KafkaProducerProperties()); - } + protected abstract ZkUtils getZkUtils(KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties); - @Before - public void init() { - String multiplier = System.getenv("KAFKA_TIMEOUT_MULTIPLIER"); - if (multiplier != null) { - timeoutMultiplier = Double.parseDouble(multiplier); - } - } + protected abstract void invokeCreateTopic(ZkUtils zkUtils, String topic, int partitions, + int replicationFactor, Properties topicConfig); - @Override - protected boolean usesExplicitRouting() { - return false; - } - - @Override - protected String getClassUnderTestName() { - return CLASS_UNDER_TEST_NAME; - } - - @Override - public Spy spyOn(final String name) { - throw new UnsupportedOperationException("'spyOn' is not used by Kafka tests"); - } - - - private ConsumerFactory consumerFactory() { - Map props = new HashMap<>(); - KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); - props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, configurationProperties.getKafkaConnectionString()); - props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); - props.put(ConsumerConfig.GROUP_ID_CONFIG, "TEST-CONSUMER-GROUP"); - Deserializer valueDecoder = new ByteArrayDeserializer(); - Deserializer keyDecoder = new ByteArrayDeserializer(); - - return new DefaultKafkaConsumerFactory<>(props, keyDecoder, valueDecoder); - - } + protected abstract int invokePartitionSize(String topic, + ZkUtils zkUtils); @Test + @SuppressWarnings("unchecked") public void testDlqAndRetry() throws Exception { - KafkaTestBinder binder = getBinder(); + Binder binder = getBinder(); DirectChannel moduleOutputChannel = new DirectChannel(); DirectChannel moduleInputChannel = new DirectChannel(); QueueChannel dlqChannel = new QueueChannel(); @@ -216,8 +135,9 @@ public class KafkaBinderTests } @Test + @SuppressWarnings("unchecked") public void testDefaultAutoCommitOnErrorWithoutDlq() throws Exception { - KafkaTestBinder binder = getBinder(); + Binder binder = getBinder(); DirectChannel moduleOutputChannel = new DirectChannel(); DirectChannel moduleInputChannel = new DirectChannel(); FailingInvocationCountingMessageHandler handler = new FailingInvocationCountingMessageHandler(); @@ -266,8 +186,9 @@ public class KafkaBinderTests } @Test + @SuppressWarnings("unchecked") public void testDefaultAutoCommitOnErrorWithDlq() throws Exception { - KafkaTestBinder binder = getBinder(); + Binder binder = getBinder(); DirectChannel moduleOutputChannel = new DirectChannel(); DirectChannel moduleInputChannel = new DirectChannel(); FailingInvocationCountingMessageHandler handler = new FailingInvocationCountingMessageHandler(); @@ -325,19 +246,39 @@ public class KafkaBinderTests producerBinding.unbind(); } + @Test + @SuppressWarnings("unchecked") + public void testAutoCreateTopicsEnabledSucceeds() throws Exception { + KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); + configurationProperties.setAutoCreateTopics(true); + Binder binder = getBinder(configurationProperties); + RetryTemplate metatadataRetrievalRetryOperations = new RetryTemplate(); + metatadataRetrievalRetryOperations.setRetryPolicy(new SimpleRetryPolicy()); + FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy(); + backOffPolicy.setBackOffPeriod(1000); + metatadataRetrievalRetryOperations.setBackOffPolicy(backOffPolicy); + setMetadataRetryOperations(binder, metatadataRetrievalRetryOperations); + DirectChannel output = new DirectChannel(); + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + String testTopicName = "nonexisting" + System.currentTimeMillis(); + Binding binding = binder.bindConsumer(testTopicName, "test", output, consumerProperties); + binding.unbind(); + } + @Test(expected = IllegalArgumentException.class) public void testValidateKafkaTopicName() { KafkaTopicUtils.validateTopicName("foo:bar"); } @Test + @SuppressWarnings("unchecked") public void testCompression() throws Exception { - final KafkaProducerProperties.CompressionType[] codecs = new KafkaProducerProperties.CompressionType[] { + final KafkaProducerProperties.CompressionType[] codecs = new KafkaProducerProperties.CompressionType[]{ KafkaProducerProperties.CompressionType.none, KafkaProducerProperties.CompressionType.gzip, KafkaProducerProperties.CompressionType.snappy}; byte[] testPayload = new byte[2048]; Arrays.fill(testPayload, (byte) 65); - KafkaTestBinder binder = getBinder(); + Binder binder = getBinder(); for (KafkaProducerProperties.CompressionType codec : codecs) { DirectChannel moduleOutputChannel = new DirectChannel(); QueueChannel moduleInputChannel = new QueueChannel(); @@ -364,13 +305,14 @@ public class KafkaBinderTests } @Test + @SuppressWarnings("unchecked") public void testCustomPartitionCountOverridesDefaultIfLarger() throws Exception { byte[] testPayload = new byte[2048]; Arrays.fill(testPayload, (byte) 65); KafkaBinderConfigurationProperties binderConfiguration = createConfigurationProperties(); binderConfiguration.setMinPartitionCount(10); - KafkaTestBinder binder = new KafkaTestBinder(binderConfiguration); + Binder binder = getBinder(binderConfiguration); QueueChannel moduleInputChannel = new QueueChannel(); ExtendedProducerProperties producerProperties = createProducerProperties(); producerProperties.setPartitionCount(10); @@ -393,23 +335,20 @@ public class KafkaBinderTests assertThat(inbound).isNotNull(); assertThat((byte[]) inbound.getPayload()).containsExactly(testPayload); - - Collection partitions = - consumerFactory().createConsumer().partitionsFor("foo" + uniqueBindingId + ".0"); - - assertThat(partitions).hasSize(10); + assertThat(partitionSize("foo" + uniqueBindingId + ".0")).isEqualTo(10); producerBinding.unbind(); consumerBinding.unbind(); } @Test + @SuppressWarnings("unchecked") public void testCustomPartitionCountDoesNotOverridePartitioningIfSmaller() throws Exception { byte[] testPayload = new byte[2048]; Arrays.fill(testPayload, (byte) 65); KafkaBinderConfigurationProperties binderConfiguration = createConfigurationProperties(); binderConfiguration.setMinPartitionCount(6); - KafkaTestBinder binder = new KafkaTestBinder(binderConfiguration); + Binder binder = getBinder(binderConfiguration); QueueChannel moduleInputChannel = new QueueChannel(); ExtendedProducerProperties producerProperties = createProducerProperties(); producerProperties.setPartitionCount(5); @@ -431,22 +370,21 @@ public class KafkaBinderTests Message inbound = receive(moduleInputChannel); assertThat(inbound).isNotNull(); assertThat((byte[]) inbound.getPayload()).containsExactly(testPayload); - Collection partitions = - consumerFactory().createConsumer().partitionsFor("foo" + uniqueBindingId + ".0"); - assertThat(partitions).hasSize(6); + assertThat(partitionSize("foo" + uniqueBindingId + ".0")).isEqualTo(6); producerBinding.unbind(); consumerBinding.unbind(); } @Test + @SuppressWarnings("unchecked") public void testCustomPartitionCountOverridesPartitioningIfLarger() throws Exception { byte[] testPayload = new byte[2048]; Arrays.fill(testPayload, (byte) 65); KafkaBinderConfigurationProperties binderConfiguration = createConfigurationProperties(); binderConfiguration.setMinPartitionCount(4); - KafkaTestBinder binder = new KafkaTestBinder(binderConfiguration); + Binder binder = getBinder(binderConfiguration); QueueChannel moduleInputChannel = new QueueChannel(); ExtendedProducerProperties producerProperties = createProducerProperties(); @@ -468,9 +406,7 @@ public class KafkaBinderTests Message inbound = receive(moduleInputChannel); assertThat(inbound).isNotNull(); assertThat((byte[]) inbound.getPayload()).containsExactly(testPayload); - Collection partitions = - consumerFactory().createConsumer().partitionsFor("foo" + uniqueBindingId + ".0"); - assertThat(partitions).hasSize(5); + assertThat(partitionSize("foo" + uniqueBindingId + ".0")).isEqualTo(5); producerBinding.unbind(); consumerBinding.unbind(); } @@ -478,11 +414,11 @@ public class KafkaBinderTests @Test @SuppressWarnings("unchecked") public void testDefaultConsumerStartsAtEarliest() throws Exception { - KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder(createConfigurationProperties()); + Binder binder = getBinder(createConfigurationProperties()); GenericApplicationContext context = new GenericApplicationContext(); context.refresh(); - binder.setApplicationContext(context); - binder.afterPropertiesSet(); + //binder.setApplicationContext(context); + //binder.afterPropertiesSet(); DirectChannel output = new DirectChannel(); QueueChannel input1 = new QueueChannel(); @@ -515,7 +451,7 @@ public class KafkaBinderTests Binding consumerBinding = null; try { - KafkaTestBinder binder = getBinder(); + Binder binder = getBinder(); DirectChannel output = new DirectChannel(); QueueChannel input1 = new QueueChannel(); @@ -552,7 +488,7 @@ public class KafkaBinderTests @Ignore("Needs further discussion") @SuppressWarnings("unchecked") public void testReset() throws Exception { - KafkaTestBinder binder = getBinder(); + Binder binder = getBinder(); DirectChannel output = new DirectChannel(); QueueChannel input1 = new QueueChannel(); @@ -605,11 +541,7 @@ public class KafkaBinderTests try { KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); - KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder(configurationProperties); - GenericApplicationContext context = new GenericApplicationContext(); - context.refresh(); - binder.setApplicationContext(context); - binder.afterPropertiesSet(); + Binder binder = getBinder(configurationProperties); DirectChannel output = new DirectChannel(); QueueChannel input1 = new QueueChannel(); @@ -653,251 +585,6 @@ public class KafkaBinderTests } } - @Test - public void testSyncProducerMetadata() throws Exception { - KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder(createConfigurationProperties()); - GenericApplicationContext context = new GenericApplicationContext(); - context.refresh(); - binder.setApplicationContext(context); - binder.afterPropertiesSet(); - DirectChannel output = new DirectChannel(); - String testTopicName = UUID.randomUUID().toString(); - ExtendedProducerProperties properties = createProducerProperties(); - properties.getExtension().setSync(true); - Binding producerBinding = binder.bindProducer(testTopicName, output, properties); - DirectFieldAccessor accessor = new DirectFieldAccessor(extractEndpoint(producerBinding)); - KafkaProducerMessageHandler wrappedInstance = (KafkaProducerMessageHandler) accessor.getWrappedInstance(); - assertThat(new DirectFieldAccessor(wrappedInstance).getPropertyValue("sync").equals(Boolean.TRUE)) - .withFailMessage("Kafka Sync Producer should have been enabled."); - producerBinding.unbind(); - } - - @Test - public void testAutoCreateTopicsDisabledFailsIfTopicMissing() throws Exception { - KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); - configurationProperties.setAutoCreateTopics(false); - KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder(configurationProperties); - GenericApplicationContext context = new GenericApplicationContext(); - context.refresh(); - binder.setApplicationContext(context); - binder.afterPropertiesSet(); - RetryTemplate metatadataRetrievalRetryOperations = new RetryTemplate(); - metatadataRetrievalRetryOperations.setRetryPolicy(new SimpleRetryPolicy()); - FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy(); - backOffPolicy.setBackOffPeriod(1000); - metatadataRetrievalRetryOperations.setBackOffPolicy(backOffPolicy); - binder.setMetadataRetryOperations(metatadataRetrievalRetryOperations); - DirectChannel output = new DirectChannel(); - ExtendedConsumerProperties consumerProperties = createConsumerProperties(); - String testTopicName = "nonexisting" + System.currentTimeMillis(); - try { - binder.doBindConsumer(testTopicName, "test", output, consumerProperties); - fail(); - } - catch (Exception e) { - assertThat(e).isInstanceOf(BinderException.class); - assertThat(e).hasMessageContaining("Topic " + testTopicName + " does not exist"); - } - } - - @Test - public void testAutoConfigureTopicsDisabledSucceedsIfTopicExisting() throws Exception { - KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); - - final ZkClient zkClient = new ZkClient(configurationProperties.getZkConnectionString(), - configurationProperties.getZkSessionTimeout(), configurationProperties.getZkConnectionTimeout(), - ZKStringSerializer$.MODULE$); - - final ZkUtils zkUtils = new ZkUtils(zkClient, null, false); - - String testTopicName = "existing" + System.currentTimeMillis(); - AdminUtils.createTopic(zkUtils, testTopicName, 5, 1, new Properties()); - - configurationProperties.setAutoCreateTopics(false); - KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder(configurationProperties); - GenericApplicationContext context = new GenericApplicationContext(); - context.refresh(); - binder.setApplicationContext(context); - binder.afterPropertiesSet(); - DirectChannel output = new DirectChannel(); - ExtendedConsumerProperties consumerProperties = createConsumerProperties(); - Binding binding = binder.doBindConsumer(testTopicName, "test", output, consumerProperties); - binding.unbind(); - } - - @Test - public void testAutoAddPartitionsDisabledFailsIfTopicUnderpartitioned() throws Exception { - KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); - - final ZkClient zkClient = new ZkClient(configurationProperties.getZkConnectionString(), - configurationProperties.getZkSessionTimeout(), configurationProperties.getZkConnectionTimeout(), - ZKStringSerializer$.MODULE$); - - final ZkUtils zkUtils = new ZkUtils(zkClient, null, false); - - String testTopicName = "existing" + System.currentTimeMillis(); - AdminUtils.createTopic(zkUtils, testTopicName, 1, 1, new Properties()); - configurationProperties.setAutoAddPartitions(false); - KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder(configurationProperties); - GenericApplicationContext context = new GenericApplicationContext(); - context.refresh(); - binder.setApplicationContext(context); - binder.afterPropertiesSet(); - DirectChannel output = new DirectChannel(); - ExtendedConsumerProperties consumerProperties = createConsumerProperties(); - // this consumer must consume from partition 2 - consumerProperties.setInstanceCount(3); - consumerProperties.setInstanceIndex(2); - try { - binder.doBindConsumer(testTopicName, "test", output, consumerProperties); - } - catch (Exception e) { - assertThat(e).isInstanceOf(BinderException.class); - assertThat(e) - .hasMessageContaining("The number of expected partitions was: 3, but 1 has been found instead"); - } - } - - @Test - public void testAutoAddPartitionsDisabledSucceedsIfTopicPartitionedCorrectly() throws Exception { - Binding binding = null; - try { - KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); - - final ZkClient zkClient = new ZkClient(configurationProperties.getZkConnectionString(), - configurationProperties.getZkSessionTimeout(), configurationProperties.getZkConnectionTimeout(), - ZKStringSerializer$.MODULE$); - - final ZkUtils zkUtils = new ZkUtils(zkClient, null, false); - - String testTopicName = "existing" + System.currentTimeMillis(); - AdminUtils.createTopic(zkUtils, testTopicName, 6, 1, new Properties()); - configurationProperties.setAutoAddPartitions(false); - KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder(configurationProperties); - GenericApplicationContext context = new GenericApplicationContext(); - RetryTemplate metatadataRetrievalRetryOperations = new RetryTemplate(); - metatadataRetrievalRetryOperations.setRetryPolicy(new SimpleRetryPolicy()); - FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy(); - backOffPolicy.setBackOffPeriod(1000); - metatadataRetrievalRetryOperations.setBackOffPolicy(backOffPolicy); - binder.setMetadataRetryOperations(metatadataRetrievalRetryOperations); - context.refresh(); - binder.setApplicationContext(context); - binder.afterPropertiesSet(); - DirectChannel output = new DirectChannel(); - ExtendedConsumerProperties consumerProperties = createConsumerProperties(); - // this consumer must consume from partition 2 - consumerProperties.setInstanceCount(3); - consumerProperties.setInstanceIndex(2); - consumerProperties.getExtension().setAutoRebalanceEnabled(false); - - binding = binder.doBindConsumer(testTopicName, "test-x", output, consumerProperties); - - TopicPartitionInitialOffset[] listenedPartitions = TestUtils.getPropertyValue(binding, - "endpoint.messageListenerContainer.containerProperties.topicPartitions", - TopicPartitionInitialOffset[].class); - assertThat(listenedPartitions).hasSize(2); - assertThat(listenedPartitions).contains(new TopicPartitionInitialOffset(testTopicName, 2), - new TopicPartitionInitialOffset(testTopicName, 5)); - Collection partitions = - consumerFactory().createConsumer().partitionsFor(testTopicName); - assertThat(partitions).hasSize(6); - } - finally { - binding.unbind(); - } - } - - @Test - public void testAutoCreateTopicsEnabledSucceeds() throws Exception { - KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); - configurationProperties.setAutoCreateTopics(true); - KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder(configurationProperties); - GenericApplicationContext context = new GenericApplicationContext(); - context.refresh(); - binder.setApplicationContext(context); - binder.afterPropertiesSet(); - RetryTemplate metatadataRetrievalRetryOperations = new RetryTemplate(); - metatadataRetrievalRetryOperations.setRetryPolicy(new SimpleRetryPolicy()); - FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy(); - backOffPolicy.setBackOffPeriod(1000); - metatadataRetrievalRetryOperations.setBackOffPolicy(backOffPolicy); - binder.setMetadataRetryOperations(metatadataRetrievalRetryOperations); - DirectChannel output = new DirectChannel(); - ExtendedConsumerProperties consumerProperties = createConsumerProperties(); - String testTopicName = "nonexisting" + System.currentTimeMillis(); - Binding binding = binder.doBindConsumer(testTopicName, "test", output, consumerProperties); - binding.unbind(); - } - - @Test - public void testPartitionCountNotReduced() throws Exception { - String testTopicName = "existing" + System.currentTimeMillis(); - - KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); - - final ZkClient zkClient = new ZkClient(configurationProperties.getZkConnectionString(), - configurationProperties.getZkSessionTimeout(), configurationProperties.getZkConnectionTimeout(), - ZKStringSerializer$.MODULE$); - - final ZkUtils zkUtils = new ZkUtils(zkClient, null, false); - - AdminUtils.createTopic(zkUtils, testTopicName, 6, 1, new Properties()); - configurationProperties.setAutoAddPartitions(true); - KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder(configurationProperties); - GenericApplicationContext context = new GenericApplicationContext(); - context.refresh(); - binder.setApplicationContext(context); - binder.afterPropertiesSet(); - RetryTemplate metatadataRetrievalRetryOperations = new RetryTemplate(); - metatadataRetrievalRetryOperations.setRetryPolicy(new SimpleRetryPolicy()); - FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy(); - backOffPolicy.setBackOffPeriod(1000); - metatadataRetrievalRetryOperations.setBackOffPolicy(backOffPolicy); - binder.setMetadataRetryOperations(metatadataRetrievalRetryOperations); - DirectChannel output = new DirectChannel(); - ExtendedConsumerProperties consumerProperties = createConsumerProperties(); - Binding binding = binder.doBindConsumer(testTopicName, "test", output, consumerProperties); - binding.unbind(); - TopicMetadata topicMetadata = AdminUtils.fetchTopicMetadataFromZk(testTopicName, - zkUtils); - assertThat(topicMetadata.partitionsMetadata().size()).isEqualTo(6); - } - - @Test - public void testPartitionCountIncreasedIfAutoAddPartitionsSet() throws Exception { - KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); - - final ZkClient zkClient = new ZkClient(configurationProperties.getZkConnectionString(), - configurationProperties.getZkSessionTimeout(), configurationProperties.getZkConnectionTimeout(), - ZKStringSerializer$.MODULE$); - - final ZkUtils zkUtils = new ZkUtils(zkClient, null, false); - - String testTopicName = "existing" + System.currentTimeMillis(); - AdminUtils.createTopic(zkUtils, testTopicName, 1, 1, new Properties()); - configurationProperties.setMinPartitionCount(6); - configurationProperties.setAutoAddPartitions(true); - KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder(configurationProperties); - GenericApplicationContext context = new GenericApplicationContext(); - context.refresh(); - binder.setApplicationContext(context); - binder.afterPropertiesSet(); - RetryTemplate metatadataRetrievalRetryOperations = new RetryTemplate(); - metatadataRetrievalRetryOperations.setRetryPolicy(new SimpleRetryPolicy()); - FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy(); - backOffPolicy.setBackOffPeriod(1000); - metatadataRetrievalRetryOperations.setBackOffPolicy(backOffPolicy); - binder.setMetadataRetryOperations(metatadataRetrievalRetryOperations); - DirectChannel output = new DirectChannel(); - ExtendedConsumerProperties consumerProperties = createConsumerProperties(); - Binding binding = binder.doBindConsumer(testTopicName, "test", output, consumerProperties); - binding.unbind(); - TopicMetadata topicMetadata = AdminUtils.fetchTopicMetadataFromZk(testTopicName, - zkUtils); - assertThat(topicMetadata.partitionsMetadata().size()).isEqualTo(6); - } - @Test @Override @SuppressWarnings("unchecked") @@ -1156,7 +843,6 @@ public class KafkaBinderTests QueueChannel input1 = new QueueChannel(); ExtendedConsumerProperties consumerProperties = createConsumerProperties(); - //consumerProperties.getExtension().setAutoRebalanceEnabled(false); Binding binding1 = binder.bindConsumer("defaultGroup.0", null, input1, consumerProperties); @@ -1203,7 +889,216 @@ public class KafkaBinderTests binding2.unbind(); } - private static final class FailingInvocationCountingMessageHandler implements MessageHandler { + @Test + @SuppressWarnings("unchecked") + public void testSyncProducerMetadata() throws Exception { + Binder binder = getBinder(createConfigurationProperties()); + DirectChannel output = new DirectChannel(); + String testTopicName = UUID.randomUUID().toString(); + ExtendedProducerProperties properties = createProducerProperties(); + properties.getExtension().setSync(true); + Binding producerBinding = binder.bindProducer(testTopicName, output, properties); + DirectFieldAccessor accessor = new DirectFieldAccessor(extractEndpoint(producerBinding)); + KafkaProducerMessageHandler wrappedInstance = (KafkaProducerMessageHandler) accessor.getWrappedInstance(); + assertThat(new DirectFieldAccessor(wrappedInstance).getPropertyValue("sync").equals(Boolean.TRUE)) + .withFailMessage("Kafka Sync Producer should have been enabled."); + producerBinding.unbind(); + } + + @Test + @SuppressWarnings("unchecked") + public void testAutoCreateTopicsDisabledFailsIfTopicMissing() throws Exception { + KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); + configurationProperties.setAutoCreateTopics(false); + Binder binder = getBinder(configurationProperties); + RetryTemplate metatadataRetrievalRetryOperations = new RetryTemplate(); + metatadataRetrievalRetryOperations.setRetryPolicy(new SimpleRetryPolicy()); + FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy(); + backOffPolicy.setBackOffPeriod(1000); + metatadataRetrievalRetryOperations.setBackOffPolicy(backOffPolicy); + setMetadataRetryOperations(binder, metatadataRetrievalRetryOperations); + DirectChannel output = new DirectChannel(); + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + String testTopicName = "nonexisting" + System.currentTimeMillis(); + try { + binder.bindConsumer(testTopicName, "test", output, consumerProperties); + fail(); + } + catch (Exception e) { + assertThat(e).isInstanceOf(BinderException.class); + assertThat(e).hasMessageContaining("Topic " + testTopicName + " does not exist"); + } + } + + @Test + @SuppressWarnings("unchecked") + public void testAutoConfigureTopicsDisabledSucceedsIfTopicExisting() throws Exception { + KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); + + final ZkUtils zkUtils = getZkUtils(configurationProperties); + + String testTopicName = "existing" + System.currentTimeMillis(); + invokeCreateTopic(zkUtils, testTopicName, 5, 1, new Properties()); + + configurationProperties.setAutoCreateTopics(false); + Binder binder = getBinder(configurationProperties); + DirectChannel output = new DirectChannel(); + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + Binding binding = binder.bindConsumer(testTopicName, "test", output, consumerProperties); + binding.unbind(); + } + + @Test + @SuppressWarnings("unchecked") + public void testPartitionCountIncreasedIfAutoAddPartitionsSet() throws Exception { + KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); + + final ZkUtils zkUtils = getZkUtils(configurationProperties); + + String testTopicName = "existing" + System.currentTimeMillis(); + invokeCreateTopic(zkUtils, testTopicName, 6, 1, new Properties()); + configurationProperties.setMinPartitionCount(6); + configurationProperties.setAutoAddPartitions(true); + Binder binder = getBinder(configurationProperties); + RetryTemplate metatadataRetrievalRetryOperations = new RetryTemplate(); + metatadataRetrievalRetryOperations.setRetryPolicy(new SimpleRetryPolicy()); + FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy(); + backOffPolicy.setBackOffPeriod(1000); + metatadataRetrievalRetryOperations.setBackOffPolicy(backOffPolicy); + setMetadataRetryOperations(binder, metatadataRetrievalRetryOperations); + DirectChannel output = new DirectChannel(); + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + Binding binding = binder.bindConsumer(testTopicName, "test", output, consumerProperties); + binding.unbind(); + assertThat(invokePartitionSize(testTopicName, zkUtils)).isEqualTo(6); + } + + @Test + @SuppressWarnings("unchecked") + public void testAutoAddPartitionsDisabledFailsIfTopicUnderpartitioned() throws Exception { + KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); + + final ZkClient zkClient = new ZkClient(configurationProperties.getZkConnectionString(), + configurationProperties.getZkSessionTimeout(), configurationProperties.getZkConnectionTimeout(), + ZKStringSerializer$.MODULE$); + + final ZkUtils zkUtils = new ZkUtils(zkClient, null, false); + + String testTopicName = "existing" + System.currentTimeMillis(); + invokeCreateTopic(zkUtils, testTopicName, 1, 1, new Properties()); + configurationProperties.setAutoAddPartitions(false); + Binder binder = getBinder(configurationProperties); + GenericApplicationContext context = new GenericApplicationContext(); + context.refresh(); + // binder.setApplicationContext(context); + // binder.afterPropertiesSet(); + DirectChannel output = new DirectChannel(); + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + // this consumer must consume from partition 2 + consumerProperties.setInstanceCount(3); + consumerProperties.setInstanceIndex(2); + Binding binding = null; + try { + binding = binder.bindConsumer(testTopicName, "test", output, consumerProperties); + } + catch (Exception e) { + assertThat(e).isInstanceOf(BinderException.class); + assertThat(e) + .hasMessageContaining("The number of expected partitions was: 3, but 1 has been found instead"); + } + finally { + if (binding != null) { + binding.unbind(); + } + } + } + + @Test + @SuppressWarnings("unchecked") + public void testAutoAddPartitionsDisabledSucceedsIfTopicPartitionedCorrectly() throws Exception { + Binding binding = null; + try { + KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); + + final ZkClient zkClient = new ZkClient(configurationProperties.getZkConnectionString(), + configurationProperties.getZkSessionTimeout(), configurationProperties.getZkConnectionTimeout(), + ZKStringSerializer$.MODULE$); + + final ZkUtils zkUtils = new ZkUtils(zkClient, null, false); + + String testTopicName = "existing" + System.currentTimeMillis(); + invokeCreateTopic(zkUtils, testTopicName, 6, 1, new Properties()); + configurationProperties.setAutoAddPartitions(false); + Binder binder = getBinder(configurationProperties); + RetryTemplate metatadataRetrievalRetryOperations = new RetryTemplate(); + metatadataRetrievalRetryOperations.setRetryPolicy(new SimpleRetryPolicy()); + FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy(); + backOffPolicy.setBackOffPeriod(1000); + metatadataRetrievalRetryOperations.setBackOffPolicy(backOffPolicy); + setMetadataRetryOperations(binder, metatadataRetrievalRetryOperations); + DirectChannel output = new DirectChannel(); + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + // this consumer must consume from partition 2 + consumerProperties.setInstanceCount(3); + consumerProperties.setInstanceIndex(2); + consumerProperties.getExtension().setAutoRebalanceEnabled(false); + + binding = binder.bindConsumer(testTopicName, "test-x", output, consumerProperties); + + TopicPartitionInitialOffset[] listenedPartitions = TestUtils.getPropertyValue(binding, + "endpoint.messageListenerContainer.containerProperties.topicPartitions", + TopicPartitionInitialOffset[].class); + assertThat(listenedPartitions).hasSize(2); + assertThat(listenedPartitions).contains(new TopicPartitionInitialOffset(testTopicName, 2), + new TopicPartitionInitialOffset(testTopicName, 5)); + int partitions = invokePartitionSize(testTopicName, zkUtils); + assertThat(partitions).isEqualTo(6); + } + finally { + if (binding != null) { + binding.unbind(); + } + } + } + + @Test + @SuppressWarnings("unchecked") + public void testPartitionCountNotReduced() throws Exception { + String testTopicName = "existing" + System.currentTimeMillis(); + + KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); + + final ZkClient zkClient; + zkClient = new ZkClient(configurationProperties.getZkConnectionString(), + configurationProperties.getZkSessionTimeout(), configurationProperties.getZkConnectionTimeout(), + ZKStringSerializer$.MODULE$); + + final ZkUtils zkUtils = new ZkUtils(zkClient, null, false); + invokeCreateTopic(zkUtils, testTopicName, 6, 1, new Properties()); + configurationProperties.setAutoAddPartitions(true); + Binder binder = getBinder(configurationProperties); + GenericApplicationContext context = new GenericApplicationContext(); + context.refresh(); + RetryTemplate metatadataRetrievalRetryOperations = new RetryTemplate(); + metatadataRetrievalRetryOperations.setRetryPolicy(new SimpleRetryPolicy()); + FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy(); + backOffPolicy.setBackOffPeriod(1000); + metatadataRetrievalRetryOperations.setBackOffPolicy(backOffPolicy); + setMetadataRetryOperations(binder, metatadataRetrievalRetryOperations); + DirectChannel output = new DirectChannel(); + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + Binding binding = binder.bindConsumer(testTopicName, "test", output, consumerProperties); + binding.unbind(); + + assertThat(partitionSize(testTopicName)).isEqualTo(6); + } + + @Override + protected void binderBindUnbindLatency() throws InterruptedException { + Thread.sleep(500); + } + + private final class FailingInvocationCountingMessageHandler implements MessageHandler { private int invocationCount; @@ -1222,7 +1117,7 @@ public class KafkaBinderTests @Override public void handleMessage(Message message) throws MessagingException { invocationCount++; - Long offset = message.getHeaders().get(KafkaHeaders.OFFSET, Long.class); + Long offset = message.getHeaders().get(KafkaBinderTests.this.getKafkaOffsetHeaderKey(), Long.class); // using the offset as key allows to ensure that we don't store duplicate // messages on retry if (!receivedMessages.containsKey(offset)) { diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/RawModeKafkaBinderTests.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/RawModeKafka09BinderTests.java similarity index 97% rename from spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/RawModeKafkaBinderTests.java rename to spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/RawModeKafka09BinderTests.java index 83e0eee69..d61be6c27 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/RawModeKafkaBinderTests.java +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/RawModeKafka09BinderTests.java @@ -40,12 +40,12 @@ import static org.assertj.core.api.Assertions.assertThat; * @author Gary Russell * @author Mark Fisher */ -public class RawModeKafkaBinderTests extends KafkaBinderTests { +public class RawModeKafka09BinderTests extends Kafka09BinderTests { @Test @Override public void testPartitionedModuleJava() throws Exception { - KafkaTestBinder binder = getBinder(); + Kafka09TestBinder binder = getBinder(); ExtendedProducerProperties properties = createProducerProperties(); properties.setHeaderMode(HeaderMode.raw); properties.setPartitionKeyExtractorClass(RawKafkaPartitionTestSupport.class); @@ -98,7 +98,7 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests { @Test @Override public void testPartitionedModuleSpEL() throws Exception { - KafkaTestBinder binder = getBinder(); + Kafka09TestBinder binder = getBinder(); ExtendedProducerProperties properties = createProducerProperties(); properties.setPartitionKeyExpression(spelExpressionParser.parseExpression("payload[0]")); properties.setPartitionSelectorExpression(spelExpressionParser.parseExpression("hashCode()")); @@ -160,7 +160,7 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests { @Test @Override public void testSendAndReceive() throws Exception { - KafkaTestBinder binder = getBinder(); + Kafka09TestBinder binder = getBinder(); DirectChannel moduleOutputChannel = new DirectChannel(); QueueChannel moduleInputChannel = new QueueChannel(); ExtendedProducerProperties producerProperties = createProducerProperties(); @@ -184,7 +184,7 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests { @Test public void testSendAndReceiveWithExplicitConsumerGroup() { - KafkaTestBinder binder = getBinder(); + Kafka09TestBinder binder = getBinder(); DirectChannel moduleOutputChannel = new DirectChannel(); // Test pub/sub by emulating how StreamPlugin handles taps QueueChannel module1InputChannel = new QueueChannel();