From e1652de1198da66a7ee32241e63e137dd70c4cda Mon Sep 17 00:00:00 2001
From: Guozhang Wang <wangguoz@gmail.com>
Date: Fri, 18 Jan 2019 11:36:55 -0800
Subject: [PATCH] MINOR: Remove the InvalidTopicException handling in
 InternalTopicManager (#6167)

Note we can only remove this handling in 2.2 but not in 2.1 since #6124 is only in 2.2.

Reviewers: Bill Bejeck <bill@confluent.io>, Damian Guy <damian@confluent.io>, Matthias J. Sax <matthias@confluent.io>
---
 .../streams/processor/internals/InternalTopicManager.java    | 5 +----
 1 file changed, 1 insertion(+), 4 deletions(-)

diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java
index ee7fd3e2cb6..f9fe0423482 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java
@@ -23,7 +23,6 @@ import org.apache.kafka.clients.admin.DescribeTopicsResult;
 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.errors.InvalidTopicException;
 import org.apache.kafka.common.errors.LeaderNotAvailableException;
 import org.apache.kafka.common.errors.TopicExistsException;
 import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
@@ -204,9 +203,7 @@ public class InternalTopicManager {
             } catch (final ExecutionException couldNotDescribeTopicException) {
                 final Throwable cause = couldNotDescribeTopicException.getCause();
                 if (cause instanceof UnknownTopicOrPartitionException ||
-                    cause instanceof LeaderNotAvailableException ||
-                    (cause instanceof InvalidTopicException &&
-                        cause.getMessage().equals("Topic " + topicName + " not found."))) {
+                    cause instanceof LeaderNotAvailableException) {
                     // This topic didn't exist or leader is not known yet, proceed to try to create it
                     log.debug("Topic {} is unknown or not found, hence not existed yet.", topicName);
                 } else {