From 2d4e9a113b6701dade3b4fe7ed80afa6cf79bdba Mon Sep 17 00:00:00 2001
From: Soby Chacko <schacko@pivotal.io>
Date: Mon, 5 Nov 2018 16:17:51 -0500
Subject: [PATCH] Use TopicExistsException in provisioner

Instead of checking for the text TopicExistsException in the exception
message, use strong type check for TopicExistsException through instanceof on
the cause of the exception.

Adding test to verify.

Resolves #209
---
 .../provisioning/KafkaTopicProvisioner.java     |  4 ++--
 .../stream/binder/kafka/KafkaBinderTests.java   | 17 ++++++++++++++++-
 2 files changed, 18 insertions(+), 3 deletions(-)

diff --git a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java
index 3ac68c4b1..d4e1d0396 100644
--- a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java
+++ b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java
@@ -39,6 +39,7 @@ import org.apache.kafka.clients.admin.NewTopic;
 import org.apache.kafka.clients.admin.TopicDescription;
 import org.apache.kafka.common.KafkaFuture;
 import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.errors.TopicExistsException;
 
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
@@ -370,8 +371,7 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 				}
 				catch (Exception ex) {
 					if (ex instanceof ExecutionException) {
-						String exceptionMessage = ex.getMessage();
-						if (exceptionMessage.contains("org.apache.kafka.common.errors.TopicExistsException")) {
+						if (ex.getCause() instanceof TopicExistsException) {
 							if (this.logger.isWarnEnabled()) {
 								this.logger.warn("Attempt to create topic: " + topicName + ". Topic already exists.");
 							}
diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java
index 2befeb05f..a79d98441 100644
--- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java
+++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java
@@ -35,7 +35,6 @@ import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
-
 import org.apache.kafka.clients.admin.AdminClient;
 import org.apache.kafka.clients.admin.AdminClientConfig;
 import org.apache.kafka.clients.admin.CreateTopicsResult;
@@ -50,6 +49,7 @@ import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.common.KafkaFuture;
+import org.apache.kafka.common.errors.TopicExistsException;
 import org.apache.kafka.common.record.TimestampType;
 import org.apache.kafka.common.serialization.ByteArrayDeserializer;
 import org.apache.kafka.common.serialization.ByteArraySerializer;
@@ -2649,6 +2649,21 @@ public class KafkaBinderTests extends
 		}
 	}
 
+	@Test (expected = TopicExistsException.class)
+	public void testSameTopicCannotBeProvisionedAgain() throws Throwable {
+		try (AdminClient admin = AdminClient.create(Collections.singletonMap(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,
+				embeddedKafka.getEmbeddedKafka().getBrokersAsString()))) {
+			admin.createTopics(Collections.singletonList(new NewTopic("fooUniqueTopic", 1, (short) 1))).all().get();
+			try {
+				admin.createTopics(Collections.singletonList(new NewTopic("fooUniqueTopic", 1, (short) 1))).all().get();
+			}
+			catch (Exception ex) {
+				assertThat(ex.getCause() instanceof TopicExistsException).isTrue();
+				throw ex.getCause();
+			}
+		}
+	}
+
 	private final class FailingInvocationCountingMessageHandler implements MessageHandler {
 
 		private int invocationCount;