Add retries to WorkerUtils#verifyTopics
Reviewers: Colin P. McCabe
@@ -44,6 +44,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
+import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.regex.Pattern;
@@ -160,7 +161,7 @@ public final class WorkerUtils {
log.warn("Topic(s) {} already exist.", topicsExists);
throw new TopicExistsException("One or more topics already exist.");
} else {
- verifyTopics(log, adminClient, topicsExists, topics);
+ verifyTopics(log, adminClient, topicsExists, topics, 3, 2500);
}
}
}
@@ -240,17 +241,20 @@ public final class WorkerUtils {
* @param topicsToVerify List of topics to verify
* @param topicsInfo Map of topic name to topic description, which includes topics in
* 'topicsToVerify' list.
+ * @param retryCount The number of times to retry the fetching of the topics
+ * @param retryBackoffMs The amount of time, in milliseconds, to wait in between retries
* @throws UnknownTopicOrPartitionException If at least one topic contained in 'topicsInfo'
- * does not exist
+ * does not exist after retrying.
* @throws RuntimeException If one or more topics have different number of partitions than
* described in 'topicsInfo'
*/
private static void verifyTopics(
Logger log, AdminClient adminClient,
- Collection<String> topicsToVerify, Map<String, NewTopic> topicsInfo) throws Throwable {
- DescribeTopicsResult topicsResult = adminClient.describeTopics(
- topicsToVerify, new DescribeTopicsOptions().timeoutMs(ADMIN_REQUEST_TIMEOUT));
- Map<String, TopicDescription> topicDescriptionMap = topicsResult.all().get();
+ Collection<String> topicsToVerify, Map<String, NewTopic> topicsInfo, int retryCount, long retryBackoffMs) throws Throwable {
+
+ Map<String, TopicDescription> topicDescriptionMap = topicDescriptions(topicsToVerify, adminClient,
+ retryCount, retryBackoffMs);
+
for (TopicDescription desc: topicDescriptionMap.values()) {
// map will always contain the topic since all topics in 'topicsExists' are in given
// 'topics' map
@@ -265,6 +269,24 @@ public final class WorkerUtils {
}
}
+ private static Map<String, TopicDescription> topicDescriptions(Collection<String> topicsToVerify,
+ AdminClient adminClient,
+ int retryCount, long retryBackoffMs)
+ throws ExecutionException, InterruptedException {
+ UnknownTopicOrPartitionException lastException = null;
+ for (int i = 0; i < retryCount; i++) {
+ try {
+ DescribeTopicsResult topicsResult = adminClient.describeTopics(
+ topicsToVerify, new DescribeTopicsOptions().timeoutMs(ADMIN_REQUEST_TIMEOUT));
+ return topicsResult.all().get();
+ } catch (UnknownTopicOrPartitionException exception) {
+ lastException = exception;
+ Thread.sleep(retryBackoffMs);
+ }
+ }
+ throw lastException;
+ }
+
/**
* Returns list of existing, not internal, topics/partitions that match given pattern and
* where partitions are in range [startPartition, endPartition]
