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 e1a803bd3..34386658f 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 @@ -271,3 +271,32 @@ Then add these dependencies at the top of the `` section in the po The versions above are provided only for the sake of the example. For best results, we recommend using the most recent 0.10-compatible versions of the projects. ==== + +==== Excluding Kafka broker jar from the classpath of the binder based application +If the inclusion of the Kafka broker is not desired at runtime, Kafka binder allows this particular dependency to be excluded from the binder so that it will not be part of the classpath. +However, topic creation will be disabled from the binder if this dependency is excluded. + +If you use Kafka 10 dependencies as advised above, all you have to do is not to include the kafka broker dependency. +If you use Kafka 0.9, then ensure that you exclude the kafka broker jar from the `spring-cloud-starter-stream-kafka` dependency as following. + +[source,xml] +---- + + org.springframework.cloud + spring-cloud-starter-stream-kafka + + + org.apache.kafka + kafka_2.11 + + + +---- + +If you exclude the kafka broker dependency and the topic is not present on the server, then Kafka broker will create the topic if auto topic creation is enabled on the server. +Please keep in mind that if you are relying on this, then the Kafka server will use the default number of partitions and replication factors. +On the other hand, if auto topic creation is disabled on the server, then care must be taken before running the application to create the topic with the desired number of partitions. + +If you want to have full control over how partitions are allocated, then leave the default settings as they are, i.e. do not exclude the kafka broker jar and ensure that `spring.cloud.stream.kafka.binder.autoCreateTopics` is set to `true` which is the default. + + 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 c1969d366..7fc2b864c 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 @@ -198,8 +198,8 @@ public class KafkaMessageChannelBinder extends ExtendedProducerProperties producerProperties) throws Exception { KafkaTopicUtils.validateTopicName(destination); - - Collection partitions = ensureTopicCreated(destination, producerProperties.getPartitionCount()); + createTopicsIfAutoCreateEnabledAndAdminUtilsPresent(destination, producerProperties.getPartitionCount()); + Collection partitions = getPartitionsForTopic(destination, producerProperties.getPartitionCount()); if (producerProperties.getPartitionCount() < partitions.size()) { if (this.logger.isInfoEnabled()) { @@ -221,13 +221,13 @@ 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()); + createTopicsIfAutoCreateEnabledAndAdminUtilsPresent(name, properties.getPartitionCount()); + Collection partitions = getPartitionsForTopic(name, properties.getPartitionCount()); if (properties.getPartitionCount() < partitions.size()) { if (this.logger.isInfoEnabled()) { this.logger.info("The `partitionCount` of the producer for topic " + name + " is " @@ -271,9 +271,9 @@ public class KafkaMessageChannelBinder extends if (properties.getInstanceCount() == 0) { throw new IllegalArgumentException("Instance count cannot be zero"); } - - Collection allPartitions = ensureTopicCreated(name, - properties.getInstanceCount() * properties.getConcurrency()); + int partitionCount = properties.getInstanceCount() * properties.getConcurrency(); + createTopicsIfAutoCreateEnabledAndAdminUtilsPresent(name, partitionCount); + Collection allPartitions = getPartitionsForTopic(name, partitionCount); Collection listenedPartitions; @@ -297,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"); @@ -427,11 +427,24 @@ public class KafkaMessageChannelBinder extends return topicPartitionInitialOffsets; } + private void createTopicsIfAutoCreateEnabledAndAdminUtilsPresent(final String topicName, final int partitionCount) { + if (this.configurationProperties.isAutoCreateTopics() && adminUtilsOperation != null) { + createTopicAndPartitions(topicName, partitionCount); + } + else if (this.configurationProperties.isAutoCreateTopics() && adminUtilsOperation == null) { + this.logger.warn("Auto creation of topics is enabled, but Kafka AdminUtils class is not present on the classpath. " + + "No topic will be created by the binder"); + } + else if (!this.configurationProperties.isAutoCreateTopics()) { + this.logger.warn("Auto creation of topics is disabled."); + } + } + /** * Creates a Kafka topic if needed, or try to increase its partition count to the * desired number. */ - private Collection ensureTopicCreated(final String topicName, final int partitionCount) { + private void createTopicAndPartitions(final String topicName, final int partitionCount) { final ZkUtils zkUtils = ZkUtils.apply(this.configurationProperties.getZkConnectionString(), this.configurationProperties.getZkSessionTimeout(), @@ -459,62 +472,59 @@ public class KafkaMessageChannelBinder extends } } else if (errorCode == ErrorMapping.UnknownTopicOrPartitionCode()) { - if (this.configurationProperties.isAutoCreateTopics()) { - // always consider minPartitionCount for topic creation - final int effectivePartitionCount = Math.max(this.configurationProperties.getMinPartitionCount(), - partitionCount); + // always consider minPartitionCount for topic creation + final int effectivePartitionCount = Math.max(this.configurationProperties.getMinPartitionCount(), + partitionCount); - this.metadataRetryOperations.execute(new RetryCallback() { + this.metadataRetryOperations.execute(new RetryCallback() { - @Override - public Object doWithRetry(RetryContext context) throws RuntimeException { + @Override + public Object doWithRetry(RetryContext context) throws RuntimeException { - adminUtilsOperation.invokeCreateTopic(zkUtils, topicName, effectivePartitionCount, - configurationProperties.getReplicationFactor(), new Properties()); - return null; - } - }); - } - else { - throw new BinderException("Topic " + topicName + " does not exist"); - } + adminUtilsOperation.invokeCreateTopic(zkUtils, topicName, effectivePartitionCount, + configurationProperties.getReplicationFactor(), new Properties()); + return null; + } + }); } else { throw new BinderException("Error fetching Kafka topic metadata: ", ErrorMapping.exceptionFor(errorCode)); } - try { - return this.metadataRetryOperations - .execute(new RetryCallback, Exception>() { - - @Override - public Collection doWithRetry(RetryContext context) throws Exception { - Collection partitions = - getProducerFactory( - new ExtendedProducerProperties<>(new KafkaProducerProperties())) - .createProducer().partitionsFor(topicName); - - // do a sanity check on the partition set - if (partitions.size() < partitionCount) { - throw new IllegalStateException("The number of expected partitions was: " - + partitionCount + ", but " + partitions.size() - + (partitions.size() > 1 ? " have " : " has ") + "been found instead"); - } - return partitions; - } - }); - } - catch (Exception e) { - this.logger.error("Cannot initialize Binder", e); - throw new BinderException("Cannot initialize binder:", e); - } - } finally { zkUtils.close(); } } + private Collection getPartitionsForTopic(final String topicName, final int partitionCount) { + try { + return this.metadataRetryOperations + .execute(new RetryCallback, Exception>() { + + @Override + public Collection doWithRetry(RetryContext context) throws Exception { + Collection partitions = + getProducerFactory( + new ExtendedProducerProperties<>(new KafkaProducerProperties())) + .createProducer().partitionsFor(topicName); + + // do a sanity check on the partition set + if (partitions.size() < partitionCount) { + throw new IllegalStateException("The number of expected partitions was: " + + partitionCount + ", but " + partitions.size() + + (partitions.size() > 1 ? " have " : " has ") + "been found instead"); + } + return partitions; + } + }); + } + catch (Exception e) { + this.logger.error("Cannot initialize Binder", e); + throw new BinderException("Cannot initialize binder:", e); + } + } + private synchronized void initDlqProducer() { try { if (this.dlqProducer == null) { 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 index 8dab2b3f3..8bc5e47e9 100644 --- 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 @@ -24,6 +24,7 @@ import java.util.Properties; import kafka.api.PartitionMetadata; import kafka.utils.ZkUtils; +import org.springframework.util.ClassUtils; import org.springframework.util.ReflectionUtils; /** @@ -31,13 +32,11 @@ import org.springframework.util.ReflectionUtils; */ 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"); + ADMIN_UTIL_CLASS = ClassUtils.forName("kafka.admin.AdminUtils", null); } catch (ClassNotFoundException e) { throw new IllegalStateException("AdminUtils class not found", e); @@ -77,7 +76,7 @@ public class Kafka10AdminUtilsOperation implements AdminUtilsOperation { 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"); + Class topicMetadataClass = ClassUtils.forName("org.apache.kafka.common.requests.MetadataResponse$TopicMetadata", null); Method errorCodeMethod = ReflectionUtils.findMethod(topicMetadataClass, "error"); Object obj = errorCodeMethod.invoke(result); @@ -103,7 +102,7 @@ public class Kafka10AdminUtilsOperation implements AdminUtilsOperation { 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"); + Class topicMetadataClass = ClassUtils.forName("org.apache.kafka.common.requests.MetadataResponse$TopicMetadata", null); Method partitionsMetadata = ReflectionUtils.findMethod(topicMetadataClass, "partitionMetadata"); List foo = (List) partitionsMetadata.invoke(result); 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/config/KafkaBinderConfiguration.java index b926412f3..766de2e1b 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/config/KafkaBinderConfiguration.java @@ -16,13 +16,14 @@ package org.springframework.cloud.stream.binder.kafka.config; -import java.lang.reflect.Method; - +import kafka.admin.AdminUtils; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.apache.kafka.common.utils.AppInfoParser; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.autoconfigure.PropertyPlaceholderAutoConfiguration; +import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.cloud.stream.binder.Binder; @@ -58,7 +59,7 @@ import org.springframework.kafka.support.ProducerListener; @EnableConfigurationProperties({KafkaBinderConfigurationProperties.class, KafkaExtendedBindingProperties.class}) public class KafkaBinderConfiguration { - protected final Log logger = LogFactory.getLog(getClass()); + protected static final Log logger = LogFactory.getLog(KafkaBinderConfiguration.class); @Autowired private Codec codec; @@ -75,6 +76,9 @@ public class KafkaBinderConfiguration { @Autowired private ApplicationContext context; + @Autowired (required = false) + private AdminUtilsOperation adminUtilsOperation; + @Bean KafkaMessageChannelBinder kafkaMessageChannelBinder() { KafkaMessageChannelBinder kafkaMessageChannelBinder = new KafkaMessageChannelBinder( @@ -82,7 +86,6 @@ 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; } @@ -112,54 +115,21 @@ public class KafkaBinderConfiguration { 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; - } - + @ConditionalOnClass(AdminUtils.class) 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; + return AppInfoParser.getVersion().startsWith("0.10"); } } + @ConditionalOnClass(AdminUtils.class) 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; + return AppInfoParser.getVersion().startsWith("0.9"); } } } 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 eabf0bd0f..8378ad3ce 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 @@ -60,7 +60,6 @@ import org.springframework.retry.policy.SimpleRetryPolicy; import org.springframework.retry.support.RetryTemplate; import static org.assertj.core.api.Assertions.assertThat; -import static org.junit.Assert.fail; /** * @author Soby Chacko @@ -907,7 +906,7 @@ public abstract class KafkaBinderTests extends PartitionCapableBinderTests 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"); - } + String testTopicName = "createdByBroker-" + System.currentTimeMillis(); + + QueueChannel input = new QueueChannel(); + Binding producerBinding = binder.bindProducer(testTopicName, output, + createProducerProperties()); + String testPayload = "foo1-" + UUID.randomUUID().toString(); + output.send(new GenericMessage<>(testPayload.getBytes())); + + Binding consumerBinding = binder.bindConsumer(testTopicName, "test", input, consumerProperties); + Message receivedMessage2 = (Message) receive(input); + assertThat(receivedMessage2).isNotNull(); + assertThat(new String(receivedMessage2.getPayload())).isEqualTo(testPayload); + + producerBinding.unbind(); + consumerBinding.unbind(); } @Test