m1a2st commented on code in PR #20384:
URL: https://github.com/apache/kafka/pull/20384#discussion_r3644251124
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedConfig.java:
##########
@@ -172,12 +176,14 @@ public final class DistributedConfig extends WorkerConfig
{
* <code>status.storage.partitions</code>
*/
public static final String STATUS_STORAGE_PARTITIONS_CONFIG =
STATUS_STORAGE_PREFIX + PARTITIONS_SUFFIX;
+ public static final int STATUS_STORAGE_PARTITIONS_DEFAULT = 5;
private static final String STATUS_STORAGE_PARTITIONS_CONFIG_DOC = "The
number of partitions used when creating the status storage topic";
/**
* <code>status.storage.replication.factor</code>
*/
public static final String STATUS_STORAGE_REPLICATION_FACTOR_CONFIG =
STATUS_STORAGE_PREFIX + REPLICATION_FACTOR_SUFFIX;
+ public static final short STATUS_STORAGE_REPLICATION_FACTOR_DEFAULT = 3;
Review Comment:
This default value is part of the public API. If we want to add it, I think
it should go through a separate KIP.
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedConfig.java:
##########
@@ -187,6 +193,12 @@ 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_CREATION_ENABLE_CONFIG =
"internal.topics.automatic.creation.enable";
+ public static final String INTERNAL_TOPICS_CREATION_ENABLE_DOC = "Whether
to automatically create internal topics used by Connect, such as the offset,
config, and status topics. "
+ + "If set to false, these topics must be created manually before
starting the Connect worker.";
+ public static final Boolean INTERNAL_TOPICS_CREATION_ENABLE_DEFAULT = true;
Review Comment:
This should be renamed to `INTERNAL_TOPICS_AUTOMATIC_CREATION_ENABLE_CONFIG`.
##########
connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaTopicBasedBackingStore.java:
##########
@@ -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());
Review Comment:
When `internal.topics.automatic.creation.enable=false`, Kafka Connect also
cannot automatically create a custom `offsets.storage.topic` specified by the
connector. However, the documentation only mentions the "offset, config, and
status topics,".
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedConfig.java:
##########
@@ -187,6 +193,12 @@ 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_CREATION_ENABLE_CONFIG =
"internal.topics.automatic.creation.enable";
+ public static final String INTERNAL_TOPICS_CREATION_ENABLE_DOC = "Whether
to automatically create internal topics used by Connect, such as the offset,
config, and status topics. "
Review Comment:
```suggestion
private static final String INTERNAL_TOPICS_CREATION_ENABLE_DOC =
"Whether to automatically create internal topics used by Connect, such as the
offset, config, and status topics. "
```
##########
tools/src/main/java/org/apache/kafka/tools/ConnectInternalTopics.java:
##########
@@ -0,0 +1,243 @@
+/*
+ * 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.Exit;
+import org.apache.kafka.common.utils.Utils;
+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, err);
+ 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, PrintStream err) 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) {
+ out.println("Creating internal topic: " + settings.topicName);
+ var topicDescription = TopicAdmin.defineTopic(settings.topicName)
+ .config(settings.topicSettings)
+ .compacted()
+ .partitions(settings.partitions)
+ .replicationFactor(settings.replicationFactor)
+ .build();
+ sharedAdmin.topicAdmin().createTopics(topicDescription);
+ out.println("Created internal topic: " + 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,
+ DistributedConfig.OFFSET_STORAGE_PARTITIONS_DEFAULT,
+ ConfigDef.Importance.LOW,
+ "")
+
.define(DistributedConfig.OFFSET_STORAGE_REPLICATION_FACTOR_CONFIG,
Review Comment:
Should we add validator for these configs? Otherwise, invalid values
accepted, resulting in confusing broker-side errors rather than clear
configuration validation errors.
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerConfig.java:
##########
@@ -410,6 +410,10 @@ public boolean connectorOffsetsTopicsPermitted() {
return false;
}
+ public boolean internalTopicsCreationEnabled() {
Review Comment:
Please also add Javadoc to this method.
##########
tools/src/main/java/org/apache/kafka/tools/ConnectInternalTopics.java:
##########
@@ -0,0 +1,243 @@
+/*
+ * 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.Exit;
+import org.apache.kafka.common.utils.Utils;
+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, err);
+ 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, PrintStream err) 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) {
+ out.println("Creating internal topic: " + settings.topicName);
+ var topicDescription = TopicAdmin.defineTopic(settings.topicName)
+ .config(settings.topicSettings)
+ .compacted()
+ .partitions(settings.partitions)
+ .replicationFactor(settings.replicationFactor)
+ .build();
+ sharedAdmin.topicAdmin().createTopics(topicDescription);
+ out.println("Created internal topic: " + settings.topicName);
Review Comment:
If the topic already exists, logging `Created internal topic: ...` is
misleading. The topic was not actually created.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
