Spring/spring-kafka
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Introduce an intermediate variable for topicNameToTopic.get(n)

Browse Source
This commit is contained in:
Johnny Lim
2017-12-14 01:31:44 +09:00
committed by Artem Bilan
parent fa2e0230ab
commit 1043cb115e
1 changed files with 7 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
spring-kafka/src/main/java/org/springframework/kafka/core/KafkaAdmin.java
View File

@@ -193,23 +193,24 @@ public class KafkaAdmin implements ApplicationContextAware, SmartInitializingSin
List<NewTopic> topicsToAdd = new ArrayList<>();
Map<String, NewPartitions> topicsToModify = new HashMap<>();
topicInfo.values().forEach((n, f) -> {
NewTopic topic = topicNameToTopic.get(n);
try {
TopicDescription topicDescription = f.get(this.operationTimeout, TimeUnit.SECONDS);
if (topicNameToTopic.get(n).numPartitions() < topicDescription.partitions().size()) {
if (topic.numPartitions() < topicDescription.partitions().size()) {
if (logger.isInfoEnabled()) {
logger.info(String.format(
"Topic '%s' exists but has a different partition count: %d not %d", n,
topicDescription.partitions().size(), topicNameToTopic.get(n).numPartitions()));
topicDescription.partitions().size(), topic.numPartitions()));
}
}
else if (topicNameToTopic.get(n).numPartitions() > topicDescription.partitions().size()) {
else if (topic.numPartitions() > topicDescription.partitions().size()) {
if (logger.isInfoEnabled()) {
logger.info(String.format(
"Topic '%s' exists but has a different partition count: %d not %d, increasing "
+ "if the broker supports it", n,
topicDescription.partitions().size(), topicNameToTopic.get(n).numPartitions()));
topicDescription.partitions().size(), topic.numPartitions()));
}
topicsToModify.put(n, NewPartitions.increaseTo(topicNameToTopic.get(n).numPartitions()));
topicsToModify.put(n, NewPartitions.increaseTo(topic.numPartitions()));
}
}
catch (InterruptedException e) {
@@ -219,7 +220,7 @@ public class KafkaAdmin implements ApplicationContextAware, SmartInitializingSin
throw new KafkaException("Timed out waiting to get existing topics", e);
}
catch (ExecutionException e) {
topicsToAdd.add(topicNameToTopic.get(n));
topicsToAdd.add(topic);
}
});
if (topicsToAdd.size() > 0) {