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;