KAFKA-20430 Add configuration to control internal topic creation in Kafka Connect

bin/connect-internal-topics.sh

@@ -0,0 +1,21 @@
+#!/bin/bash
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
+  export KAFKA_HEAP_OPTS="-Xms256M -Xmx2G"
+fi
+
+exec $(dirname $0)/kafka-run-class.sh org.apache.kafka.tools.ConnectInternalTopics "$@"

bin/windows/connect-internal-topics.bat

@@ -0,0 +1,21 @@
+@echo off
+rem Licensed to the Apache Software Foundation (ASF) under one or more
+rem contributor license agreements.  See the NOTICE file distributed with
+rem this work for additional information regarding copyright ownership.
+rem The ASF licenses this file to You under the Apache License, Version 2.0
+rem (the "License"); you may not use this file except in compliance with
+rem the License.  You may obtain a copy of the License at
+rem
+rem     http://www.apache.org/licenses/LICENSE-2.0
+rem
+rem Unless required by applicable law or agreed to in writing, software
+rem distributed under the License is distributed on an "AS IS" BASIS,
+rem WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+rem See the License for the specific language governing permissions and
+rem limitations under the License.
+
+IF ["%KAFKA_HEAP_OPTS%"] EQU [""] (
+	set KAFKA_HEAP_OPTS=-Xms256M -Xmx2G
+)
+
+"%~dp0kafka-run-class.bat" org.apache.kafka.tools.ConnectInternalTopics %*

checkstyle/import-control.xml

@@ -308,6 +308,7 @@
     <allow pkg="org.apache.kafka.test" />
     <allow pkg="org.apache.kafka.connect.runtime" />
     <allow pkg="org.apache.kafka.connect.runtime.isolation" />
+    <allow pkg="org.apache.kafka.connect.util" />
     <allow pkg="com.fasterxml.jackson" />
     <allow pkg="org.jose4j" />
     <allow pkg="net.sourceforge.argparse4j" />

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerConfig.java

@@ -410,6 +410,18 @@ public boolean connectorOffsetsTopicsPermitted() {
         return false;
     }
 
+    /**
+     * Determine whether this worker should automatically create internal topics used by Connect
+     * (such as the offset, config, and status topics).
+     * The default implementation returns {@code true}. Subclasses may override this method
+     * to respect user-provided configuration.
+     *
+     * @return whether the worker should automatically create internal topics
+     */
+    public boolean internalTopicsCreationEnabled() {
+        return true;
+    }
+
     /**
      * @return the offset commit interval for tasks created by this worker
      */

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedConfig.java

@@ -20,6 +20,7 @@
 import org.apache.kafka.clients.MetadataRecoveryStrategy;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.config.ConfigDef.Type;
 import org.apache.kafka.common.config.ConfigException;
 import org.apache.kafka.common.config.TopicConfig;
 import org.apache.kafka.common.security.auth.SecurityProtocol;
