SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Listened partitions assertions

Browse Source 
Avoid unnecessary assertions on listened parttions when
autorebalancing is enabled, but no listened partitions are found.

Polish concurrency assignment when listened partition are empty

polishing the provisioner

Resolves #512
Resolves #587
This commit is contained in:
Soby Chacko
2019-03-23 16:52:33 -04:00
committed by Oleg Zhurakousky
parent 73d3d79651
commit 5660c7cf76
2 changed files with 13 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java
View File

@@ -463,6 +463,8 @@ public class KafkaTopicProvisioner implements
}
this.logger.error("Failed to obtain partition information", ex);
}
// In some cases, the above partition query may not throw an UnknownTopic..Exception for various reasons.
// For that, we are forcing another query to ensure that the topic is present on the server.
if (CollectionUtils.isEmpty(partitions)) {
final AdminClient adminClient = AdminClient
.create(this.adminClientProperties);
@@ -483,7 +485,7 @@ public class KafkaTopicProvisioner implements
}
}
// do a sanity check on the partition set
int partitionSize = partitions.size();
int partitionSize = CollectionUtils.isEmpty(partitions) ? 0 : partitions.size();
if (partitionSize < partitionCount) {
if (tolerateLowerPartitionsOnBroker) {
this.logger.warn("The number of expected partitions was: "

14
spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
View File

@@ -483,8 +483,10 @@ public class KafkaMessageChannelBinder extends
for (int i = 0; i < topics.length; i++) {
topics[i] = topics[i].trim();
}
Assert.isTrue(usingPatterns || !CollectionUtils.isEmpty(listenedPartitions),
"A list of partitions must be provided");
if (!usingPatterns && !groupManagement) {
Assert.isTrue(!CollectionUtils.isEmpty(listenedPartitions),
"A list of partitions must be provided");
}
final TopicPartitionInitialOffset[] topicPartitionInitialOffsets = getTopicPartitionInitialOffsets(
listenedPartitions);
final ContainerProperties containerProperties = anonymous
@@ -504,6 +506,11 @@ public class KafkaMessageChannelBinder extends
int concurrency = usingPatterns ? extendedConsumerProperties.getConcurrency()
: Math.min(extendedConsumerProperties.getConcurrency(),
listenedPartitions.size());
// in the event that auto rebalance is enabled, but no listened partitions are found
// we want to make sure that concurrency is a non-zero value.
if (groupManagement && listenedPartitions.isEmpty()) {
concurrency = extendedConsumerProperties.getConcurrency();
}
resetOffsets(extendedConsumerProperties, consumerFactory, groupManagement,
containerProperties);
@SuppressWarnings("rawtypes")
@@ -844,8 +851,7 @@ public class KafkaMessageChannelBinder extends
extendedConsumerProperties.getExtension().isAutoRebalanceEnabled(),
() -> {
try (Consumer<?, ?> consumer = consumerFactory.createConsumer()) {
List<PartitionInfo> partitionsFor = consumer.partitionsFor(topic);
return partitionsFor;
return consumer.partitionsFor(topic);
}
}, topic);
}