diff --git a/connect/api/src/main/java/org/apache/kafka/connect/source/SourceConnector.java b/connect/api/src/main/java/org/apache/kafka/connect/source/SourceConnector.java index 7fc2a5d11cfea..3bd012f9fbce0 100644 --- a/connect/api/src/main/java/org/apache/kafka/connect/source/SourceConnector.java +++ b/connect/api/src/main/java/org/apache/kafka/connect/source/SourceConnector.java @@ -63,9 +63,8 @@ public ExactlyOnceSupport exactlyOnceSupport(Map 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 */ diff --git a/connect/api/src/main/java/org/apache/kafka/connect/source/SourceTask.java b/connect/api/src/main/java/org/apache/kafka/connect/source/SourceTask.java index 2159e68e8a88a..559f02340ca17 100644 --- a/connect/api/src/main/java/org/apache/kafka/connect/source/SourceTask.java +++ b/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()); } diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/SourceConnectorConfig.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/SourceConnectorConfig.java index e38072b9b6e50..65d376b08f7ed 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/SourceConnectorConfig.java +++ b/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 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 prop } public SourceConnectorConfig(Plugins plugins, Map 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,6 +260,13 @@ public SourceConnectorConfig(Plugins plugins, Map 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 props) { + return props.entrySet().stream().anyMatch(e -> e.getKey().startsWith(TOPIC_CREATION_PREFIX)); } @Override @@ -142,6 +274,18 @@ 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 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)); } } 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 5894283c93531..38dbeb87e1bf7 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 @@ -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 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 postProcessParsedConfig(final Map parsedValues) { return CommonClientConfigs.postProcessReconnectBackoffConfigs(this, parsedValues); 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 23083f5e8e009..849a5969468a5 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 @@ -17,6 +17,7 @@ package org.apache.kafka.connect.runtime.distributed; import org.apache.kafka.clients.CommonClientConfigs; +import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.common.config.ConfigDef; import org.apache.kafka.common.config.ConfigException; import org.apache.kafka.common.config.TopicConfig; @@ -32,6 +33,7 @@ import java.security.NoSuchAlgorithmException; import java.util.Collections; import java.util.List; +import java.util.Locale; import java.util.Map; import java.util.Optional; import java.util.concurrent.TimeUnit; @@ -39,6 +41,7 @@ import static org.apache.kafka.common.config.ConfigDef.Range.atLeast; import static org.apache.kafka.common.config.ConfigDef.Range.between; import static org.apache.kafka.common.config.ConfigDef.ValidString.in; +import static org.apache.kafka.common.utils.Utils.enumOptions; import static org.apache.kafka.connect.runtime.TopicCreationConfig.PARTITIONS_VALIDATOR; import static org.apache.kafka.connect.runtime.TopicCreationConfig.REPLICATION_FACTOR_VALIDATOR; @@ -194,6 +197,34 @@ public class DistributedConfig extends WorkerConfig { public static final String INTER_WORKER_VERIFICATION_ALGORITHMS_DOC = "A list of permitted algorithms for verifying internal requests"; public static final List INTER_WORKER_VERIFICATION_ALGORITHMS_DEFAULT = Collections.singletonList(INTER_WORKER_SIGNATURE_ALGORITHM_DEFAULT); + private enum ExactlyOnceSourceSupport { + DISABLED(false), + PREPARING(true), + ENABLED(true); + + public final boolean usesTransactionalLeader; + + ExactlyOnceSourceSupport(boolean usesTransactionalLeader) { + this.usesTransactionalLeader = usesTransactionalLeader; + } + + public static ExactlyOnceSourceSupport fromProperty(String property) { + return ExactlyOnceSourceSupport.valueOf(property.toUpperCase(Locale.ROOT)); + } + + @Override + public String toString() { + return name().toLowerCase(Locale.ROOT); + } + } + + public static final String EXACTLY_ONCE_SOURCE_SUPPORT_CONFIG = "exactly.once.source.support"; + public static final String EXACTLY_ONCE_SOURCE_SUPPORT_DOC = "Whether to enable exactly-once support for source connectors in the cluster " + + "by using transactions to write source records and their source offsets, and by proactively fencing out old task generations before bringing up new ones. "; + // TODO: https://issues.apache.org/jira/browse/KAFKA-13709 + // + "See the exactly-once source support documentation at [add docs link here] for more information on this feature."; + public static final String EXACTLY_ONCE_SOURCE_SUPPORT_DEFAULT = ExactlyOnceSourceSupport.DISABLED.toString(); + @SuppressWarnings("unchecked") private static final ConfigDef CONFIG = baseConfigDef() .define(GROUP_ID_CONFIG, @@ -215,6 +246,12 @@ public class DistributedConfig extends WorkerConfig { Math.toIntExact(TimeUnit.SECONDS.toMillis(3)), ConfigDef.Importance.HIGH, HEARTBEAT_INTERVAL_MS_DOC) + .define(EXACTLY_ONCE_SOURCE_SUPPORT_CONFIG, + ConfigDef.Type.STRING, + EXACTLY_ONCE_SOURCE_SUPPORT_DEFAULT, + ConfigDef.CaseInsensitiveValidString.in(enumOptions(ExactlyOnceSourceSupport.class)), + ConfigDef.Importance.HIGH, + EXACTLY_ONCE_SOURCE_SUPPORT_DOC) .define(CommonClientConfigs.METADATA_MAX_AGE_CONFIG, ConfigDef.Type.LONG, TimeUnit.MINUTES.toMillis(5), @@ -399,13 +436,57 @@ public class DistributedConfig extends WorkerConfig { ConfigDef.Importance.LOW, INTER_WORKER_VERIFICATION_ALGORITHMS_DOC); + private final ExactlyOnceSourceSupport exactlyOnceSourceSupport; + @Override public Integer getRebalanceTimeout() { return getInt(DistributedConfig.REBALANCE_TIMEOUT_MS_CONFIG); } + @Override + public boolean exactlyOnceSourceEnabled() { + return exactlyOnceSourceSupport == ExactlyOnceSourceSupport.ENABLED; + } + + /** + * @return whether the Connect cluster's leader should use a transactional producer to perform writes to the config + * topic, which is useful for ensuring that zombie leaders are fenced out and unable to write to the topic after a + * new leader has been elected. + */ + public boolean transactionalLeaderEnabled() { + return exactlyOnceSourceSupport.usesTransactionalLeader; + } + + /** + * @return the {@link ProducerConfig#TRANSACTIONAL_ID_CONFIG transactional ID} to use for the worker's producer if + * using a transactional producer for writes to internal topics such as the config topic. + */ + public String transactionalProducerId() { + return transactionalProducerId(groupId()); + } + + public static String transactionalProducerId(String groupId) { + return "connect-cluster-" + groupId; + } + + @Override + public String offsetsTopic() { + return getString(OFFSET_STORAGE_TOPIC_CONFIG); + } + + @Override + public boolean connectorOffsetsTopicsPermitted() { + return true; + } + + @Override + public String groupId() { + return getString(GROUP_ID_CONFIG); + } + public DistributedConfig(Map props) { super(CONFIG, props); + exactlyOnceSourceSupport = ExactlyOnceSourceSupport.fromProperty(getString(EXACTLY_ONCE_SOURCE_SUPPORT_CONFIG)); getInternalRequestKeyGenerator(); // Check here for a valid key size + key algorithm to fail fast if either are invalid validateKeyAlgorithmAndVerificationAlgorithms(); } diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractHerderTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractHerderTest.java index 5b9e199e5a1ee..862f31c077087 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractHerderTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractHerderTest.java @@ -53,6 +53,7 @@ import org.junit.Test; import org.junit.runner.RunWith; import org.powermock.api.easymock.PowerMock; +import org.powermock.api.easymock.annotation.Mock; import org.powermock.api.easymock.annotation.MockStrict; import org.powermock.core.classloader.annotations.PrepareForTest; import org.powermock.modules.junit4.PowerMockRunner; @@ -142,10 +143,10 @@ public class AbstractHerderTest { @MockStrict private Worker worker; @MockStrict private WorkerConfigTransformer transformer; - @MockStrict private Plugins plugins; - @MockStrict private ClassLoader classLoader; @MockStrict private ConfigBackingStore configStore; @MockStrict private StatusBackingStore statusStore; + @MockStrict private ClassLoader classLoader; + @Mock private Plugins plugins; @Test public void testConnectors() { @@ -436,13 +437,18 @@ public void testConfigValidationMissingName() { // We expect there to be errors due to the missing name and .... Note that these assertions depend heavily on // the config fields for SourceConnectorConfig, but we expect these to change rarely. assertEquals(SampleSourceConnector.class.getName(), result.name()); - assertEquals(Arrays.asList(ConnectorConfig.COMMON_GROUP, ConnectorConfig.TRANSFORMS_GROUP, - ConnectorConfig.PREDICATES_GROUP, ConnectorConfig.ERROR_GROUP, SourceConnectorConfig.TOPIC_CREATION_GROUP), result.groups()); + assertEquals( + Arrays.asList( + ConnectorConfig.COMMON_GROUP, ConnectorConfig.TRANSFORMS_GROUP, + ConnectorConfig.PREDICATES_GROUP, ConnectorConfig.ERROR_GROUP, + SourceConnectorConfig.TOPIC_CREATION_GROUP, SourceConnectorConfig.EXACTLY_ONCE_SUPPORT_GROUP, + SourceConnectorConfig.OFFSETS_TOPIC_GROUP), + result.groups()); assertEquals(2, result.errorCount()); Map infos = result.values().stream() .collect(Collectors.toMap(info -> info.configKey().name(), Function.identity())); - // Base connector config has 14 fields, connector's configs add 2 - assertEquals(17, infos.size()); + // Base connector config has 14 fields, connector's configs add 7 + assertEquals(21, infos.size()); // Missing name should generate an error assertEquals(ConnectorConfig.NAME_CONFIG, infos.get(ConnectorConfig.NAME_CONFIG).configValue().name()); @@ -531,6 +537,8 @@ public void testConfigValidationTransformsExtendResults() { ConnectorConfig.PREDICATES_GROUP, ConnectorConfig.ERROR_GROUP, SourceConnectorConfig.TOPIC_CREATION_GROUP, + SourceConnectorConfig.EXACTLY_ONCE_SUPPORT_GROUP, + SourceConnectorConfig.OFFSETS_TOPIC_GROUP, "Transforms: xformA", "Transforms: xformB" ); @@ -538,7 +546,7 @@ public void testConfigValidationTransformsExtendResults() { assertEquals(2, result.errorCount()); Map infos = result.values().stream() .collect(Collectors.toMap(info -> info.configKey().name(), Function.identity())); - assertEquals(22, infos.size()); + assertEquals(26, infos.size()); // Should get 2 type fields from the transforms, first adds its own config since it has a valid class assertEquals("transforms.xformA.type", infos.get("transforms.xformA.type").configValue().name()); @@ -590,6 +598,8 @@ public void testConfigValidationPredicatesExtendResults() { ConnectorConfig.PREDICATES_GROUP, ConnectorConfig.ERROR_GROUP, SourceConnectorConfig.TOPIC_CREATION_GROUP, + SourceConnectorConfig.EXACTLY_ONCE_SUPPORT_GROUP, + SourceConnectorConfig.OFFSETS_TOPIC_GROUP, "Transforms: xformA", "Predicates: predX", "Predicates: predY" @@ -598,7 +608,7 @@ public void testConfigValidationPredicatesExtendResults() { assertEquals(2, result.errorCount()); Map infos = result.values().stream() .collect(Collectors.toMap(info -> info.configKey().name(), Function.identity())); - assertEquals(24, infos.size()); + assertEquals(28, infos.size()); // Should get 2 type fields from the transforms, first adds its own config since it has a valid class assertEquals("transforms.xformA.type", infos.get("transforms.xformA.type").configValue().name()); @@ -659,12 +669,14 @@ public void testConfigValidationPrincipalOnlyOverride() { ConnectorConfig.TRANSFORMS_GROUP, ConnectorConfig.PREDICATES_GROUP, ConnectorConfig.ERROR_GROUP, - SourceConnectorConfig.TOPIC_CREATION_GROUP + SourceConnectorConfig.TOPIC_CREATION_GROUP, + SourceConnectorConfig.EXACTLY_ONCE_SUPPORT_GROUP, + SourceConnectorConfig.OFFSETS_TOPIC_GROUP ); assertEquals(expectedGroups, result.groups()); assertEquals(1, result.errorCount()); - // Base connector config has 14 fields, connector's configs add 2, and 2 producer overrides - assertEquals(19, result.values().size()); + // Base connector config has 14 fields, connector's configs add 7, and 2 producer overrides + assertEquals(23, result.values().size()); assertTrue(result.values().stream().anyMatch( configInfo -> ackConfigKey.equals(configInfo.configValue().name()) && !configInfo.configValue().errors().isEmpty())); assertTrue(result.values().stream().anyMatch( diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedConfigTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedConfigTest.java index 21364ae8e1738..12085b21d9602 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedConfigTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedConfigTest.java @@ -28,11 +28,14 @@ import java.util.List; import java.util.Map; +import static org.apache.kafka.connect.runtime.distributed.DistributedConfig.EXACTLY_ONCE_SOURCE_SUPPORT_CONFIG; +import static org.apache.kafka.connect.runtime.distributed.DistributedConfig.GROUP_ID_CONFIG; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertThrows; -import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.Assert.assertTrue; public class DistributedConfigTest { @@ -306,4 +309,42 @@ public void testInvalidSecurityProtocol() { () -> new DistributedConfig(configs)); assertTrue(ce.getMessage().contains(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG)); } + + @Test + public void shouldIdentifyNeedForTransactionalLeader() { + Map workerProps = configs(); + + workerProps.put(EXACTLY_ONCE_SOURCE_SUPPORT_CONFIG, "disabled"); + assertFalse(new DistributedConfig(workerProps).transactionalLeaderEnabled()); + + workerProps.put(EXACTLY_ONCE_SOURCE_SUPPORT_CONFIG, "preparing"); + assertTrue(new DistributedConfig(workerProps).transactionalLeaderEnabled()); + + workerProps.put(EXACTLY_ONCE_SOURCE_SUPPORT_CONFIG, "enabled"); + assertTrue(new DistributedConfig(workerProps).transactionalLeaderEnabled()); + } + + @Test + public void shouldConstructExpectedTransactionalId() { + Map workerProps = configs(); + + workerProps.put(GROUP_ID_CONFIG, "why did i stay up all night writing unit tests"); + assertEquals( + "connect-cluster-why did i stay up all night writing unit tests", + new DistributedConfig(workerProps).transactionalProducerId() + ); + + workerProps.put(GROUP_ID_CONFIG, "connect-cluster"); + assertEquals( + "connect-cluster-connect-cluster", + new DistributedConfig(workerProps).transactionalProducerId() + ); + + workerProps.put(GROUP_ID_CONFIG, "\u2603"); + assertEquals( + "connect-cluster-\u2603", + new DistributedConfig(workerProps).transactionalProducerId() + ); + } + }