This is an automated email from the ASF dual-hosted git repository.
chia7712 pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new b992620602d KAFKA-20430 Add configuration to control internal topic
creation in Kafka Connect (#20384)
b992620602d is described below
commit b992620602d72f633dcdeabed498638c899e87f8
Author: Anton Liauchuk <[email protected]>
AuthorDate: Wed Jul 29 06:04:19 2026 +0300
KAFKA-20430 Add configuration to control internal topic creation in Kafka
Connect (#20384)
Implementation of
https://cwiki.apache.org/confluence/display/KAFKA/KIP-1209:+Add+configuration+to+control+internal+topic+creation+in+Kafka+Connect
Reviewers: Ken Huang <[email protected]>, Chia-Ping Tsai
<[email protected]>
---
bin/connect-internal-topics.sh | 21 ++
bin/windows/connect-internal-topics.bat | 21 ++
checkstyle/import-control.xml | 1 +
.../apache/kafka/connect/runtime/WorkerConfig.java | 12 +
.../runtime/distributed/DistributedConfig.java | 19 ++
.../storage/KafkaTopicBasedBackingStore.java | 35 ++-
.../integration/InternalTopicsIntegrationTest.java | 28 +++
docs/getting-started/upgrade.md | 1 +
.../apache/kafka/tools/ConnectInternalTopics.java | 246 +++++++++++++++++++++
.../kafka/tools/ConnectInternalTopicsTest.java | 215 ++++++++++++++++++
10 files changed, 592 insertions(+), 7 deletions(-)
diff --git a/bin/connect-internal-topics.sh b/bin/connect-internal-topics.sh
new file mode 100755
index 00000000000..3905c165c8b
--- /dev/null
+++ b/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 "$@"
diff --git a/bin/windows/connect-internal-topics.bat
b/bin/windows/connect-internal-topics.bat
new file mode 100644
index 00000000000..c4d9f01225f
--- /dev/null
+++ b/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 %*
\ No newline at end of file
diff --git a/checkstyle/import-control.xml b/checkstyle/import-control.xml
index 3a0b41cabd1..3ca56f078c7 100644
--- a/checkstyle/import-control.xml
+++ b/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" />
diff --git
a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerConfig.java
b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerConfig.java
index 0dacbb909c3..dbc05c78563 100644
---
a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerConfig.java
+++
b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerConfig.java
@@ -410,6 +410,18 @@ public class WorkerConfig extends AbstractConfig {
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
*/
diff --git
a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedConfig.java
b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedConfig.java
index dd4ea3889d2..83a862c27a7 100644
---
a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedConfig.java
+++
b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedConfig.java
@@ -20,6 +20,7 @@ import org.apache.kafka.clients.CommonClientConfigs;
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>
*/
@@ -433,6 +441,12 @@ public final class DistributedConfig extends WorkerConfig {
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,
@@ -596,6 +610,11 @@ public final class DistributedConfig extends WorkerConfig {
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);
diff --git
a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaTopicBasedBackingStore.java
b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaTopicBasedBackingStore.java
index 4678311e4c7..f4215d95486 100644
---
a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaTopicBasedBackingStore.java
+++
b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaTopicBasedBackingStore.java
@@ -18,10 +18,13 @@ package org.apache.kafka.connect.storage;
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());
diff --git
a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/InternalTopicsIntegrationTest.java
b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/InternalTopicsIntegrationTest.java
index 50871fa7257..eb88ecb432c 100644
---
a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/InternalTopicsIntegrationTest.java
+++
b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/InternalTopicsIntegrationTest.java
@@ -249,6 +249,34 @@ public class InternalTopicsIntegrationTest {
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
diff --git a/docs/getting-started/upgrade.md b/docs/getting-started/upgrade.md
index c504d0c1630..034d157a68a 100644
--- a/docs/getting-started/upgrade.md
+++ b/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.ena [...]
* 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.or [...]
* 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 DL [...]
+ * 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://cwik [...]
## Upgrading to 4.3.0
diff --git
a/tools/src/main/java/org/apache/kafka/tools/ConnectInternalTopics.java
b/tools/src/main/java/org/apache/kafka/tools/ConnectInternalTopics.java
new file mode 100644
index 00000000000..a7251aaa05b
--- /dev/null
+++ b/tools/src/main/java/org/apache/kafka/tools/ConnectInternalTopics.java
@@ -0,0 +1,246 @@
+/*
+ * 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.
+ */
+package org.apache.kafka.tools;
+
+import org.apache.kafka.common.config.AbstractConfig;
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.config.ConfigException;
+import org.apache.kafka.common.config.TopicConfig;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.common.utils.internals.Exit;
+import org.apache.kafka.connect.runtime.distributed.DistributedConfig;
+import org.apache.kafka.connect.util.SharedTopicAdmin;
+import org.apache.kafka.connect.util.TopicAdmin;
+
+import net.sourceforge.argparse4j.ArgumentParsers;
+import net.sourceforge.argparse4j.inf.ArgumentParser;
+import net.sourceforge.argparse4j.inf.ArgumentParserException;
+import net.sourceforge.argparse4j.inf.Namespace;
+
+import java.io.IOException;
+import java.io.PrintStream;
+import java.util.HashMap;
+import java.util.Map;
+
+import static net.sourceforge.argparse4j.impl.Arguments.store;
+
+public class ConnectInternalTopics {
+
+ private static final String CREATE_COMMAND = "create";
+
+ public static void main(String[] args) {
+ Exit.exit(mainNoExit(args, System.out, System.err));
+ }
+
+ static int mainNoExit(String[] args, PrintStream out, PrintStream err) {
+ var parser = parser();
+ try {
+ var namespace = parser.parseArgs(args);
+ var workerProperties = parseWorkerProperties(parser, namespace);
+ out.println("Parsed arguments and loaded worker properties");
+ execute(parser, namespace, workerProperties, out);
+ out.println("Command executed successfully");
+ return 0;
+ } catch (ArgumentParserException e) {
+ parser.handleError(e);
+ return 1;
+ } catch (TerseException | ConfigException e) {
+ err.println(e.getMessage());
+ return 2;
+ } catch (Throwable e) {
+ err.println("Unexpected error: " + e.getMessage());
+ err.println(Utils.stackTrace(e));
+ return 3;
+ }
+ }
+
+ private static void execute(ArgumentParser parser, Namespace namespace,
Map<String, String> workerProperties, PrintStream out) throws
ArgumentParserException {
+ var subcommand = namespace.getString("subcommand");
+ out.println("Subcommand: " + subcommand);
+ if (subcommand == null) {
+ throw new ArgumentParserException("No subcommand specified",
parser);
+ }
+ if (CREATE_COMMAND.equals(subcommand)) {
+ var internalTopicsConfig = new
InternalTopicsConfig(workerProperties);
+ internalTopicsConfig.validateTopicNames();
+ out.println("Running create command for internal topics");
+ runCommand(internalTopicsConfig, out);
+ } else {
+ throw new ArgumentParserException("Unrecognized subcommand: '" +
subcommand + "'", parser);
+ }
+ }
+
+ private static void runCommand(InternalTopicsConfig config, PrintStream
out) {
+ var adminProps = new HashMap<>(config.originals());
+ out.println("Admin properties loaded for topic admin");
+ try (var sharedAdmin = new SharedTopicAdmin(adminProps)) {
+ createInternalTopic(sharedAdmin, buildOffsetTopicSettings(config,
out), out);
+ createInternalTopic(sharedAdmin, buildConfigTopicSettings(config,
out), out);
+ createInternalTopic(sharedAdmin, buildStatusTopicSettings(config,
out), out);
+ }
+ }
+
+ private static void createInternalTopic(SharedTopicAdmin sharedAdmin,
TopicSettings settings, PrintStream out) {
+ var topicDescription = TopicAdmin.defineTopic(settings.topicName)
+ .config(settings.topicSettings)
+ .compacted()
+ .partitions(settings.partitions)
+ .replicationFactor(settings.replicationFactor)
+ .build();
+ var created = sharedAdmin.topicAdmin().createTopics(topicDescription);
+ if (created.contains(settings.topicName)) {
+ out.println("Created internal topic: " + settings.topicName);
+ } else {
+ out.println("Internal topic already exists: " +
settings.topicName);
+ }
+ }
+
+ private static TopicSettings buildOffsetTopicSettings(InternalTopicsConfig
config, PrintStream out) {
+ return new TopicSettings(
+
config.getString(DistributedConfig.OFFSET_STORAGE_TOPIC_CONFIG),
+ config.topicSettings(DistributedConfig.OFFSET_STORAGE_PREFIX,
out),
+
config.getInt(DistributedConfig.OFFSET_STORAGE_PARTITIONS_CONFIG),
+
config.getShort(DistributedConfig.OFFSET_STORAGE_REPLICATION_FACTOR_CONFIG)
+ );
+ }
+
+ private static TopicSettings buildConfigTopicSettings(InternalTopicsConfig
config, PrintStream out) {
+ return new TopicSettings(
+ config.getString(DistributedConfig.CONFIG_TOPIC_CONFIG),
+ config.topicSettings(DistributedConfig.CONFIG_STORAGE_PREFIX,
out),
+ 1,
+
config.getShort(DistributedConfig.CONFIG_STORAGE_REPLICATION_FACTOR_CONFIG)
+ );
+ }
+
+ private static TopicSettings buildStatusTopicSettings(InternalTopicsConfig
config, PrintStream out) {
+ return new TopicSettings(
+
config.getString(DistributedConfig.STATUS_STORAGE_TOPIC_CONFIG),
+ config.topicSettings(DistributedConfig.STATUS_STORAGE_PREFIX,
out),
+
config.getInt(DistributedConfig.STATUS_STORAGE_PARTITIONS_CONFIG),
+
config.getShort(DistributedConfig.STATUS_STORAGE_REPLICATION_FACTOR_CONFIG)
+ );
+ }
+
+ private record TopicSettings(String topicName, Map<String, Object>
topicSettings, int partitions,
+ short replicationFactor) {
+ }
+
+ private static Map<String, String> parseWorkerProperties(ArgumentParser
parser, Namespace namespace) throws ArgumentParserException, TerseException {
+ String workerConfigPath = namespace.getString("worker_config");
+ if (workerConfigPath == null || workerConfigPath.isBlank()) {
+ throw new ArgumentParserException("--worker-config must be
specified and non-blank", parser);
+ }
+
+ try {
+ return Utils.propsToStringMap(Utils.loadProps(workerConfigPath));
+ } catch (IOException e) {
+ throw new TerseException("Unable to read worker config at " +
workerConfigPath);
+ }
+ }
+
+ private static ArgumentParser parser() {
+ var parser =
ArgumentParsers.newArgumentParser("connect-internal-topics")
+ .defaultHelp(true)
+ .description("Manage internal topics required by Kafka Connect
clusters (config, status, and offset topics).");
+
+ parser.addSubparsers()
+ .description("Create internal topics required for Kafka
Connect operation using the provided worker configuration.")
+ .dest("subcommand")
+ .addParser(CREATE_COMMAND)
+ .addArgument("--worker-config")
+ .setDefault("")
+ .type(String.class)
+ .action(store())
+ .help("Path to a Connect worker configuration file. This file
must define the internal topic names and connection information for the Kafka
cluster.");
+
+ return parser;
+ }
+
+ private static class InternalTopicsConfig extends AbstractConfig {
+ private static final ConfigDef CONFIG_DEF = new ConfigDef()
+ .define(DistributedConfig.OFFSET_STORAGE_TOPIC_CONFIG,
+ ConfigDef.Type.STRING,
+ ConfigDef.Importance.HIGH,
+ "")
+ .define(DistributedConfig.OFFSET_STORAGE_PARTITIONS_CONFIG,
+ ConfigDef.Type.INT,
+ 25,
+ ConfigDef.Importance.LOW,
+ "")
+
.define(DistributedConfig.OFFSET_STORAGE_REPLICATION_FACTOR_CONFIG,
+ ConfigDef.Type.SHORT,
+ (short) 3,
+ ConfigDef.Importance.LOW,
+ "")
+ .define(DistributedConfig.CONFIG_TOPIC_CONFIG,
+ ConfigDef.Type.STRING,
+ ConfigDef.Importance.HIGH,
+ "")
+
.define(DistributedConfig.CONFIG_STORAGE_REPLICATION_FACTOR_CONFIG,
+ ConfigDef.Type.SHORT,
+ (short) 3,
+ ConfigDef.Importance.LOW,
+ "")
+ .define(DistributedConfig.STATUS_STORAGE_TOPIC_CONFIG,
+ ConfigDef.Type.STRING,
+ ConfigDef.Importance.HIGH,
+ "")
+ .define(DistributedConfig.STATUS_STORAGE_PARTITIONS_CONFIG,
+ ConfigDef.Type.INT,
+ 5,
+ ConfigDef.Importance.LOW,
+ "")
+
.define(DistributedConfig.STATUS_STORAGE_REPLICATION_FACTOR_CONFIG,
+ ConfigDef.Type.SHORT,
+ (short) 3,
+ ConfigDef.Importance.LOW,
+ "");
+
+ InternalTopicsConfig(Map<String, String> props) {
+ super(CONFIG_DEF, props, false);
+ }
+
+ void validateTopicNames() {
+ validateTopicName(DistributedConfig.OFFSET_STORAGE_TOPIC_CONFIG);
+ validateTopicName(DistributedConfig.CONFIG_TOPIC_CONFIG);
+ validateTopicName(DistributedConfig.STATUS_STORAGE_TOPIC_CONFIG);
+ }
+
+ private void validateTopicName(String config) {
+ var value = getString(config);
+ if (value == null || value.trim().isEmpty()) {
+ throw new ConfigException("Must specify non-empty value for
required internal topic config: '" + config + "'.");
+ }
+ }
+
+ private Map<String, Object> topicSettings(String prefix, PrintStream
out) {
+ var result = originalsWithPrefix(prefix);
+ if (DistributedConfig.CONFIG_STORAGE_PREFIX.equals(prefix) &&
result.containsKey(DistributedConfig.PARTITIONS_SUFFIX)) {
+ out.println("Ignoring '" + prefix +
DistributedConfig.PARTITIONS_SUFFIX + "=" +
result.get(DistributedConfig.PARTITIONS_SUFFIX) + "' setting, since config
topic partitions is always 1");
+ }
+ var removedPolicy =
result.remove(TopicConfig.CLEANUP_POLICY_CONFIG);
+ if (removedPolicy != null) {
+ out.println("Ignoring '" + prefix + "cleanup.policy=" +
removedPolicy + "' setting, since compaction is always used");
+ }
+ result.remove(DistributedConfig.TOPIC_SUFFIX);
+ result.remove(DistributedConfig.REPLICATION_FACTOR_SUFFIX);
+ result.remove(DistributedConfig.PARTITIONS_SUFFIX);
+ return result;
+ }
+ }
+}
diff --git
a/tools/src/test/java/org/apache/kafka/tools/ConnectInternalTopicsTest.java
b/tools/src/test/java/org/apache/kafka/tools/ConnectInternalTopicsTest.java
new file mode 100644
index 00000000000..fadff2bebbd
--- /dev/null
+++ b/tools/src/test/java/org/apache/kafka/tools/ConnectInternalTopicsTest.java
@@ -0,0 +1,215 @@
+/*
+ * 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.
+ */
+package org.apache.kafka.tools;
+
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.common.config.ConfigResource;
+import org.apache.kafka.common.test.ClusterInstance;
+import org.apache.kafka.common.test.api.ClusterTest;
+import org.apache.kafka.test.TestUtils;
+
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.PrintStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Collections;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class ConnectInternalTopicsTest {
+
+ private static final String CONFIG_TOPIC_NAME = "config";
+ private static final String STATUS_TOPIC_NAME = "status";
+ private static final String OFFSET_TOPIC_NAME = "offset";
+
+ @TempDir
+ Path workspace;
+
+ @ClusterTest(brokers = 3)
+ void testCreateInternalTopicsWithDefaultValues(ClusterInstance cluster)
throws Exception {
+ var properties = new Properties();
+ properties.setProperty("bootstrap.servers",
cluster.bootstrapServers());
+ properties.setProperty("config.storage.topic", CONFIG_TOPIC_NAME);
+ properties.setProperty("status.storage.topic", STATUS_TOPIC_NAME);
+ properties.setProperty("offset.storage.topic", OFFSET_TOPIC_NAME);
+ var workerConfigPath =
setupWorkerConfig(workspace.resolve("worker.properties"), properties);
+ var res = runCommand("create", "--worker-config",
workerConfigPath.toString());
+ assertEquals(0, res.returnCode);
+ try (var adminClient = cluster.admin()) {
+ waitForTopics(adminClient, Set.of(CONFIG_TOPIC_NAME,
STATUS_TOPIC_NAME, OFFSET_TOPIC_NAME));
+ assertTopicPartitions(adminClient, CONFIG_TOPIC_NAME, 1);
+ assertTopicPartitions(adminClient, STATUS_TOPIC_NAME, 5);
+ assertTopicPartitions(adminClient, OFFSET_TOPIC_NAME, 25);
+ assertTopicReplicationFactor(adminClient, CONFIG_TOPIC_NAME,
(short) 3);
+ assertTopicReplicationFactor(adminClient, STATUS_TOPIC_NAME,
(short) 3);
+ assertTopicReplicationFactor(adminClient, OFFSET_TOPIC_NAME,
(short) 3);
+ }
+ }
+
+ @ClusterTest
+ void testNoWorkerConfig() {
+ var res = runCommand("create");
+ assertNotEquals(0, res.returnCode);
+ }
+
+ @ClusterTest
+ void testWorkerConfigBlank() {
+ var res = runCommand("create", "--worker-config", "");
+ assertNotEquals(0, res.returnCode);
+ }
+
+ @ClusterTest
+ void testWorkerConfigFileDoesNotExist() {
+ var nonExistentPath =
workspace.resolve("nonexistent-worker.properties").toString();
+ var res = runCommand("create", "--worker-config", nonExistentPath);
+ assertNotEquals(0, res.returnCode);
+ assertTrue(res.err.contains("Unable to read worker config"));
+ }
+
+ @ClusterTest(brokers = 3)
+ void testNoTopicNamesInWorkerConfig(ClusterInstance cluster) throws
IOException {
+ var properties = new Properties();
+ properties.setProperty("bootstrap.servers",
cluster.bootstrapServers());
+ var configPath =
setupWorkerConfig(workspace.resolve("worker-no-topics.properties"), properties);
+ var res = runCommand("create", "--worker-config",
configPath.toString());
+ assertNotEquals(0, res.returnCode);
+ assertEquals("Missing required configuration \"offset.storage.topic\"
which has no default value.\n", res.err);
+ }
+
+ @ClusterTest(brokers = 3)
+ void testEmptyTopicNamesInWorkerConfig(ClusterInstance cluster) throws
IOException {
+ var properties = new Properties();
+ properties.setProperty("bootstrap.servers",
cluster.bootstrapServers());
+ properties.setProperty("config.storage.topic", "config");
+ properties.setProperty("status.storage.topic", "status");
+ properties.setProperty("offset.storage.topic", "");
+ var configPath =
setupWorkerConfig(workspace.resolve("worker-empty-topics.properties"),
properties);
+ var res = runCommand("create", "--worker-config",
configPath.toString());
+ assertNotEquals(0, res.returnCode);
+ assertEquals("Must specify non-empty value for required internal topic
config: 'offset.storage.topic'.\n", res.err);
+ }
+
+ @ClusterTest(brokers = 3)
+ void testTopicConfigOverrides(ClusterInstance cluster) throws Exception {
+ var properties = new Properties();
+ properties.setProperty("bootstrap.servers",
cluster.bootstrapServers());
+ properties.setProperty("config.storage.topic", CONFIG_TOPIC_NAME);
+ properties.setProperty("config.storage.retention.ms", "1000");
+ properties.setProperty("status.storage.topic", STATUS_TOPIC_NAME);
+ properties.setProperty("status.storage.retention.ms", "2000");
+ properties.setProperty("offset.storage.topic", OFFSET_TOPIC_NAME);
+ properties.setProperty("offset.storage.retention.ms", "3000");
+ var configPath =
setupWorkerConfig(workspace.resolve("worker-topic-overrides.properties"),
properties);
+ var res = runCommand("create", "--worker-config",
configPath.toString());
+ assertEquals(0, res.returnCode);
+ try (var adminClient = cluster.admin()) {
+ waitForTopics(adminClient, Set.of(CONFIG_TOPIC_NAME,
STATUS_TOPIC_NAME, OFFSET_TOPIC_NAME));
+ assertTopicConfig(adminClient, CONFIG_TOPIC_NAME, "retention.ms",
"1000");
+ assertTopicConfig(adminClient, STATUS_TOPIC_NAME, "retention.ms",
"2000");
+ assertTopicConfig(adminClient, OFFSET_TOPIC_NAME, "retention.ms",
"3000");
+ assertTopicReplicationFactor(adminClient, CONFIG_TOPIC_NAME,
(short) 3);
+ assertTopicReplicationFactor(adminClient, STATUS_TOPIC_NAME,
(short) 3);
+ assertTopicReplicationFactor(adminClient, OFFSET_TOPIC_NAME,
(short) 3);
+ }
+ }
+
+ @ClusterTest(brokers = 3)
+ void testCreateMissingTopics(ClusterInstance cluster) throws Exception {
+ try (var adminClient = cluster.admin()) {
+ adminClient.createTopics(Collections.singleton(
+ new NewTopic(CONFIG_TOPIC_NAME, 1, (short) 1)
+ .configs(Map.of("retention.ms", "1000"))
+ ));
+ waitForTopics(adminClient, Set.of(CONFIG_TOPIC_NAME));
+ }
+ var properties = new Properties();
+ properties.setProperty("bootstrap.servers",
cluster.bootstrapServers());
+ properties.setProperty("config.storage.topic", CONFIG_TOPIC_NAME);
+ properties.setProperty("status.storage.topic", STATUS_TOPIC_NAME);
+ properties.setProperty("offset.storage.topic", OFFSET_TOPIC_NAME);
+ var configPath =
setupWorkerConfig(workspace.resolve("worker-partial-topics.properties"),
properties);
+ var res = runCommand("create", "--worker-config",
configPath.toString());
+ assertEquals(0, res.returnCode);
+ try (var adminClient = cluster.admin()) {
+ waitForTopics(adminClient, Set.of(CONFIG_TOPIC_NAME,
STATUS_TOPIC_NAME, OFFSET_TOPIC_NAME));
+ assertTopicConfig(adminClient, CONFIG_TOPIC_NAME, "retention.ms",
"1000");
+ assertTopicPartitions(adminClient, CONFIG_TOPIC_NAME, 1);
+ assertTopicPartitions(adminClient, STATUS_TOPIC_NAME, 5);
+ assertTopicPartitions(adminClient, OFFSET_TOPIC_NAME, 25);
+ assertTopicReplicationFactor(adminClient, CONFIG_TOPIC_NAME,
(short) 1);
+ assertTopicReplicationFactor(adminClient, STATUS_TOPIC_NAME,
(short) 3);
+ assertTopicReplicationFactor(adminClient, OFFSET_TOPIC_NAME,
(short) 3);
+ }
+ }
+
+ private record CommandResult(int returnCode, String out, String err) {
+ }
+
+ private static CommandResult runCommand(String... args) {
+ var out = new ByteArrayOutputStream();
+ var err = new ByteArrayOutputStream();
+ var code = ConnectInternalTopics.mainNoExit(
+ args,
+ new PrintStream(out, true),
+ new PrintStream(err, true)
+ );
+ return new CommandResult(code, out.toString(), err.toString());
+ }
+
+ private static void assertTopicConfig(Admin admin, String topic, String
configKey, String expectedValue) throws Exception {
+ var resource = new ConfigResource(ConfigResource.Type.TOPIC, topic);
+ var describeResult =
admin.describeConfigs(Collections.singleton(resource));
+ var config = describeResult.all().get().get(resource);
+ assertEquals(expectedValue, config.get(configKey).value());
+ }
+
+ private static void assertTopicReplicationFactor(Admin admin, String
topic, short expectedReplicationFactor) throws Exception {
+ var topicDescriptionFuture =
admin.describeTopics(Collections.singleton(topic)).topicNameValues().get(topic);
+ var topicDescription = topicDescriptionFuture.get();
+ var replicationFactor =
topicDescription.partitions().get(0).replicas().size();
+ assertEquals(expectedReplicationFactor, replicationFactor);
+ }
+
+ private static void assertTopicPartitions(Admin admin, String topic, int
expectedPartitions) throws Exception {
+ var topicDescriptionFuture =
admin.describeTopics(Collections.singleton(topic)).topicNameValues().get(topic);
+ var topicDescription = topicDescriptionFuture.get();
+ var partitions = topicDescription.partitions().size();
+ assertEquals(expectedPartitions, partitions);
+ }
+
+ private static void waitForTopics(Admin admin, Set<String> expectedTopics)
throws InterruptedException {
+ TestUtils.waitForCondition(() ->
admin.listTopics().names().get().containsAll(expectedTopics),
+ "timed out waiting for topics");
+ }
+
+ private static Path setupWorkerConfig(Path path, Properties properties)
throws IOException {
+ path.getParent().toFile().mkdirs();
+ try (var outputStream = Files.newOutputStream(path)) {
+ properties.store(outputStream, "worker properties file");
+ }
+ return path;
+ }
+}
