From 776385fbc91369b731e607a4c23153385f86fa24 Mon Sep 17 00:00:00 2001 From: Arabelle Hou Date: Thu, 15 Nov 2018 14:25:31 -0800 Subject: [PATCH] KAFKA-7633: Allow Kafka Connect to access internal topics without cluster ACLs When Kafka Connect does not have cluster ACLs to create topics, it fails to even access its internal topics which already exist. This was originally fixed in KAFKA-6250 by ignoring the cluster authorization error, but now Kafka 2.0 returns a different response code that corresponds to a different error. Add a patch to ignore this new error as well. --- .../apache/kafka/connect/util/TopicAdmin.java | 7 +++++++ .../kafka/connect/util/TopicAdminTest.java | 16 ++++++++++++++++ 2 files changed, 23 insertions(+) diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/util/TopicAdmin.java b/connect/runtime/src/main/java/org/apache/kafka/connect/util/TopicAdmin.java index ad21561baf259..72a5981217231 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/util/TopicAdmin.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/util/TopicAdmin.java @@ -22,6 +22,7 @@ import org.apache.kafka.clients.admin.NewTopic; import org.apache.kafka.common.KafkaFuture; import org.apache.kafka.common.errors.ClusterAuthorizationException; +import org.apache.kafka.common.errors.TopicAuthorizationException; import org.apache.kafka.common.errors.TimeoutException; import org.apache.kafka.common.errors.TopicExistsException; import org.apache.kafka.common.errors.UnsupportedVersionException; @@ -246,6 +247,12 @@ public Set createTopics(NewTopic... topics) { topicNameList, bootstrapServers); return Collections.emptySet(); } + if (cause instanceof TopicAuthorizationException) { + log.debug("Not authorized to create topic(s) '{}'." + + " Falling back to assume topic(s) exist or will be auto-created by the broker.", + topicNameList, bootstrapServers); + return Collections.emptySet(); + } if (cause instanceof TimeoutException) { // Timed out waiting for the operation to complete throw new ConnectException("Timed out while checking for or creating topic(s) '" + topicNameList + "'." + diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/util/TopicAdminTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/util/TopicAdminTest.java index 26c8e71f987d8..ad935dd29cddd 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/util/TopicAdminTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/util/TopicAdminTest.java @@ -71,6 +71,18 @@ public void returnNullWithClusterAuthorizationFailure() { } } + @Test + public void returnNullWithTopicAuthorizationFailure() { + final NewTopic newTopic = TopicAdmin.defineTopic("myTopic").partitions(1).compacted().build(); + Cluster cluster = createCluster(1); + try (AdminClientUnitTestEnv env = new AdminClientUnitTestEnv(new MockTime(), cluster)) { + env.kafkaClient().prepareResponse(createTopicResponseWithTopicAuthorizationException(newTopic)); + TopicAdmin admin = new TopicAdmin(null, env.adminClient()); + boolean created = admin.createTopic(newTopic); + assertFalse(created); + } + } + @Test public void shouldNotCreateTopicWhenItAlreadyExists() { NewTopic newTopic = TopicAdmin.defineTopic("myTopic").partitions(1).compacted().build(); @@ -135,6 +147,10 @@ private CreateTopicsResponse createTopicResponseWithClusterAuthorizationExceptio return createTopicResponse(new ApiError(Errors.CLUSTER_AUTHORIZATION_FAILED, "Not authorized to create topic(s)"), topics); } + private CreateTopicsResponse createTopicResponseWithTopicAuthorizationException(NewTopic... topics) { + return createTopicResponse(new ApiError(Errors.TOPIC_AUTHORIZATION_FAILED, "Not authorized to create topic(s)"), topics); + } + private CreateTopicsResponse createTopicResponse(ApiError error, NewTopic... topics) { if (error == null) error = new ApiError(Errors.NONE, ""); Map topicResults = new HashMap<>();