You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by da...@apache.org on 2017/06/23 14:56:35 UTC
kafka git commit: HOTFIX: Don't check metadata unless you are
creating topic
Repository: kafka
Updated Branches:
refs/heads/trunk 9ada0f816 -> b49036873
HOTFIX: Don't check metadata unless you are creating topic
During a broker rolling upgrade, it's likely we don't have enough brokers ready yet. If streams does not need to create a topic it shouldn't check how many brokers are up.
The system test for this is in a separate PR: https://github.com/apache/kafka/pull/3411
Author: Eno Thereska <en...@gmail.com>
Reviewers: Damian Guy <da...@gmail.com>
Closes #3418 from enothereska/hotfix-replication
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/b4903687
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/b4903687
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/b4903687
Branch: refs/heads/trunk
Commit: b490368735b3bfe82acab3f2db4894a757abb966
Parents: 9ada0f8
Author: Eno Thereska <en...@gmail.com>
Authored: Fri Jun 23 15:56:29 2017 +0100
Committer: Damian Guy <da...@gmail.com>
Committed: Fri Jun 23 15:56:29 2017 +0100
----------------------------------------------------------------------
.../internals/InternalTopicManager.java | 14 ++++++-----
.../internals/InternalTopicManagerTest.java | 25 ++++++++++++++++++++
2 files changed, 33 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/b4903687/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java
index fd602cc..cd57867 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java
@@ -63,13 +63,15 @@ public class InternalTopicManager {
final MetadataResponse metadata = streamsKafkaClient.fetchMetadata();
final Map<String, Integer> existingTopicPartitions = fetchExistingPartitionCountByTopic(metadata);
final Map<InternalTopicConfig, Integer> topicsToBeCreated = validateTopicPartitions(topics, existingTopicPartitions);
- if (metadata.brokers().size() < replicationFactor) {
- throw new StreamsException("Found only " + metadata.brokers().size() + " brokers, " +
- " but replication factor is " + replicationFactor + "." +
- " Decrease replication factor for internal topics via StreamsConfig parameter \"replication.factor\"" +
- " or add more brokers to your cluster.");
+ if (topicsToBeCreated.size() > 0) {
+ if (metadata.brokers().size() < replicationFactor) {
+ throw new StreamsException("Found only " + metadata.brokers().size() + " brokers, " +
+ " but replication factor is " + replicationFactor + "." +
+ " Decrease replication factor for internal topics via StreamsConfig parameter \"replication.factor\"" +
+ " or add more brokers to your cluster.");
+ }
+ streamsKafkaClient.createTopics(topicsToBeCreated, replicationFactor, windowChangeLogAdditionalRetention, metadata);
}
- streamsKafkaClient.createTopics(topicsToBeCreated, replicationFactor, windowChangeLogAdditionalRetention, metadata);
return;
} catch (StreamsException ex) {
log.warn("Could not create internal topics: {} Retry #{}", ex.getMessage(), i);
http://git-wip-us.apache.org/repos/asf/kafka/blob/b4903687/streams/src/test/java/org/apache/kafka/streams/processor/internals/InternalTopicManagerTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/InternalTopicManagerTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/InternalTopicManagerTest.java
index bf3259e..aef0d35 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/InternalTopicManagerTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/InternalTopicManagerTest.java
@@ -81,6 +81,31 @@ public class InternalTopicManagerTest {
Assert.assertTrue(exceptionWasThrown);
}
+ @Test
+ public void shouldNotThrowExceptionIfExistsWithDifferentReplication() throws Exception {
+
+ // create topic the first time with replication 2
+ InternalTopicManager internalTopicManager = new InternalTopicManager(streamsKafkaClient, 2,
+ WINDOW_CHANGE_LOG_ADDITIONAL_RETENTION_DEFAULT, time);
+ internalTopicManager.makeReady(Collections.singletonMap(new InternalTopicConfig(topic, Collections.singleton(InternalTopicConfig.CleanupPolicy.compact), null), 1));
+
+ // attempt to create it again with replication 1
+ InternalTopicManager internalTopicManager2 = new InternalTopicManager(streamsKafkaClient, 1,
+ WINDOW_CHANGE_LOG_ADDITIONAL_RETENTION_DEFAULT, time);
+ try {
+ internalTopicManager2.makeReady(Collections.singletonMap(new InternalTopicConfig(topic, Collections.singleton(InternalTopicConfig.CleanupPolicy.compact), null), 1));
+ } catch (StreamsException e) {
+ Assert.fail("did not expect an exception since topic is already there.");
+ }
+ }
+
+ @Test
+ public void shouldNotThrowExceptionForEmptyTopicMap() throws Exception {
+ InternalTopicManager internalTopicManager = new InternalTopicManager(streamsKafkaClient, 1,
+ WINDOW_CHANGE_LOG_ADDITIONAL_RETENTION_DEFAULT, time);
+ internalTopicManager.makeReady(Collections.EMPTY_MAP);
+ }
+
private Properties configProps() {
return new Properties() {
{