@@ -187,6 +188,13 @@ public final class DistributedConfig extends WorkerConfig {
     public static final String CONNECT_PROTOCOL_DOC = "Compatibility mode for Kafka Connect Protocol";
     public static final String CONNECT_PROTOCOL_DEFAULT = ConnectProtocolCompatibility.SESSIONED.toString();
 
+
+    public static final String INTERNAL_TOPICS_AUTOMATIC_CREATION_ENABLE_CONFIG = "internal.topics.automatic.creation.enable";
+    private static final String INTERNAL_TOPICS_AUTOMATIC_CREATION_ENABLE_DOC = "Whether to automatically create internal topics used by Connect. "
+            + "This includes the offset, config, and status topics, as well as connector-specific offset topics "
+            + "configured via 'offsets.storage.topic' in the source connector configuration.";
+    public static final boolean INTERNAL_TOPICS_AUTOMATIC_CREATION_ENABLE_DEFAULT = true;
+
     /**
      * <code>scheduled.rebalance.max.delay.ms</code>
      */
@@ -427,6 +435,12 @@ private static ConfigDef config(Crypto crypto) {
                     WORKER_UNSYNC_BACKOFF_MS_DEFAULT,
                     ConfigDef.Importance.MEDIUM,
                     WORKER_UNSYNC_BACKOFF_MS_DOC)
+            .define(
+                    INTERNAL_TOPICS_AUTOMATIC_CREATION_ENABLE_CONFIG,
+                    Type.BOOLEAN,
+                    INTERNAL_TOPICS_AUTOMATIC_CREATION_ENABLE_DEFAULT,
+                    ConfigDef.Importance.MEDIUM,
+                    INTERNAL_TOPICS_AUTOMATIC_CREATION_ENABLE_DOC)
             .define(OFFSET_STORAGE_TOPIC_CONFIG,
                     ConfigDef.Type.STRING,
                     ConfigDef.Importance.HIGH,
@@ -590,6 +604,11 @@ public boolean connectorOffsetsTopicsPermitted() {
         return true;
     }
 
+    @Override
+    public boolean internalTopicsCreationEnabled() {
+        return Boolean.TRUE.equals(getBoolean(INTERNAL_TOPICS_AUTOMATIC_CREATION_ENABLE_CONFIG));
+    }
+
     @Override
     public String groupId() {
         return getString(GROUP_ID_CONFIG);

connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaTopicBasedBackingStore.java

@@ -18,10 +18,13 @@
 
 import org.apache.kafka.clients.admin.AdminClientConfig;
 import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.admin.TopicDescription;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.common.config.TopicConfig;
 import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.connect.errors.ConnectException;
 import org.apache.kafka.connect.runtime.WorkerConfig;
+import org.apache.kafka.connect.runtime.distributed.DistributedConfig;
 import org.apache.kafka.connect.util.Callback;
 import org.apache.kafka.connect.util.KafkaBasedLog;
 import org.apache.kafka.connect.util.TopicAdmin;
@@ -39,17 +42,35 @@ public abstract class KafkaTopicBasedBackingStore {
 
     Consumer<TopicAdmin> topicInitializer(String topic, NewTopic topicDescription, WorkerConfig config, Time time) {
         return admin -> {
-            log.debug("Creating Connect internal topic for {}", getTopicPurpose());
-            // Create the topic if it doesn't exist
-            Set<String> newTopics = createTopics(topicDescription, admin, config, time);
-            if (!newTopics.contains(topic)) {
-                // It already existed, so check that the topic cleanup policy is compact only and not delete
-                log.debug("Using admin client to check cleanup policy of '{}' topic is '{}'", topic, TopicConfig.CLEANUP_POLICY_COMPACT);
-                admin.verifyTopicCleanupPolicyOnlyCompact(topic, getTopicConfig(), getTopicPurpose());
+            if (config.internalTopicsCreationEnabled()) {
+                log.debug("Creating Connect internal topic for {}", getTopicPurpose());
+                // Create the topic if it doesn't exist
+                Set<String> newTopics = createTopics(topicDescription, admin, config, time);
+                if (!newTopics.contains(topic)) {
+                    verifyTopicConfig(topic, admin);
+                }
+            } else {
+                log.debug("Skipping creation of Connect internal topic for {} because automatic topic creation is disabled", getTopicPurpose());
+                Map<String, TopicDescription> existing = admin.describeTopics(topic);
+                if (existing.isEmpty()) {
+                    String msg = String.format("Topic '%s' specified via the '%s' property is missing." +
+                                    " The config '%s' is set to '%s', so automatic creation of internal topics is disabled." +
+                                    " Either enable automatic creation or create the topics manually before starting the worker.",
+                            topic, getTopicConfig(), DistributedConfig.INTERNAL_TOPICS_AUTOMATIC_CREATION_ENABLE_CONFIG, config.internalTopicsCreationEnabled());
+                    throw new ConnectException(msg);
+                }
+
+                verifyTopicConfig(topic, admin);
             }
         };
     }
 
+    private void verifyTopicConfig(String topic, TopicAdmin admin) {
+        // It already existed, so check that the topic cleanup policy is compact only and not delete
+        log.debug("Using admin client to check cleanup policy of '{}' topic is '{}'", topic, TopicConfig.CLEANUP_POLICY_COMPACT);
+        admin.verifyTopicCleanupPolicyOnlyCompact(topic, getTopicConfig(), getTopicPurpose());
+    }
+
     private Set<String> createTopics(NewTopic topicDescription, TopicAdmin admin, WorkerConfig config, Time time) {
         // get the prefixless default api timeout and retry backoff for topic creation retry configs
         AdminClientConfig adminClientConfig = new AdminClientConfig(config.originals());

connect/runtime/src/test/java/org/apache/kafka/connect/integration/InternalTopicsIntegrationTest.java

@@ -249,6 +249,34 @@ public void testFailToStartWhenInternalTopicsAreNotCompacted() throws Interrupte
         connect.assertions().assertAtLeastNumWorkersAreUp(1, "Worker did not start in time.");
     }
 
+    @Test
+    void testFailToStartWhenInternalTopicsAreMissingWithDisabledInternalTopicCreation() throws InterruptedException {
+        workerProps.put(DistributedConfig.CONFIG_TOPIC_CONFIG, "non-existent-config");
+        workerProps.put(DistributedConfig.OFFSET_STORAGE_TOPIC_CONFIG, "non-existent-offset");
+        workerProps.put(DistributedConfig.STATUS_STORAGE_TOPIC_CONFIG, "non-existent-status");
+        workerProps.put(DistributedConfig.CONFIG_STORAGE_REPLICATION_FACTOR_CONFIG, "1");
+        workerProps.put(DistributedConfig.OFFSET_STORAGE_REPLICATION_FACTOR_CONFIG, "1");
+        workerProps.put(DistributedConfig.STATUS_STORAGE_REPLICATION_FACTOR_CONFIG, "1");
+        workerProps.put(DistributedConfig.INTERNAL_TOPICS_AUTOMATIC_CREATION_ENABLE_CONFIG, "false");
+        connect = new EmbeddedConnectCluster.Builder().name("connect-cluster-1")
+                .workerProps(workerProps)
+                .numWorkers(0)
+                .numBrokers(1)
+                .brokerProps(brokerProps)
+                .build();
+        connect.start();
+
+        var worker = connect.addWorker();
+
+        assertFalse(connect.anyWorkersHealthy());
+        var herderTask = worker.herderTask();
+        assertThrows(
+                ExecutionException.class,
+                () -> herderTask.get(1, TimeUnit.MINUTES)
+        );
+        connect.assertions().assertTopicsDoNotExist("non-existent-config", "non-existent-offset", "non-existent-status");
+    }
+
     @Test
     public void testStartWhenInternalTopicsCreatedManuallyWithCompactForBrokersDefaultCleanupPolicy() throws InterruptedException {
         // Change the broker default cleanup policy to compact, which is good for Connect

docs/getting-started/upgrade.md

@@ -49,6 +49,7 @@ type: docs
   * Brokers can now record a human-readable description of each streams group's processing topology via a pluggable backend, retrievable through `Admin#describeStreamsGroups` and `kafka-streams-groups.sh --describe --topology`. The feature is disabled unless the new broker configuration `group.streams.topology.description.plugin.class` is set to a `StreamsGroupTopologyDescriptionPlugin` implementation; on the client side, the new Kafka Streams configuration `topology.description.push.enabled` (default `true`) controls whether the client pushes topology descriptions when requested. This adds a new RPC, `StreamsGroupTopologyDescriptionUpdate`, bumps `StreamsGroupDescribe` and `StreamsGroupHeartbeat` to version 1, and introduces the error codes `GROUP_DELETION_FAILED` (134) and `STREAMS_TOPOLOGY_DESCRIPTION_UPDATE_FAILED` (135). `DeleteGroups` is bumped to version 3, adding a per-group `ErrorMessage` field so brokers can report why a group deletion failed (for example, when the plugin fails to delete its stored topology description, the group is not deleted and `GROUP_DELETION_FAILED` is returned; retrying the deletion is safe). For further details, please refer to [KIP-1331](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1331%3A+Streams+Group+Topology+Description+Plugin) and the [Topology Description Plugin](/{version}/streams/developer-guide/topology-description-plugin/) documentation.
   * The `kafka-producer-perf-test.sh` tool now supports `--record-key-range`, `--key-distribution`, and `--random-seed` options to control the distribution of record keys. Use `--key-distribution range` for sequential key assignment (round-robin over the key range) or `--key-distribution random` for random key selection. The `--random-seed` option allows reproducible benchmark runs when using random key distribution. For further details, please refer to [KIP-1299](https://cwiki.apache.org/confluence/x/XpQ8G).
   * Share groups now support dead-letter queue functionality as outlined in [KIP-1191](https://cwiki.apache.org/confluence/x/fApJFg). Any records which are released (beyond max delivery count) or rejected by the share consumer become eligible for DLQ. Share group DLQ gets enabled when the Kafka feature `share.version` is upgraded to 2. The user can configure a DLQ topic on a share group by setting the dynamic config `errors.deadletterqueue.topic.name` (default `""`) to the name of the DLQ topic. The cluster can be configured to auto create the DLQ topics by setting the dynamic cluster config `errors.deadletterqueue.auto.create.topics.enable` to `true` (default `false`). If auto create is not enabled, the user must create the DLQ topic like a standard Kafka topic and set the dynamic config `errors.deadletterqueue.group.enable` to `true` on the DLQ topic. The DLQ topic name must be prefixed by the value set in the dynamic cluster config `errors.deadletterqueue.topic.name.prefix` (default `dlq.`). The records sent to the DLQ topic by default only contain source record metadata like group, topic name, partition id, offset and delivery count. If original record data is also required, the user must set the dynamic config `errors.deadletterqueue.copy.record.enable` to `true` on the share group.
+  * Kafka Connect distributed workers now support the `internal.topics.automatic.creation.enable` configuration (default: `true`). When set to `false`, Connect will not automatically create internal topics (offset, config, status, and connector-specific offset topics) and will instead fail at startup if any of these topics are missing. A new `connect-internal-topics.sh` tool is also available for manually creating these topics. For further details, please refer to [KIP-1209](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1209:+Add+configuration+to+control+internal+topic+creation+in+Kafka+Connect).
 
 ## Upgrading to 4.3.0