Browse Source

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>
pull/6172/head
Guozhang Wang 6 years ago committed by GitHub
parent
commit
e1652de119
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
  1. 5
      streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java

5
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.NewTopic;
import org.apache.kafka.clients.admin.TopicDescription; import org.apache.kafka.clients.admin.TopicDescription;
import org.apache.kafka.common.KafkaFuture; 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.LeaderNotAvailableException;
import org.apache.kafka.common.errors.TopicExistsException; import org.apache.kafka.common.errors.TopicExistsException;
import org.apache.kafka.common.errors.UnknownTopicOrPartitionException; import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
@ -204,9 +203,7 @@ public class InternalTopicManager {
} catch (final ExecutionException couldNotDescribeTopicException) { } catch (final ExecutionException couldNotDescribeTopicException) {
final Throwable cause = couldNotDescribeTopicException.getCause(); final Throwable cause = couldNotDescribeTopicException.getCause();
if (cause instanceof UnknownTopicOrPartitionException || if (cause instanceof UnknownTopicOrPartitionException ||
cause instanceof LeaderNotAvailableException || cause instanceof LeaderNotAvailableException) {
(cause instanceof InvalidTopicException &&
cause.getMessage().equals("Topic " + topicName + " not found."))) {
// This topic didn't exist or leader is not known yet, proceed to try to create it // 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); log.debug("Topic {} is unknown or not found, hence not existed yet.", topicName);
} else { } else {

Loading…
Cancel
Save