KAFKA-10000: Add all public-facing config properties related to exactly-once source support (KIP-618)

connect/api/src/main/java/org/apache/kafka/connect/source/SourceConnector.java

@@ -63,9 +63,8 @@ public ExactlyOnceSupport exactlyOnceSupport(Map<String, String> connectorConfig
      *
      * @param connectorConfig the configuration that will be used for the connector
      * @return {@link ConnectorTransactionBoundaries#SUPPORTED} if the connector will define its own transaction boundaries,
-     * or {@link ConnectorTransactionBoundaries#UNSUPPORTED} otherwise. If this method is overridden by a
-     * connector, should not be {@code null}, but if {@code null}, it will be assumed that the connector cannot define its own
-     * transaction boundaries.
+     * or {@link ConnectorTransactionBoundaries#UNSUPPORTED} otherwise; may never be {@code null}. The default implementation
+     * returns {@link ConnectorTransactionBoundaries#UNSUPPORTED}.
      * @since 3.3
      * @see TransactionContext
      */

connect/api/src/main/java/org/apache/kafka/connect/source/SourceTask.java

@@ -22,6 +22,7 @@
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
+import java.util.Objects;
 
 /**
  * SourceTask is a Task that pulls records from another system for storage in Kafka.
@@ -64,6 +65,7 @@ public enum TransactionBoundary {
          * @throws IllegalArgumentException if there is no transaction boundary type with the given name
          */
         public static TransactionBoundary fromProperty(String property) {
+            Objects.requireNonNull(property, "Value for transaction boundary property may not be null");
             return TransactionBoundary.valueOf(property.toUpperCase(Locale.ROOT).trim());
         }
 

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

@@ -20,21 +20,31 @@
 import org.apache.kafka.common.config.ConfigDef;
 import org.apache.kafka.common.config.ConfigException;
 import org.apache.kafka.connect.runtime.isolation.Plugins;
+import org.apache.kafka.connect.source.SourceTask;
 
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.stream.Collectors;
 
+import static org.apache.kafka.connect.runtime.SourceConnectorConfig.ExactlyOnceSupportLevel.REQUESTED;
+import static org.apache.kafka.connect.runtime.SourceConnectorConfig.ExactlyOnceSupportLevel.REQUIRED;
 import static org.apache.kafka.connect.runtime.TopicCreationConfig.DEFAULT_TOPIC_CREATION_GROUP;
 import static org.apache.kafka.connect.runtime.TopicCreationConfig.DEFAULT_TOPIC_CREATION_PREFIX;
 import static org.apache.kafka.connect.runtime.TopicCreationConfig.EXCLUDE_REGEX_CONFIG;
 import static org.apache.kafka.connect.runtime.TopicCreationConfig.INCLUDE_REGEX_CONFIG;
 import static org.apache.kafka.connect.runtime.TopicCreationConfig.PARTITIONS_CONFIG;
 import static org.apache.kafka.connect.runtime.TopicCreationConfig.REPLICATION_FACTOR_CONFIG;
+import static org.apache.kafka.connect.source.SourceTask.TransactionBoundary;
+import static org.apache.kafka.connect.source.SourceTask.TransactionBoundary.CONNECTOR;
+import static org.apache.kafka.connect.source.SourceTask.TransactionBoundary.DEFAULT;
+import static org.apache.kafka.connect.source.SourceTask.TransactionBoundary.INTERVAL;
+import static org.apache.kafka.connect.source.SourceTask.TransactionBoundary.POLL;
+import static org.apache.kafka.common.utils.Utils.enumOptions;
 
 public class SourceConnectorConfig extends ConnectorConfig {
 
@@ -47,6 +57,57 @@ public class SourceConnectorConfig extends ConnectorConfig {
             + "created by source connectors";
     private static final String TOPIC_CREATION_GROUPS_DISPLAY = "Topic Creation Groups";
 
+    protected static final String EXACTLY_ONCE_SUPPORT_GROUP = "Exactly Once Support";
+
+    public enum ExactlyOnceSupportLevel {
+        REQUESTED,
+        REQUIRED;
+
+        public static ExactlyOnceSupportLevel fromProperty(String property) {
+            return valueOf(property.toUpperCase(Locale.ROOT).trim());
+        }
+
+        @Override
+        public String toString() {
+            return name().toLowerCase(Locale.ROOT);
+        }
+    }
+
+    public static final String EXACTLY_ONCE_SUPPORT_CONFIG = "exactly.once.support";
+    private static final String EXACTLY_ONCE_SUPPORT_DOC = "Permitted values are " + String.join(", ", enumOptions(ExactlyOnceSupportLevel.class)) + ". "
+            + "If set to \"" + REQUIRED + "\", forces a preflight check for the connector to ensure that it can provide exactly-once delivery "
+            + "with the given configuration. Some connectors may be capable of providing exactly-once delivery but not signal to "
+            + "Connect that they support this; in that case, documentation for the connector should be consulted carefully before "
+            + "creating it, and the value for this property should be set to \"" + REQUESTED + "\". "
+            + "Additionally, if the value is set to \"" + REQUIRED + "\" but the worker that performs preflight validation does not have "
+            + "exactly-once support enabled for source connectors, requests to create or validate the connector will fail.";
+    private static final String EXACTLY_ONCE_SUPPORT_DISPLAY = "Exactly once support";
+
+    public static final String TRANSACTION_BOUNDARY_CONFIG = SourceTask.TRANSACTION_BOUNDARY_CONFIG;
+    private static final String TRANSACTION_BOUNDARY_DOC = "Permitted values are: " + String.join(", ", enumOptions(TransactionBoundary.class)) + ". "
+            + "If set to '" + POLL + "', a new producer transaction will be started and committed for every batch of records that each task from "
+            + "this connector provides to Connect. If set to '" + CONNECTOR + "', relies on connector-defined transaction boundaries; note that "
+            + "not all connectors are capable of defining their own transaction boundaries, and in that case, attempts to instantiate a connector with "
+            + "this value will fail. Finally, if set to '" + INTERVAL + "', commits transactions only after a user-defined time interval has passed.";
+    private static final String TRANSACTION_BOUNDARY_DISPLAY = "Transaction Boundary";
+
+    public static final String TRANSACTION_BOUNDARY_INTERVAL_CONFIG = "transaction.boundary.interval.ms";
+    private static final String TRANSACTION_BOUNDARY_INTERVAL_DOC = "If '" + TRANSACTION_BOUNDARY_CONFIG + "' is set to '" + INTERVAL
+            + "', determines the interval for producer transaction commits by connector tasks. If unset, defaults to the value of the worker-level "
+            + "'" + WorkerConfig.OFFSET_COMMIT_INTERVAL_MS_CONFIG + "' property. It has no effect if a different "
+            + TRANSACTION_BOUNDARY_CONFIG + " is specified.";
+    private static final String TRANSACTION_BOUNDARY_INTERVAL_DISPLAY = "Transaction boundary interval";
+
+    protected static final String OFFSETS_TOPIC_GROUP = "offsets.topic";
+
+    public static final String OFFSETS_TOPIC_CONFIG = "offsets.storage.topic";
+    private static final String OFFSETS_TOPIC_DOC = "The name of a separate offsets topic to use for this connector. "
+            + "If empty or not specified, the worker’s global offsets topic name will be used. "
+            + "If specified, the offsets topic will be created if it does not already exist on the Kafka cluster targeted by this connector "
+            + "(which may be different from the one used for the worker's global offsets topic if the bootstrap.servers property of the connector's producer "
+            + "has been overridden from the worker's). Only applicable in distributed mode; in standalone mode, setting this property will have no effect.";
+    private static final String OFFSETS_TOPIC_DISPLAY = "Offsets topic";
+
     private static class EnrichedSourceConnectorConfig extends ConnectorConfig {
         EnrichedSourceConnectorConfig(Plugins plugins, ConfigDef configDef, Map<String, String> props) {
             super(plugins, configDef, props);
@@ -58,23 +119,87 @@ public Object get(String key) {
         }
     }
 
-    private static final ConfigDef CONFIG = SourceConnectorConfig.configDef();
+    private final TransactionBoundary transactionBoundary;
+    private final Long transactionBoundaryInterval;
     private final EnrichedSourceConnectorConfig enrichedSourceConfig;
+    private final String offsetsTopic;
 
     public static ConfigDef configDef() {
+        ConfigDef.Validator atLeastZero = ConfigDef.Range.atLeast(0);
         int orderInGroup = 0;
         return new ConfigDef(ConnectorConfig.configDef())
-                .define(TOPIC_CREATION_GROUPS_CONFIG, ConfigDef.Type.LIST, Collections.emptyList(),
-                        ConfigDef.CompositeValidator.of(new ConfigDef.NonNullValidator(), ConfigDef.LambdaValidator.with(
+                .define(
+                        TOPIC_CREATION_GROUPS_CONFIG,
+                        ConfigDef.Type.LIST,
+                        Collections.emptyList(),
+                        ConfigDef.CompositeValidator.of(
+                                new ConfigDef.NonNullValidator(),
+                                ConfigDef.LambdaValidator.with(
+                                    (name, value) -> {
+                                        List<?> groupAliases = (List<?>) value;
+                                        if (groupAliases.size() > new HashSet<>(groupAliases).size()) {
+                                            throw new ConfigException(name, value, "Duplicate alias provided.");
+                                        }
+                                    },
+                                    () -> "unique topic creation groups")),
+                        ConfigDef.Importance.LOW,
+                        TOPIC_CREATION_GROUPS_DOC,
+                        TOPIC_CREATION_GROUP,
+                        ++orderInGroup,
+                        ConfigDef.Width.LONG,
+                        TOPIC_CREATION_GROUPS_DISPLAY)
+                .define(
+                        EXACTLY_ONCE_SUPPORT_CONFIG,
+                        ConfigDef.Type.STRING,
+                        REQUESTED.toString(),
+                        ConfigDef.CaseInsensitiveValidString.in(enumOptions(ExactlyOnceSupportLevel.class)),
+                        ConfigDef.Importance.MEDIUM,
+                        EXACTLY_ONCE_SUPPORT_DOC,
+                        EXACTLY_ONCE_SUPPORT_GROUP,
+                        ++orderInGroup,
+                        ConfigDef.Width.SHORT,
+                        EXACTLY_ONCE_SUPPORT_DISPLAY)
+                .define(
+                        TRANSACTION_BOUNDARY_CONFIG,
+                        ConfigDef.Type.STRING,
+                        DEFAULT.toString(),
+                        ConfigDef.CaseInsensitiveValidString.in(enumOptions(TransactionBoundary.class)),
+                        ConfigDef.Importance.MEDIUM,
+                        TRANSACTION_BOUNDARY_DOC,
+                        EXACTLY_ONCE_SUPPORT_GROUP,
+                        ++orderInGroup,
+                        ConfigDef.Width.SHORT,
+                        TRANSACTION_BOUNDARY_DISPLAY)
+                .define(
+                        TRANSACTION_BOUNDARY_INTERVAL_CONFIG,
+                        ConfigDef.Type.LONG,
+                        null,
+                        ConfigDef.LambdaValidator.with(
                             (name, value) -> {
-                                List<?> groupAliases = (List<?>) value;
-                                if (groupAliases.size() > new HashSet<>(groupAliases).size()) {
-                                    throw new ConfigException(name, value, "Duplicate alias provided.");
+                                if (value == null) {
+                                    return;
                                 }
+                                atLeastZero.ensureValid(name, value);
                             },
-                            () -> "unique topic creation groups")),
-                        ConfigDef.Importance.LOW, TOPIC_CREATION_GROUPS_DOC, TOPIC_CREATION_GROUP,
-                        ++orderInGroup, ConfigDef.Width.LONG, TOPIC_CREATION_GROUPS_DISPLAY);
+                            atLeastZero::toString
+                        ),
+                        ConfigDef.Importance.LOW,
+                        TRANSACTION_BOUNDARY_INTERVAL_DOC,
+                        EXACTLY_ONCE_SUPPORT_GROUP,
+                        ++orderInGroup,
+                        ConfigDef.Width.SHORT,
+                        TRANSACTION_BOUNDARY_INTERVAL_DISPLAY)
+                .define(
+                        OFFSETS_TOPIC_CONFIG,
+                        ConfigDef.Type.STRING,
+                        null,
+                        new ConfigDef.NonEmptyString(),
+                        ConfigDef.Importance.LOW,
+                        OFFSETS_TOPIC_DOC,
+                        OFFSETS_TOPIC_GROUP,
+                        orderInGroup = 1,
+                        ConfigDef.Width.LONG,
+                        OFFSETS_TOPIC_DISPLAY);
     }
 
     public static ConfigDef embedDefaultGroup(ConfigDef baseConfigDef) {
@@ -116,9 +241,9 @@ public static ConfigDef enrich(ConfigDef baseConfigDef, Map<String, String> prop
     }
 
     public SourceConnectorConfig(Plugins plugins, Map<String, String> props, boolean createTopics) {
-        super(plugins, CONFIG, props);
+        super(plugins, configDef(), props);
         if (createTopics && props.entrySet().stream().anyMatch(e -> e.getKey().startsWith(TOPIC_CREATION_PREFIX))) {
-            ConfigDef defaultConfigDef = embedDefaultGroup(CONFIG);
+            ConfigDef defaultConfigDef = embedDefaultGroup(configDef());
             // This config is only used to set default values for partitions and replication
             // factor from the default group and otherwise it remains unused
             AbstractConfig defaultGroup = new AbstractConfig(defaultConfigDef, props, false);
@@ -135,13 +260,32 @@ public SourceConnectorConfig(Plugins plugins, Map<String, String> props, boolean
         } else {
             enrichedSourceConfig = null;
         }
+        transactionBoundary = TransactionBoundary.fromProperty(getString(TRANSACTION_BOUNDARY_CONFIG));
+        transactionBoundaryInterval = getLong(TRANSACTION_BOUNDARY_INTERVAL_CONFIG);
+        offsetsTopic = getString(OFFSETS_TOPIC_CONFIG);
+    }
+
+    public static boolean usesTopicCreation(Map<String, String> props) {
+        return props.entrySet().stream().anyMatch(e -> e.getKey().startsWith(TOPIC_CREATION_PREFIX));
     }
 
     @Override
     public Object get(String key) {
         return enrichedSourceConfig != null ? enrichedSourceConfig.get(key) : super.get(key);
     }
 
+    public TransactionBoundary transactionBoundary() {
+        return transactionBoundary;
+    }
+
+    public Long transactionBoundaryInterval() {
+        return transactionBoundaryInterval;
+    }
+
+    public String offsetsTopic() {
+        return offsetsTopic;
+    }
+
     /**
      * Returns whether this configuration uses topic creation properties.
      *
@@ -181,6 +325,6 @@ public Map<String, Object> topicCreationOtherConfigs(String group) {
     }
 
     public static void main(String[] args) {
-        System.out.println(CONFIG.toHtml(4, config -> "sourceconnectorconfigs_" + config));
+        System.out.println(configDef().toHtml(4, config -> "sourceconnectorconfigs_" + config));
     }
 }

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

@@ -111,7 +111,8 @@ public class WorkerConfig extends AbstractConfig {
     private static final String OFFSET_COMMIT_TIMEOUT_MS_DOC
             = "Maximum number of milliseconds to wait for records to flush and partition offset data to be"
             + " committed to offset storage before cancelling the process and restoring the offset "
-            + "data to be committed in a future attempt.";
+            + "data to be committed in a future attempt. This property has no effect for source connectors "
+            + "running with exactly-once support.";
     public static final long OFFSET_COMMIT_TIMEOUT_MS_DEFAULT = 5000L;
 
     public static final String LISTENERS_CONFIG = "listeners";
@@ -343,6 +344,15 @@ private void logPluginPathConfigProviderWarning(Map<String, String> rawOriginals
         }
     }
 
+    /**
+     * @return the {@link CommonClientConfigs#BOOTSTRAP_SERVERS_CONFIG bootstrap servers} property
+     * used by the worker when instantiating Kafka clients for connectors and tasks (unless overridden)
+     * and its internal topics (if running in distributed mode)
+     */
+    public String bootstrapServers() {
+        return String.join(",", getList(BOOTSTRAP_SERVERS_CONFIG));
+    }
+
     public Integer getRebalanceTimeout() {
         return null;
     }
@@ -351,6 +361,54 @@ public boolean topicCreationEnable() {
         return getBoolean(TOPIC_CREATION_ENABLE_CONFIG);
     }
 
+    /**
+     * Whether this worker is configured with exactly-once support for source connectors.
+     * The default implementation returns {@code false} and should be overridden by subclasses
+     * if the worker mode for the subclass provides exactly-once support for source connectors.
+     * @return whether exactly-once support is enabled for source connectors on this worker
+     */
+    public boolean exactlyOnceSourceEnabled() {
+        return false;
+    }
+
+    /**
+     * Get the internal topic used by this worker to store source connector offsets.
+     * The default implementation returns {@code null} and should be overridden by subclasses
+     * if the worker mode for the subclass uses an internal offsets topic.
+     * @return the name of the internal offsets topic, or {@code null} if the worker does not use
+     * an internal offsets topic
+     */
+    public String offsetsTopic() {
+        return null;
+    }
+
+    /**
+     * Determine whether this worker supports per-connector source offsets topics.
+     * The default implementation returns {@code false} and should be overridden by subclasses
+     * if the worker mode for the subclass supports per-connector offsets topics.
+     * @return whether the worker supports per-connector offsets topics
+     */
+    public boolean connectorOffsetsTopicsPermitted() {
+        return false;
+    }
+
+    /**
+     * @return the offset commit interval for tasks created by this worker
+     */
+    public long offsetCommitInterval() {
+        return getLong(OFFSET_COMMIT_INTERVAL_MS_CONFIG);
+    }
+
+    /**
+     * Get the {@link CommonClientConfigs#GROUP_ID_CONFIG group ID} used by this worker to form a cluster.
+     * The default implementation returns {@code null} and should be overridden by subclasses
+     * if the worker mode for the subclass is capable of forming a cluster using Kafka's group coordination API.
+     * @return the group ID for the worker's cluster, or {@code null} if the worker is not capable of forming a cluster.
+     */
+    public String groupId() {
+        return null;
+    }
+
     @Override
     protected Map<String, Object> postProcessParsedConfig(final Map<String, Object> parsedValues) {
         return CommonClientConfigs.postProcessReconnectBackoffConfigs(this, parsedValues);