From cc04b0b13b12436dfeaf3f704f03c6c7f106d76a Mon Sep 17 00:00:00 2001
From: omercelikceng <omercelik.ceng@gmail.com>
Date: Sun, 16 Oct 2022 14:58:06 +0300
Subject: [PATCH] KafkaTopidProvisioner retry refactoring

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
---
 .../provisioning/KafkaTopicProvisioner.java   | 76 +++++++++----------
 1 file changed, 35 insertions(+), 41 deletions(-)

diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java
index 61723b9b4..d3ff0713d 100644
--- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java
+++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java
@@ -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);