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

KafkaTopidProvisioner retry refactoring

Browse Source 
Use a retry template for topic description method call through admin client
when provisioning consumer destinations. We are retrying because in the event
this operation gets failed, it is retried with the default retry settings in the provisioner.

Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2520
This commit is contained in:
omercelikceng
2022-10-16 14:58:06 +03:00
committed by Soby Chacko
parent 0e5754f24f
commit cc04b0b13b
1 changed files with 35 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -85,6 +85,7 @@ import org.springframework.util.StringUtils;
* @author Oleg Zhurakousky
* @author Aldo Sinanaj
* @author Yi Liu
* @author Omer Celik
*/
public class KafkaTopicProvisioner implements
// @checkstyle:off
@@ -179,28 +180,7 @@ public class KafkaTopicProvisioner implements
try (AdminClient adminClient = createAdminClient()) {
createTopic(adminClient, name, properties.getPartitionCount(), false,
properties.getExtension().getTopic());
int partitions = 0;
Map<String, TopicDescription> topicDescriptions = new HashMap<>();
this.metadataRetryOperations.execute(context -> {
try {
if (logger.isDebugEnabled()) {
logger.debug("Attempting to retrieve the description for the topic: " + name);
}
DescribeTopicsResult describeTopicsResult = adminClient
.describeTopics(Collections.singletonList(name));
KafkaFuture<Map<String, TopicDescription>> all = describeTopicsResult
.allTopicNames();
topicDescriptions.putAll(all.get(this.operationTimeout, TimeUnit.SECONDS));
}
catch (Exception ex) {
throw new ProvisioningException("Problems encountered with partitions finding for: " + name, ex);
}
return null;
});
TopicDescription topicDescription = topicDescriptions.get(name);
if (topicDescription != null) {
partitions = topicDescription.partitions().size();
}
int partitions = getPartitionsForTopic(name, adminClient);
return new KafkaProducerDestination(name, partitions);
}
}
@@ -250,30 +230,44 @@ public class KafkaTopicProvisioner implements
createTopic(adminClient, name, partitionCount,
properties.getExtension().isAutoRebalanceEnabled(),
properties.getExtension().getTopic());
DescribeTopicsResult describeTopicsResult = adminClient
.describeTopics(Collections.singletonList(name));
KafkaFuture<Map<String, TopicDescription>> all = describeTopicsResult
.allTopicNames();
try {
Map<String, TopicDescription> topicDescriptions = all
.get(this.operationTimeout, TimeUnit.SECONDS);
TopicDescription topicDescription = topicDescriptions.get(name);
int partitions = topicDescription.partitions().size();
consumerDestination = createDlqIfNeedBe(adminClient, name, group,
properties, anonymous, partitions);
if (consumerDestination == null) {
consumerDestination = new KafkaConsumerDestination(name,
partitions);
}
return consumerDestination;
}
catch (Exception ex) {
throw new ProvisioningException("Provisioning exception encountered for " + name, ex);
int partitions = getPartitionsForTopic(name, adminClient);
consumerDestination = createDlqIfNeedBe(adminClient, name, group,
properties, anonymous, partitions);
if (consumerDestination == null) {
consumerDestination = new KafkaConsumerDestination(name,
partitions);
}
return consumerDestination;
}
}
return kafkaConsumerDestination;
}
private int getPartitionsForTopic(String topicName, AdminClient adminClient) {
int partitions = 0;
Map<String, TopicDescription> topicDescriptions = retrieveTopicDescriptions(topicName, adminClient);
TopicDescription topicDescription = topicDescriptions.get(topicName);
if (topicDescription != null) {
partitions = topicDescription.partitions().size();
}
return partitions;
}
private Map<String, TopicDescription> retrieveTopicDescriptions(String topicName, AdminClient adminClient) {
return this.metadataRetryOperations.execute(context -> {
try {
if (logger.isDebugEnabled()) {
logger.debug("Attempting to retrieve the description for the topic: " + topicName);
}
DescribeTopicsResult describeTopicsResult = adminClient
.describeTopics(Collections.singletonList(topicName));
KafkaFuture<Map<String, TopicDescription>> all = describeTopicsResult
.allTopicNames();
return all.get(this.operationTimeout, TimeUnit.SECONDS);
} catch (Exception ex) {
throw new ProvisioningException("Problems encountered with partitions finding for: " + topicName, ex);
}
});
}
AdminClient createAdminClient() {
return AdminClient.create(this.adminClientProperties);