From 02ee8d4d88fc3bebfcb6a013ec72b16cbf5513ca Mon Sep 17 00:00:00 2001 From: Marius Bogoevici Date: Tue, 3 May 2016 16:39:58 -0400 Subject: [PATCH] GH-509: Rework Topic Autoconfiguration Fixes #509 - prevent autoconfiguration from reducing the existing partition count and/or replication factor Split `autoConfigureTopics` in two separate settings, one for `autoCreateTopics` and one for `autoAddPartitions` (to existing topics) Renamed `defaultMinPartitionCount` to just `minPartitionCount` on the Kafka binder --- .../kafka/KafkaMessageChannelBinder.java | 117 ++++++++++++------ .../config/KafkaBinderConfiguration.java | 5 +- .../KafkaBinderConfigurationProperties.java | 20 ++- .../stream/binder/kafka/KafkaBinderTests.java | 86 ++++++++++--- .../stream/binder/kafka/KafkaTestBinder.java | 2 +- .../binder/kafka/RawModeKafkaBinderTests.java | 10 +- .../spring-cloud-stream-overview.adoc | 30 ++++- .../cloud/stream/binder/BinderException.java | 7 +- 8 files changed, 205 insertions(+), 72 deletions(-) diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java index 375f0d028..44c732138 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java @@ -34,6 +34,8 @@ import java.util.concurrent.atomic.AtomicInteger; import kafka.admin.AdminUtils; import kafka.api.OffsetRequest; +import kafka.api.TopicMetadata; +import kafka.common.ErrorMapping; import kafka.serializer.Decoder; import kafka.serializer.DefaultDecoder; import kafka.utils.ZKStringSerializer$; @@ -126,7 +128,9 @@ public class KafkaMessageChannelBinder public static final DaemonThreadFactory DAEMON_THREAD_FACTORY = new DaemonThreadFactory(); - private boolean autoConfigureTopics = true; + private boolean autoCreateTopics = true; + + private boolean autoAddPartitions = false; private RetryOperations metadataRetryOperations; @@ -152,7 +156,7 @@ public class KafkaMessageChannelBinder private int fetchSize = 1024 * 1024; - private int defaultMinPartitionCount = 1; + private int minPartitionCount = 1; private ConnectionFactory connectionFactory; @@ -298,8 +302,8 @@ public class KafkaMessageChannelBinder this.fetchSize = fetchSize; } - public void setDefaultMinPartitionCount(int defaultMinPartitionCount) { - this.defaultMinPartitionCount = defaultMinPartitionCount; + public void setMinPartitionCount(int minPartitionCount) { + this.minPartitionCount = minPartitionCount; } public void setMaxWait(int maxWait) { @@ -322,12 +326,20 @@ public class KafkaMessageChannelBinder this.zkConnectionTimeout = zkConnectionTimeout; } - public boolean isAutoConfigureTopics() { - return autoConfigureTopics; + public boolean isAutoCreateTopics() { + return autoCreateTopics; } - public void setAutoConfigureTopics(boolean autoConfigureTopics) { - this.autoConfigureTopics = autoConfigureTopics; + public void setAutoCreateTopics(boolean autoCreateTopics) { + this.autoCreateTopics = autoCreateTopics; + } + + public boolean isAutoAddPartitions() { + return autoAddPartitions; + } + + public void setAutoAddPartitions(boolean autoAddPartitions) { + this.autoAddPartitions = autoAddPartitions; } @Override @@ -381,8 +393,15 @@ public class KafkaMessageChannelBinder validateTopicName(name); - int numPartitions = Math.max(defaultMinPartitionCount, properties.getPartitionCount()); - Collection partitions = ensureTopicCreated(name, numPartitions, replicationFactor); + Collection partitions = ensureTopicCreated(name, properties.getPartitionCount()); + + if (properties.getPartitionCount() < partitions.size()) { + if (logger.isInfoEnabled()) { + logger.info("The `partitionCount` of the producer for topic " + name + " is " + + properties.getPartitionCount() + ", smaller than the actual partition count of " + + partitions.size() + " of the topic. The larger number will be used instead."); + } + } topicsInUse.put(name, partitions); @@ -429,28 +448,53 @@ public class KafkaMessageChannelBinder * 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 minExpectedPartitions, - int replicationFactor) { + private Collection ensureTopicCreated(final String topicName, final int partitionCount) { final ZkClient zkClient = new ZkClient(zkAddress, getZkSessionTimeout(), getZkConnectionTimeout(), ZKStringSerializer$.MODULE$); try { - // The following is basically copy/paste from AdminUtils.createTopic() with - // createOrUpdateTopicPartitionAssignmentPathInZK(..., update=true) final Properties topicConfig = new Properties(); - if (isAutoConfigureTopics()) { - Seq brokerList = ZkUtils.getSortedBrokerList(zkClient); - final scala.collection.Map> replicaAssignment = AdminUtils - .assignReplicasToBrokers(brokerList, minExpectedPartitions, replicationFactor, -1, -1); - metadataRetryOperations.execute(new RetryCallback() { - - @Override - public Object doWithRetry(RetryContext context) throws RuntimeException { - AdminUtils.createOrUpdateTopicPartitionAssignmentPathInZK(zkClient, topicName, replicaAssignment, - topicConfig, true); - return null; + TopicMetadata topicMetadata = AdminUtils.fetchTopicMetadataFromZk(topicName, zkClient); + if (topicMetadata.errorCode() == ErrorMapping.NoError()) { + // only consider minPartitionCount for resizing if autoAddPartitions is true + int effectivePartitionCount = isAutoAddPartitions() ? Math.max(minPartitionCount, partitionCount) + : partitionCount; + if (topicMetadata.partitionsMetadata().size() < effectivePartitionCount) { + if (isAutoAddPartitions()) { + AdminUtils.addPartitions(zkClient, topicName, effectivePartitionCount, null, false, + new Properties()); } - }); + 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." + + "Consider either increasing the partition count of the topic or enabling `autoAddPartitions`"); + } + } + } + else if (topicMetadata.errorCode() == ErrorMapping.UnknownTopicOrPartitionCode()) { + if (isAutoCreateTopics()) { + Seq brokerList = ZkUtils.getSortedBrokerList(zkClient); + // always consider minPartitionCount for topic creation + int effectivePartitionCount = Math.max(this.minPartitionCount, partitionCount); + final scala.collection.Map> replicaAssignment = AdminUtils + .assignReplicasToBrokers(brokerList, effectivePartitionCount, replicationFactor, -1, -1); + metadataRetryOperations.execute(new RetryCallback() { + @Override + public Object doWithRetry(RetryContext context) throws RuntimeException { + AdminUtils.createOrUpdateTopicPartitionAssignmentPathInZK(zkClient, topicName, + replicaAssignment, topicConfig, true); + return null; + } + }); + } + else { + throw new BinderException("Topic " + topicName + " does not exist"); + } + } + else { + throw new BinderException("Error fetching Kafka topic metadata: ", + ErrorMapping.exceptionFor(topicMetadata.errorCode())); } try { Collection partitions = metadataRetryOperations @@ -460,11 +504,11 @@ public class KafkaMessageChannelBinder public Collection doWithRetry(RetryContext context) throws Exception { connectionFactory.refreshMetadata(Collections.singleton(topicName)); Collection partitions = connectionFactory.getPartitions(topicName); - if (partitions.size() < minExpectedPartitions) { - throw new IllegalStateException( - "The number of expected partitions was: " + minExpectedPartitions + ", but " - + partitions.size() + (partitions.size() > 1 ? " have " : " has ") - + "been found instead"); + // 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"); } connectionFactory.getLeaders(partitions); return partitions; @@ -487,26 +531,25 @@ public class KafkaMessageChannelBinder ExtendedConsumerProperties properties, String group, long referencePoint) { validateTopicName(name); - int instance = properties.getInstanceCount(); - if (instance == 0) { + if (properties.getInstanceCount() == 0) { throw new IllegalArgumentException("Instance count cannot be zero"); } - int numPartitions = Math.max(this.defaultMinPartitionCount, instance * properties.getConcurrency()); - Collection allPartitions = ensureTopicCreated(name, numPartitions, replicationFactor); + Collection allPartitions = ensureTopicCreated(name, + properties.getInstanceCount() * properties.getConcurrency()); Decoder valueDecoder = new DefaultDecoder(null); Decoder keyDecoder = new DefaultDecoder(null); Collection listenedPartitions; - if (instance == 1) { + if (properties.getInstanceCount() == 1) { listenedPartitions = allPartitions; } else { listenedPartitions = new ArrayList<>(); for (Partition partition : allPartitions) { // divide partitions across modules - if ((partition.getId() % instance) == properties.getInstanceIndex()) { + if ((partition.getId() % properties.getInstanceCount()) == properties.getInstanceIndex()) { listenedPartitions.add(partition); } } diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java index 83c44e391..2a1a461bc 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java @@ -84,12 +84,13 @@ public class KafkaBinderConfiguration { kafkaMessageChannelBinder.setZkConnectionTimeout(kafkaBinderConfigurationProperties.getZkConnectionTimeout()); kafkaMessageChannelBinder.setFetchSize(kafkaBinderConfigurationProperties.getFetchSize()); - kafkaMessageChannelBinder.setDefaultMinPartitionCount(kafkaBinderConfigurationProperties.getMinPartitionCount()); + kafkaMessageChannelBinder.setMinPartitionCount(kafkaBinderConfigurationProperties.getMinPartitionCount()); kafkaMessageChannelBinder.setQueueSize(kafkaBinderConfigurationProperties.getQueueSize()); kafkaMessageChannelBinder.setReplicationFactor(kafkaBinderConfigurationProperties.getReplicationFactor()); kafkaMessageChannelBinder.setRequiredAcks(kafkaBinderConfigurationProperties.getRequiredAcks()); kafkaMessageChannelBinder.setMaxWait(kafkaBinderConfigurationProperties.getMaxWait()); - kafkaMessageChannelBinder.setAutoConfigureTopics(kafkaBinderConfigurationProperties.isAutoConfigureTopics()); + kafkaMessageChannelBinder.setAutoCreateTopics(kafkaBinderConfigurationProperties.isAutoCreateTopics()); + kafkaMessageChannelBinder.setAutoAddPartitions(kafkaBinderConfigurationProperties.isAutoAddPartitions()); kafkaMessageChannelBinder.setProducerListener(producerListener); kafkaMessageChannelBinder.setExtendedBindingProperties(kafkaExtendedBindingProperties); return kafkaMessageChannelBinder; diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfigurationProperties.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfigurationProperties.java index c266595cd..9bf123074 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfigurationProperties.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfigurationProperties.java @@ -45,7 +45,9 @@ public class KafkaBinderConfigurationProperties { private int maxWait = 100; - private boolean autoConfigureTopics = true; + private boolean autoCreateTopics = true; + + private boolean autoAddPartitions = false; /** * ZK session timeout in milliseconds. @@ -206,11 +208,19 @@ public class KafkaBinderConfigurationProperties { this.queueSize = queueSize; } - public boolean isAutoConfigureTopics() { - return autoConfigureTopics; + public boolean isAutoCreateTopics() { + return autoCreateTopics; } - public void setAutoConfigureTopics(boolean autoConfigureTopics) { - this.autoConfigureTopics = autoConfigureTopics; + public void setAutoCreateTopics(boolean autoCreateTopics) { + this.autoCreateTopics = autoCreateTopics; + } + + public boolean isAutoAddPartitions() { + return autoAddPartitions; + } + + public void setAutoAddPartitions(boolean autoAddPartitions) { + this.autoAddPartitions = autoAddPartitions; } } diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java index f8675ebe1..959281c92 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java @@ -36,6 +36,8 @@ import java.util.Collection; import java.util.Properties; import java.util.UUID; +import kafka.admin.AdminUtils; +import kafka.api.TopicMetadata; import org.hamcrest.collection.IsCollectionWithSize; import org.junit.Before; import org.junit.ClassRule; @@ -69,8 +71,6 @@ import org.springframework.retry.backoff.FixedBackOffPolicy; import org.springframework.retry.policy.SimpleRetryPolicy; import org.springframework.retry.support.RetryTemplate; -import kafka.admin.AdminUtils; - /** * Integration tests for the {@link KafkaMessageChannelBinder}. @@ -465,13 +465,13 @@ public class KafkaBinderTests extends PartitionCapableBinderTests consumerProperties = createConsumerProperties(); + Binding binding = binder.doBindConsumer(testTopicName, "test", output, consumerProperties); + binding.unbind(); + TopicMetadata topicMetadata = AdminUtils.fetchTopicMetadataFromZk(testTopicName, kafkaTestSupport.getZkClient()); + assertThat(topicMetadata.partitionsMetadata().size(), equalTo(6)); + } + + @Test + public void testPartitionCountIncreasedIfAutoAddPartitionsSet() throws Exception { + String testTopicName = "existing" + System.currentTimeMillis(); + AdminUtils.createTopic(kafkaTestSupport.getZkClient(), testTopicName, 1, 1, new Properties()); + + KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder( + new ZookeeperConnect(kafkaTestSupport.getZkConnectString()), kafkaTestSupport.getBrokerAddress(), + kafkaTestSupport.getZkConnectString()); + + binder.setMinPartitionCount(6); + + GenericApplicationContext context = new GenericApplicationContext(); + binder.setAutoAddPartitions(true); + 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, kafkaTestSupport.getZkClient()); + assertThat(topicMetadata.partitionsMetadata().size(), equalTo(6)); + } } diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTestBinder.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTestBinder.java index 06dfeb5f9..e88e01b91 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTestBinder.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTestBinder.java @@ -62,7 +62,7 @@ public class KafkaTestBinder extends binder.setApplicationContext(context); binder.setFetchSize(binderConfiguration.getFetchSize()); binder.setMaxWait(binderConfiguration.getMaxWait()); - binder.setDefaultMinPartitionCount(binderConfiguration.getMinPartitionCount()); + binder.setMinPartitionCount(binderConfiguration.getMinPartitionCount()); binder.afterPropertiesSet(); this.setBinder(binder); } diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/RawModeKafkaBinderTests.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/RawModeKafkaBinderTests.java index 5f88e7009..9da1a3caf 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/RawModeKafkaBinderTests.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/RawModeKafkaBinderTests.java @@ -26,6 +26,9 @@ import static org.junit.Assert.assertThat; import java.util.Arrays; +import org.junit.Ignore; +import org.junit.Test; + import org.springframework.cloud.stream.binder.BinderHeaders; import org.springframework.cloud.stream.binder.Binding; import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; @@ -40,9 +43,6 @@ import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.support.GenericMessage; -import org.junit.Ignore; -import org.junit.Test; - /** * @author Marius Bogoevici * @author David Turanski @@ -59,7 +59,7 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests { properties.setHeaderMode(HeaderMode.raw); properties.setPartitionKeyExtractorClass(RawKafkaPartitionTestSupport.class); properties.setPartitionSelectorClass(RawKafkaPartitionTestSupport.class); - properties.setPartitionCount(3); + properties.setPartitionCount(6); DirectChannel output = new DirectChannel(); output.setBeanName("test.output"); @@ -113,7 +113,7 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests { ExtendedProducerProperties properties = createProducerProperties(); properties.setPartitionKeyExpression(spelExpressionParser.parseExpression("payload[0]")); properties.setPartitionSelectorExpression(spelExpressionParser.parseExpression("hashCode()")); - properties.setPartitionCount(3); + properties.setPartitionCount(6); properties.setHeaderMode(HeaderMode.raw); DirectChannel output = new DirectChannel(); diff --git a/spring-cloud-stream-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc b/spring-cloud-stream-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc index 6a757d6a8..d5db979d1 100644 --- a/spring-cloud-stream-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc +++ b/spring-cloud-stream-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc @@ -982,20 +982,29 @@ spring.cloud.stream.kafka.binder.requiredAcks:: + Default: `1`. spring.cloud.stream.kafka.binder.minPartitionCount:: - The minimum number of partitions expected by the consumer if it creates the consumed topic automatically. + Effective only if `autoCreateTopics` or `autoAddPartitions` is set. + The global minimum number of partitions that the binder will configure on topics on which it produces/consumes data. +It can be superseded by the `partitionCount` setting of the producer or by the value of +`instanceCount` * `concurrency` settings of the producer (if either is larger). + Default: `1`. spring.cloud.stream.kafka.binder.replicationFactor:: - The replication factor of auto-created topics if `autoConfigureTopics` is active. + The replication factor of auto-created topics if `autoCreateTopics` is active. + Default: `1`. -spring.cloud.stream.kafka.binder.autoConfigureTopics:: - If set to `true`, the binder will create new topics or add new partitions to existing topics automatically to match the requirements of the producers and consumers. +spring.cloud.stream.kafka.binder.autoCreateTopics:: + If set to `true`, the binder will create new topics automatically. If set to `false`, the binder will rely on the topics being already configured. -In the latter case, if the topics do not exist or the partition count is smaller than the expected partition count, the binder will fail to start. +In the latter case, if the topics do not exist, the binder will fail to start. Of note, this setting is independent of the `auto.topic.create.enable` setting of the broker and it does not influence it: if the server is set to auto-create topics, they may be created as part of the metadata retrieval request, with default broker settings. + Default: `true`. +spring.cloud.stream.kafka.binder.autoAddPartitions:: + If set to `true`, the binder will create add new partitions if required. +If set to `false`, the binder will rely on the partition size of the topic being already configured. +If the partition count of the target topic is smaller than the expected value, the binder will fail to start. +Default: `false`. + ==== Kafka Consumer Properties @@ -1168,7 +1177,7 @@ When scaling up Spring Cloud Stream applications, each instance can receive info Spring Cloud Stream does this through the `spring.cloud.stream.instanceCount` and `spring.cloud.stream.instanceIndex` properties. For example, if there are three instances of a HDFS sink application, all three instances will have `spring.cloud.stream.instanceCount` set to `3`, and the individual applications will have `spring.cloud.stream.instanceIndex` set to `0`, `1`, and `2`, respectively. -When Spring Cloud Stream applications are deployed via Spring Cloud Dataflow, these properties are configured automatically; when Spring Cloud Stream applications are launched independently, these properties must be set correctly. +When Spring Cloud Stream applications are deployed via Spring Cloud Data Flow, these properties are configured automatically; when Spring Cloud Stream applications are launched independently, these properties must be set correctly. By default, `spring.cloud.stream.instanceCount` is `1`, and `spring.cloud.stream.instanceIndex` is `0`. In a scaled-up scenario, correct configuration of these two properties is important for addressing partitioning behavior (see below) in general, and the two properties are always required by certain binders (e.g., the Kafka binder) in order to ensure that data are split correctly across multiple consumer instances. @@ -1202,6 +1211,15 @@ This can be customized on the binding, either by setting a SpEL expression to be Additional properties can be configured for more advanced scenarios, as described in the following section. +[NOTE] +==== +The Kafka binder will use the `partitionCount` setting as a hint to create a topic with the given partition count (in conjunction with the `minPartitionCount`, the maximum of the two being the value being used). +Exercise caution when configuring both `minPartitionCount` for a binder and `partitionCount` for an application, as the larger value will be used. +If a topic already exists with a smaller partition count and `autoAddPartitions` is disabled (the default), then the binder will fail to start. +If a topic already exists with a smaller partition count and `autoAddPartitions` is enabled, new partitions will be added. +If a topic already exists with a larger number of partitions than the maximum of (`minPartitionCount` and `partitionCount`), the existing partition count will be used. +==== + ===== Configuring Input Bindings for Partitioning An input binding is configured to receive partitioned data by setting its `partitioned` property, as well as the `instanceIndex` and `instanceCount` properties on the application itself, as in the following example: diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderException.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderException.java index 957d8e7bd..ad1766f27 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderException.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderException.java @@ -1,5 +1,5 @@ /* - * Copyright 2014 the original author or authors. + * 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. @@ -22,10 +22,15 @@ package org.springframework.cloud.stream.binder; * likely a configuration error). * * @author Gary Russell + * @author Marius Bogoevici */ @SuppressWarnings("serial") public class BinderException extends RuntimeException { + public BinderException(String message) { + super(message); + } + public BinderException(String message, Throwable cause) { super(message, cause); }