GH-942: KafkaAdmin Improvements

Resolves #942

Suppress topic/partition already exist exceptions - possible race when multiple
instances of the same app start concurrently.

Author: garyrussell

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2018 the original author or authors.
+ * Copyright 2017-2019 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -36,6 +36,8 @@
 import org.apache.kafka.clients.admin.NewPartitions;
 import org.apache.kafka.clients.admin.NewTopic;
 import org.apache.kafka.clients.admin.TopicDescription;
+import org.apache.kafka.common.errors.InvalidPartitionsException;
+import org.apache.kafka.common.errors.TopicExistsException;
 import org.apache.kafka.common.errors.UnsupportedVersionException;
 
 import org.springframework.beans.BeansException;
@@ -206,6 +208,7 @@ private void addTopicsIfNeeded(AdminClient adminClient, Collection<NewTopic> top
 
 	private Map<String, NewPartitions> checkPartitions(Map<String, NewTopic> topicNameToTopic,
 			DescribeTopicsResult topicInfo, List<NewTopic> topicsToAdd) {
+
 		Map<String, NewPartitions> topicsToModify = new HashMap<>();
 		topicInfo.values().forEach((n, f) -> {
 			NewTopic topic = topicNameToTopic.get(n);
@@ -254,8 +257,13 @@ private void addTopics(AdminClient adminClient, List<NewTopic> topicsToAdd) {
 			throw new KafkaException("Timed out waiting for create topics results", e);
 		}
 		catch (ExecutionException e) {
-			logger.error("Failed to create topics", e.getCause());
-			throw new KafkaException("Failed to create topics", e.getCause()); // NOSONAR
+			if (e.getCause() instanceof TopicExistsException) { // Possible race with another app instance
+				logger.debug("Failed to create topics", e.getCause());
+			}
+			else {
+				logger.error("Failed to create topics", e.getCause());
+				throw new KafkaException("Failed to create topics", e.getCause()); // NOSONAR
+			}
 		}
 	}
 
@@ -272,9 +280,14 @@ private void modifyTopics(AdminClient adminClient, Map<String, NewPartitions> to
 			throw new KafkaException("Timed out waiting for create partitions results", e);
 		}
 		catch (ExecutionException e) {
-			logger.error("Failed to create partitions", e.getCause());
-			if (!(e.getCause() instanceof UnsupportedVersionException)) {
-				throw new KafkaException("Failed to create partitions", e.getCause()); // NOSONAR
+			if (e.getCause() instanceof InvalidPartitionsException) { // Possible race with another app instance
+				logger.debug("Failed to create partitions", e.getCause());
+			}
+			else {
+				logger.error("Failed to create partitions", e.getCause());
+				if (!(e.getCause() instanceof UnsupportedVersionException)) {
+					throw new KafkaException("Failed to create partitions", e.getCause()); // NOSONAR
+				}
 			}
 		}
 	}

spring-kafka/src/test/java/org/springframework/kafka/core/KafkaAdminTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2018 the original author or authors.
+ * Copyright 2017-2019 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -18,14 +18,18 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 
+import java.lang.reflect.Method;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
 
 import org.apache.kafka.clients.admin.AdminClient;
 import org.apache.kafka.clients.admin.AdminClientConfig;
 import org.apache.kafka.clients.admin.DescribeTopicsResult;
+import org.apache.kafka.clients.admin.NewPartitions;
 import org.apache.kafka.clients.admin.NewTopic;
 import org.apache.kafka.clients.admin.TopicDescription;
 import org.junit.Test;
@@ -38,6 +42,7 @@
 import org.springframework.kafka.test.EmbeddedKafkaBroker;
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.junit4.SpringRunner;
+import org.springframework.util.ReflectionUtils;
 import org.springframework.util.StringUtils;
 
 /**
@@ -72,6 +77,28 @@ public void testAddTopics() throws Exception {
 		adminClient.close(10, TimeUnit.SECONDS);
 	}
 
+	@Test
+	public void alreadyExists() throws Exception {
+		AtomicReference<Method> addTopics = new AtomicReference<>();
+		AtomicReference<Method> modifyTopics = new AtomicReference<>();
+		ReflectionUtils.doWithMethods(KafkaAdmin.class, m -> {
+			m.setAccessible(true);
+			if (m.getName().equals("addTopics")) {
+				addTopics.set(m);
+			}
+			else if (m.getName().equals("modifyTopics")) {
+				modifyTopics.set(m);
+			}
+		}, m -> {
+			return m.getName().endsWith("Topics");
+		});
+		try (AdminClient adminClient = AdminClient.create(this.admin.getConfig())) {
+			addTopics.get().invoke(this.admin, adminClient, Collections.singletonList(this.topic1));
+			modifyTopics.get().invoke(this.admin, adminClient, Collections.singletonMap(
+					this.topic1.name(), NewPartitions.increaseTo(this.topic1.numPartitions())));
+		}
+	}
+
 	@Configuration
 	public static class Config {