diff --git a/checkstyle/suppressions.xml b/checkstyle/suppressions.xml index 10af4e66c28dd..93aaadec0eebb 100644 --- a/checkstyle/suppressions.xml +++ b/checkstyle/suppressions.xml @@ -130,7 +130,7 @@ + files="(DistributedHerder|KafkaConfigBackingStore|Values|IncrementalCooperativeAssignor).java"/> connectorTargetStateChanges = new HashSet<>(); private boolean needsReconfigRebalance; + private volatile boolean fencedFromConfigTopic; private volatile int generation; private volatile long scheduledRebalance; private volatile SecretKey sessionKey; @@ -288,6 +291,7 @@ public DistributedHerder(DistributedConfig config, configState = ClusterConfigState.EMPTY; rebalanceResolved = true; // If we still need to follow up after a rebalance occurred, starting up tasks needsReconfigRebalance = false; + fencedFromConfigTopic = false; canReadConfigs = true; // We didn't try yet, but Configs are readable until proven otherwise scheduledRebalance = Long.MAX_VALUE; keyExpiration = Long.MAX_VALUE; @@ -372,18 +376,36 @@ public void tick() { return; } + if (fencedFromConfigTopic) { + if (isLeader()) { + // We were accidentally fenced out, possibly by a zombie leader + try { + log.debug("Reclaiming write privileges for config topic after being fenced out"); + configBackingStore.claimWritePrivileges(); + fencedFromConfigTopic = false; + log.debug("Successfully reclaimed write privileges for config topic after being fenced out"); + } catch (Exception e) { + log.warn("Unable to claim write privileges for config topic. Will backoff and possibly retry if still the leader", e); + backoff(CONFIG_TOPIC_WRITE_PRIVILEGES_BACKOFF_MS); + return; + } + } else { + log.trace("Relinquished write privileges for config topic after being fenced out, since worker is no longer the leader of the cluster"); + // We were meant to be fenced out because we fell out of the group and a new leader was elected + fencedFromConfigTopic = false; + } + } + long now = time.milliseconds(); if (checkForKeyRotation(now)) { log.debug("Distributing new session key"); keyExpiration = Long.MAX_VALUE; try { - configBackingStore.putSessionKey(new SessionKey( - keyGenerator.generateKey(), - now - )); + SessionKey newSessionKey = new SessionKey(keyGenerator.generateKey(), now); + writeToConfigTopicAsLeader(() -> configBackingStore.putSessionKey(newSessionKey)); } catch (Exception e) { - log.info("Failed to write new session key to config topic; forcing a read to the end of the config topic before possibly retrying"); + log.info("Failed to write new session key to config topic; forcing a read to the end of the config topic before possibly retrying", e); canReadConfigs = false; return; } @@ -492,6 +514,12 @@ private boolean checkForKeyRotation(long now) { SecretKey key; long expiration; synchronized (this) { + // This happens on startup; the snapshot contains the session key, + // but no callback in the config update listener has been fired for it yet. + if (sessionKey == null && configState.sessionKey() != null) { + sessionKey = configState.sessionKey().key(); + keyExpiration = configState.sessionKey().creationTimestamp() + keyRotationIntervalMs; + } key = sessionKey; expiration = keyExpiration; } @@ -511,10 +539,6 @@ private boolean checkForKeyRotation(long now) { + "than required by current worker configuration. Distributing new key now."); return true; } - } else if (key == null && configState.sessionKey() != null) { - // This happens on startup for follower workers; the snapshot contains the session key, - // but no callback in the config update listener has been fired for it yet. - sessionKey = configState.sessionKey().key(); } } return false; @@ -836,7 +860,7 @@ public void deleteConnectorConfig(final String connName, final Callback configBackingStore.removeConnectorConfig(connName)); callback.onCompletion(null, new Created<>(false, null)); } return null; @@ -1008,7 +1032,7 @@ public void putConnectorConfig(final String connName, final Map } log.trace("Submitting connector config {} {} {}", connName, allowReplace, configState.connectors()); - configBackingStore.putConnectorConfig(connName, config); + writeToConfigTopicAsLeader(() -> configBackingStore.putConnectorConfig(connName, config)); // Note that we use the updated connector config despite the fact that we don't have an updated // snapshot yet. The existing task info should still be accurate. @@ -1107,7 +1131,7 @@ public void putTaskConfigs(final String connName, final List else if (!configState.contains(connName)) callback.onCompletion(new NotFoundException("Connector " + connName + " not found"), null); else { - configBackingStore.putTaskConfigs(connName, configs); + writeToConfigTopicAsLeader(() -> configBackingStore.putTaskConfigs(connName, configs)); callback.onCompletion(null, null); } return null; @@ -1328,6 +1352,25 @@ private String leaderUrl() { return assignment.leaderUrl(); } + /** + * Perform an action that writes to the config topic, and if it fails because the leader has been fenced out, make note of that + * fact so that we can try to reclaim write ownership (if still the leader of the cluster) in a subsequent iteration of the tick loop. + * Note that it is not necessary to wrap every write to the config topic in this method, only the writes that should be performed + * exclusively by the leader. For example, {@link ConfigBackingStore#putTargetState(String, TargetState)} does not require this + * method, as it can be invoked by any worker in the cluster. + * @param write the action that writes to the config topic, such as {@link ConfigBackingStore#putSessionKey(SessionKey)} or + * {@link ConfigBackingStore#putConnectorConfig(String, Map)}. + */ + private void writeToConfigTopicAsLeader(Runnable write) { + try { + write.run(); + } catch (PrivilegedWriteException e) { + log.warn("Failed to write to config topic as leader; will rejoin group if necessary and, if still leader, attempt to reclaim write privileges for the config topic", e); + fencedFromConfigTopic = true; + throw new ConnectException("Failed to write to config topic; this may be due to a transient error and the request can be safely retried", e); + } + } + /** * Handle post-assignment operations, either trying to resolve issues that kept assignment from completing, getting * this node into sync and its work started. @@ -1700,7 +1743,7 @@ private void reconfigureConnector(final String connName, final Callback cb if (changed) { List> rawTaskProps = reverseTransform(connName, configState, taskProps); if (isLeader()) { - configBackingStore.putTaskConfigs(connName, rawTaskProps); + writeToConfigTopicAsLeader(() -> configBackingStore.putTaskConfigs(connName, rawTaskProps)); cb.onCompletion(null, null); } else { // We cannot forward the request on the same thread because this reconfiguration can happen as a result of connector @@ -2009,12 +2052,20 @@ public void onAssigned(ExtendedAssignment assignment, int generation) { herderMetrics.rebalanceStarted(time.milliseconds()); } - // Delete the statuses of all connectors and tasks removed prior to the start of this rebalance. This - // has to be done after the rebalance completes to avoid race conditions as the previous generation - // attempts to change the state to UNASSIGNED after tasks have been stopped. if (isLeader()) { + // Delete the statuses of all connectors and tasks removed prior to the start of this rebalance. This + // has to be done after the rebalance completes to avoid race conditions as the previous generation + // attempts to change the state to UNASSIGNED after tasks have been stopped. updateDeletedConnectorStatus(); updateDeletedTaskStatus(); + // As the leader, we're now allowed to write directly to the config topic for important things like + // connector configs, session keys, and task count records + try { + configBackingStore.claimWritePrivileges(); + } catch (Exception e) { + fencedFromConfigTopic = true; + log.error("Unable to claim write privileges for config topic after being elected leader during rebalance", e); + } } // We *must* interrupt any poll() call since this could occur when the poll starts, and we might then diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/EagerAssignor.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/EagerAssignor.java index d86feaaaaf967..f4edb98ed60fb 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/EagerAssignor.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/EagerAssignor.java @@ -18,6 +18,7 @@ import org.apache.kafka.common.utils.CircularIterator; import org.apache.kafka.common.utils.LogContext; +import org.apache.kafka.connect.storage.ClusterConfigState; import org.apache.kafka.connect.util.ConnectorTaskId; import org.slf4j.Logger; diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/IncrementalCooperativeAssignor.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/IncrementalCooperativeAssignor.java index 8ed80903c58ec..57e7b004857f7 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/IncrementalCooperativeAssignor.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/IncrementalCooperativeAssignor.java @@ -23,6 +23,7 @@ import org.apache.kafka.connect.runtime.distributed.WorkerCoordinator.ConnectorsAndTasks; import org.apache.kafka.connect.runtime.distributed.WorkerCoordinator.WorkerLoad; import org.apache.kafka.connect.util.ConnectUtils; +import org.apache.kafka.connect.storage.ClusterConfigState; import org.apache.kafka.connect.util.ConnectorTaskId; import org.slf4j.Logger; diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/WorkerCoordinator.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/WorkerCoordinator.java index 77fb16eaa3c84..1ab3543361211 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/WorkerCoordinator.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/WorkerCoordinator.java @@ -25,6 +25,7 @@ import org.apache.kafka.common.utils.LogContext; import org.apache.kafka.common.utils.Time; import org.apache.kafka.common.utils.Timer; +import org.apache.kafka.connect.storage.ClusterConfigState; import org.apache.kafka.connect.storage.ConfigBackingStore; import org.apache.kafka.connect.util.ConnectorTaskId; import org.slf4j.Logger; diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/standalone/StandaloneHerder.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/standalone/StandaloneHerder.java index dac389ba0e346..eb3c3393cebe0 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/standalone/StandaloneHerder.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/standalone/StandaloneHerder.java @@ -31,7 +31,7 @@ import org.apache.kafka.connect.runtime.SourceConnectorConfig; import org.apache.kafka.connect.runtime.TargetState; import org.apache.kafka.connect.runtime.Worker; -import org.apache.kafka.connect.runtime.distributed.ClusterConfigState; +import org.apache.kafka.connect.storage.ClusterConfigState; import org.apache.kafka.connect.runtime.rest.InternalRequestSignature; import org.apache.kafka.connect.runtime.rest.entities.ConfigInfos; import org.apache.kafka.connect.runtime.rest.entities.ConnectorInfo; diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/ClusterConfigState.java b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/ClusterConfigState.java similarity index 96% rename from connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/ClusterConfigState.java rename to connect/runtime/src/main/java/org/apache/kafka/connect/storage/ClusterConfigState.java index 717120d8508ee..7a766905b81d6 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/ClusterConfigState.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/ClusterConfigState.java @@ -14,7 +14,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.kafka.connect.runtime.distributed; +package org.apache.kafka.connect.storage; import org.apache.kafka.common.config.provider.ConfigProvider; import org.apache.kafka.connect.runtime.SessionKey; @@ -46,12 +46,12 @@ public class ClusterConfigState { private final long offset; private final SessionKey sessionKey; - private final Map connectorTaskCounts; - private final Map> connectorConfigs; - private final Map connectorTargetStates; - private final Map> taskConfigs; - private final Set inconsistentConnectors; private final WorkerConfigTransformer configTransformer; + final Map connectorTaskCounts; + final Map> connectorConfigs; + final Map connectorTargetStates; + final Map> taskConfigs; + final Set inconsistentConnectors; public ClusterConfigState(long offset, SessionKey sessionKey, diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/ConfigBackingStore.java b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/ConfigBackingStore.java index 826f934ffd550..24539076adc0f 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/ConfigBackingStore.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/ConfigBackingStore.java @@ -19,7 +19,6 @@ import org.apache.kafka.connect.runtime.RestartRequest; import org.apache.kafka.connect.runtime.SessionKey; import org.apache.kafka.connect.runtime.TargetState; -import org.apache.kafka.connect.runtime.distributed.ClusterConfigState; import org.apache.kafka.connect.util.ConnectorTaskId; import java.util.Collection; @@ -90,6 +89,10 @@ public interface ConfigBackingStore { */ void putTargetState(String connector, TargetState state); + /** + * Store a new {@link SessionKey} that can be used to validate internal (i.e., non-user-triggered) inter-worker communication. + * @param sessionKey the session key to store + */ void putSessionKey(SessionKey sessionKey); /** @@ -98,6 +101,15 @@ public interface ConfigBackingStore { */ void putRestartRequest(RestartRequest restartRequest); + /** + * Prepare to write to the backing config store. May be required by some implementations (such as those that only permit a single + * writer at a time across a cluster of workers) before performing mutating operations like writing configurations, target states, etc. + * The default implementation is a no-op; it is the responsibility of the implementing class to override this and document any expectations for + * when it must be invoked. + */ + default void claimWritePrivileges() { + } + /** * Set an update listener to get notifications when there are config/target state * changes. diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaConfigBackingStore.java b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaConfigBackingStore.java index 94b98cb9eb38d..d17864458bf69 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaConfigBackingStore.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaConfigBackingStore.java @@ -19,9 +19,14 @@ import org.apache.kafka.clients.admin.NewTopic; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.producer.KafkaProducer; +import org.apache.kafka.clients.producer.Producer; import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.common.IsolationLevel; import org.apache.kafka.common.config.ConfigException; import org.apache.kafka.common.config.TopicConfig; +import org.apache.kafka.common.errors.UnsupportedVersionException; import org.apache.kafka.common.serialization.ByteArrayDeserializer; import org.apache.kafka.common.serialization.ByteArraySerializer; import org.apache.kafka.common.serialization.StringDeserializer; @@ -39,7 +44,6 @@ import org.apache.kafka.connect.runtime.TargetState; import org.apache.kafka.connect.runtime.WorkerConfig; import org.apache.kafka.connect.runtime.WorkerConfigTransformer; -import org.apache.kafka.connect.runtime.distributed.ClusterConfigState; import org.apache.kafka.connect.runtime.distributed.DistributedConfig; import org.apache.kafka.connect.util.Callback; import org.apache.kafka.connect.util.ConnectUtils; @@ -51,6 +55,7 @@ import org.slf4j.LoggerFactory; import javax.crypto.spec.SecretKeySpec; +import java.time.Duration; import java.util.ArrayList; import java.util.Arrays; import java.util.Base64; @@ -58,6 +63,7 @@ import java.util.HashMap; import java.util.HashSet; import java.util.List; +import java.util.Locale; import java.util.Map; import java.util.Set; import java.util.TreeSet; @@ -233,21 +239,23 @@ public static String RESTART_KEY(String connectorName) { // initialization as long as we always read the volatile variable "started" before we access the listener. private UpdateListener updateListener; + private final Map baseProducerProps; + private final String topic; // Data is passed to the log already serialized. We use a converter to handle translating to/from generic Connect // format to serialized form private final KafkaBasedLog configLog; // Connector -> # of tasks - private final Map connectorTaskCounts = new HashMap<>(); + final Map connectorTaskCounts = new HashMap<>(); // Connector and task configs: name or id -> config map - private final Map> connectorConfigs = new HashMap<>(); - private final Map> taskConfigs = new HashMap<>(); + final Map> connectorConfigs = new HashMap<>(); + final Map> taskConfigs = new HashMap<>(); private final Supplier topicAdminSupplier; private SharedTopicAdmin ownTopicAdmin; // Set of connectors where we saw a task commit with an incomplete set of task config updates, indicating the data // is in an inconsistent state and we cannot safely use them until they have been refreshed. - private final Set inconsistent = new HashSet<>(); + final Set inconsistent = new HashSet<>(); // The most recently read offset. This does not take into account deferred task updates/commits, so we may have // outstanding data to be applied. private volatile long offset; @@ -257,22 +265,37 @@ public static String RESTART_KEY(String connectorName) { // Connector -> Map[ConnectorTaskId -> Configs] private final Map>> deferredTaskUpdates = new HashMap<>(); - private final Map connectorTargetStates = new HashMap<>(); + final Map connectorTargetStates = new HashMap<>(); private final WorkerConfigTransformer configTransformer; + private final boolean usesFencableWriter; + private volatile Producer fencableProducer; + private final Map fencableProducerProps; + @Deprecated - public KafkaConfigBackingStore(Converter converter, WorkerConfig config, WorkerConfigTransformer configTransformer) { + public KafkaConfigBackingStore(Converter converter, DistributedConfig config, WorkerConfigTransformer configTransformer) { this(converter, config, configTransformer, null); } - public KafkaConfigBackingStore(Converter converter, WorkerConfig config, WorkerConfigTransformer configTransformer, Supplier adminSupplier) { + public KafkaConfigBackingStore(Converter converter, DistributedConfig config, WorkerConfigTransformer configTransformer, Supplier adminSupplier) { this.lock = new Object(); this.started = false; this.converter = converter; this.offset = -1; this.topicAdminSupplier = adminSupplier; + this.baseProducerProps = baseProducerProps(config); + // By default, Connect disables idempotent behavior for all producers, even though idempotence became + // default for Kafka producers. This is to ensure Connect continues to work with many Kafka broker versions, including older brokers that do not support + // idempotent producers or require explicit steps to enable them (e.g. adding the IDEMPOTENT_WRITE ACL to brokers older than 2.8). + // These settings might change when https://cwiki.apache.org/confluence/display/KAFKA/KIP-318%3A+Make+Kafka+Connect+Source+idempotent + // gets approved and scheduled for release. + baseProducerProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "false"); + + this.fencableProducerProps = fencableProducerProps(config); + + this.usesFencableWriter = config.transactionalLeaderEnabled(); this.topic = config.getString(DistributedConfig.CONFIG_TOPIC_CONFIG); if (this.topic == null || this.topic.trim().length() == 0) throw new ConfigException("Must specify topic for connector configuration."); @@ -291,7 +314,16 @@ public void start() { log.info("Starting KafkaConfigBackingStore"); // Before startup, callbacks are *not* invoked. You can grab a snapshot after starting -- just take care that // updates can continue to occur in the background - configLog.start(); + try { + configLog.start(); + } catch (UnsupportedVersionException e) { + throw new ConnectException( + "Enabling exactly-once support for source connectors requires a Kafka broker version that allows " + + "admin clients to read consumer offsets. Please either disable the worker's exactly-once " + + "support for source connectors, or use a newer Kafka broker version.", + e + ); + } int partitionCount = configLog.partitionCount(); if (partitionCount > 1) { @@ -309,16 +341,67 @@ public void start() { @Override public void stop() { log.info("Closing KafkaConfigBackingStore"); - try { - configLog.stop(); - } finally { - if (ownTopicAdmin != null) { - ownTopicAdmin.close(); - } + + if (fencableProducer != null) { + Utils.closeQuietly(() -> fencableProducer.close(Duration.ZERO), "fencable producer for config topic"); } + Utils.closeQuietly(ownTopicAdmin, "admin for config topic"); + Utils.closeQuietly(configLog::stop, "KafkaBasedLog for config topic"); + log.info("Closed KafkaConfigBackingStore"); } + @Override + public void claimWritePrivileges() { + if (usesFencableWriter && fencableProducer == null) { + try { + fencableProducer = createFencableProducer(); + fencableProducer.initTransactions(); + } catch (Exception e) { + relinquishWritePrivileges(); + throw new ConnectException("Failed to create and initialize fencable producer for config topic", e); + } + } + } + + private Map baseProducerProps(WorkerConfig workerConfig) { + Map producerProps = new HashMap<>(workerConfig.originals()); + String kafkaClusterId = ConnectUtils.lookupKafkaClusterId(workerConfig); + producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); + producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName()); + producerProps.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, Integer.MAX_VALUE); + ConnectUtils.addMetricsContextProperties(producerProps, workerConfig, kafkaClusterId); + return producerProps; + } + + // Visible for testing + Map fencableProducerProps(DistributedConfig workerConfig) { + Map result = new HashMap<>(baseProducerProps(workerConfig)); + + // Always require producer acks to all to ensure durable writes + result.put(ProducerConfig.ACKS_CONFIG, "all"); + // We can set this to 5 instead of 1 without risking reordering because we are using an idempotent producer + result.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5); + + ConnectUtils.ensureProperty( + result, ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true", + "for the worker's config topic producer when exactly-once source support is enabled or in preparation to be enabled", + false + ); + ConnectUtils.ensureProperty( + result, ProducerConfig.TRANSACTIONAL_ID_CONFIG, workerConfig.transactionalProducerId(), + "for the worker's config topic producer when exactly-once source support is enabled or in preparation to be enabled", + true + ); + + return result; + } + + // Visible in order to be mocked during testing + Producer createFencableProducer() { + return new KafkaProducer<>(fencableProducerProps); + } + /** * Get a snapshot of the current state of the cluster. */ @@ -349,10 +432,15 @@ public boolean contains(String connector) { /** * Write this connector configuration to persistent storage and wait until it has been acknowledged and read back by - * tailing the Kafka log with a consumer. + * tailing the Kafka log with a consumer. {@link #claimWritePrivileges()} must be successfully invoked before calling + * this method if the worker is configured to use a fencable producer for writes to the config topic. * * @param connector name of the connector to write data for * @param properties the configuration to write + * @throws IllegalStateException if {@link #claimWritePrivileges()} is required, but was not successfully invoked before + * this method was called + * @throws PrivilegedWriteException if the worker is configured to use a fencable producer for writes to the config topic + * and the write fails */ @Override public void putConnectorConfig(String connector, Map properties) { @@ -360,19 +448,30 @@ public void putConnectorConfig(String connector, Map properties) Struct connectConfig = new Struct(CONNECTOR_CONFIGURATION_V0); connectConfig.put("properties", properties); byte[] serializedConfig = converter.fromConnectData(topic, CONNECTOR_CONFIGURATION_V0, connectConfig); - updateConnectorConfig(connector, serializedConfig); + try { + maybeSendFencably(CONNECTOR_KEY(connector), serializedConfig); + configLog.readToEnd().get(READ_TO_END_TIMEOUT_MS, TimeUnit.MILLISECONDS); + } catch (InterruptedException | ExecutionException | TimeoutException e) { + log.error("Failed to write connector configuration to Kafka: ", e); + throw new ConnectException("Error writing connector configuration to Kafka", e); + } } /** - * Remove configuration for a given connector. + * Remove configuration for a given connector. {@link #claimWritePrivileges()} must be successfully invoked before calling + * this method if the worker is configured to use a fencable producer for writes to the config topic. * @param connector name of the connector to remove + * @throws IllegalStateException if {@link #claimWritePrivileges()} is required, but was not successfully invoked before + * this method was called + * @throws PrivilegedWriteException if the worker is configured to use a fencable producer for writes to the config topic + * and the write fails */ @Override public void removeConnectorConfig(String connector) { log.debug("Removing connector configuration for connector '{}'", connector); try { - configLog.send(CONNECTOR_KEY(connector), null); - configLog.send(TARGET_STATE_KEY(connector), null); + maybeSendFencably(CONNECTOR_KEY(connector), null); + maybeSendFencably(TARGET_STATE_KEY(connector), null); configLog.readToEnd().get(READ_TO_END_TIMEOUT_MS, TimeUnit.MILLISECONDS); } catch (InterruptedException | ExecutionException | TimeoutException e) { log.error("Failed to remove connector configuration from Kafka: ", e); @@ -385,24 +484,20 @@ public void removeTaskConfigs(String connector) { throw new UnsupportedOperationException("Removal of tasks is not currently supported"); } - private void updateConnectorConfig(String connector, byte[] serializedConfig) { - try { - configLog.send(CONNECTOR_KEY(connector), serializedConfig); - configLog.readToEnd().get(READ_TO_END_TIMEOUT_MS, TimeUnit.MILLISECONDS); - } catch (InterruptedException | ExecutionException | TimeoutException e) { - log.error("Failed to write connector configuration to Kafka: ", e); - throw new ConnectException("Error writing connector configuration to Kafka", e); - } - } - /** * Write these task configurations and associated commit messages, unless an inconsistency is found that indicates - * that we would be leaving one of the referenced connectors with an inconsistent state. + * that we would be leaving one of the referenced connectors with an inconsistent state. {@link #claimWritePrivileges()} + * must be successfully invoked before calling this method if the worker is configured to use a fencable producer for + * writes to the config topic. * * @param connector the connector to write task configuration * @param configs list of task configurations for the connector * @throws ConnectException if the task configurations do not resolve inconsistencies found in the existing root * and task configurations. + * @throws IllegalStateException if {@link #claimWritePrivileges()} is required, but was not successfully invoked before + * this method was called + * @throws PrivilegedWriteException if the worker is configured to use a fencable producer for writes to the config topic + * and the write fails */ @Override public void putTaskConfigs(String connector, List> configs) { @@ -425,7 +520,7 @@ public void putTaskConfigs(String connector, List> configs) byte[] serializedConfig = converter.fromConnectData(topic, TASK_CONFIGURATION_V0, connectConfig); log.debug("Writing configuration for connector '{}' task {}", connector, index); ConnectorTaskId connectorTaskId = new ConnectorTaskId(connector, index); - configLog.send(TASK_KEY(connectorTaskId), serializedConfig); + maybeSendFencably(TASK_KEY(connectorTaskId), serializedConfig); index++; } @@ -441,7 +536,7 @@ public void putTaskConfigs(String connector, List> configs) connectConfig.put("tasks", taskCount); byte[] serializedConfig = converter.fromConnectData(topic, CONNECTOR_TASKS_COMMIT_V0, connectConfig); log.debug("Writing commit for connector '{}' with {} tasks.", connector, taskCount); - configLog.send(COMMIT_TASKS_KEY(connector), serializedConfig); + maybeSendFencably(COMMIT_TASKS_KEY(connector), serializedConfig); // Read to end to ensure all the commit messages have been written configLog.readToEnd().get(READ_TO_END_TIMEOUT_MS, TimeUnit.MILLISECONDS); @@ -460,6 +555,12 @@ public void refresh(long timeout, TimeUnit unit) throws TimeoutException { } } + /** + * Write a new {@link TargetState} for the connector. Note that {@link #claimWritePrivileges()} does not need to be + * invoked before invoking this method. + * @param connector the name of the connector + * @param state the desired target state for the connector + */ @Override public void putTargetState(String connector, TargetState state) { Struct connectTargetState = new Struct(TARGET_STATE_V0); @@ -469,6 +570,16 @@ public void putTargetState(String connector, TargetState state) { configLog.send(TARGET_STATE_KEY(connector), serializedTargetState); } + /** + * Write a session key to persistent storage and wait until it has been acknowledged and read back by tailing the Kafka log + * with a consumer. {@link #claimWritePrivileges()} must be successfully invoked before calling this method if the worker + * is configured to use a fencable producer for writes to the config topic. + * @param sessionKey the session key to distributed + * @throws IllegalStateException if {@link #claimWritePrivileges()} is required, but was not successfully invoked before + * this method was called + * @throws PrivilegedWriteException if the worker is configured to use a fencable producer for writes to the config topic + * and the write fails + */ @Override public void putSessionKey(SessionKey sessionKey) { log.debug("Distributing new session key"); @@ -478,7 +589,7 @@ public void putSessionKey(SessionKey sessionKey) { sessionKeyStruct.put("creation-timestamp", sessionKey.creationTimestamp()); byte[] serializedSessionKey = converter.fromConnectData(topic, SESSION_KEY_V0, sessionKeyStruct); try { - configLog.send(SESSION_KEY_KEY, serializedSessionKey); + maybeSendFencably(SESSION_KEY_KEY, serializedSessionKey); configLog.readToEnd().get(READ_TO_END_TIMEOUT_MS, TimeUnit.MILLISECONDS); } catch (InterruptedException | ExecutionException | TimeoutException e) { log.error("Failed to write session key to Kafka: ", e); @@ -486,6 +597,12 @@ public void putSessionKey(SessionKey sessionKey) { } } + /** + * Write a restart request for the connector and optionally its tasks to persistent storage and wait until it has been + * acknowledged and read back by tailing the Kafka log with a consumer. {@link #claimWritePrivileges()} must be successfully + * invoked before calling this method if the worker is configured to use a fencable producer for writes to the config topic. + * @param restartRequest the restart request details + */ @Override public void putRestartRequest(RestartRequest restartRequest) { log.debug("Writing {} to Kafka", restartRequest); @@ -495,7 +612,7 @@ public void putRestartRequest(RestartRequest restartRequest) { value.put(ONLY_FAILED_FIELD_NAME, restartRequest.onlyFailed()); byte[] serializedValue = converter.fromConnectData(topic, value.schema(), value); try { - configLog.send(key, serializedValue); + maybeSendFencably(key, serializedValue); configLog.readToEnd().get(READ_TO_END_TIMEOUT_MS, TimeUnit.MILLISECONDS); } catch (InterruptedException | ExecutionException | TimeoutException e) { log.error("Failed to write {} to Kafka: ", restartRequest, e); @@ -505,24 +622,22 @@ public void putRestartRequest(RestartRequest restartRequest) { // package private for testing KafkaBasedLog setupAndCreateKafkaBasedLog(String topic, final WorkerConfig config) { + Map producerProps = new HashMap<>(baseProducerProps); + String clusterId = ConnectUtils.lookupKafkaClusterId(config); Map originals = config.originals(); - Map producerProps = new HashMap<>(originals); - producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); - producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName()); - producerProps.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, Integer.MAX_VALUE); - // By default, Connect disables idempotent behavior for all producers, even though idempotence became - // default for Kafka producers. This is to ensure Connect continues to work with many Kafka broker versions, including older brokers that do not support - // idempotent producers or require explicit steps to enable them (e.g. adding the IDEMPOTENT_WRITE ACL to brokers older than 2.8). - // These settings might change when https://cwiki.apache.org/confluence/display/KAFKA/KIP-318%3A+Make+Kafka+Connect+Source+idempotent - // gets approved and scheduled for release. - producerProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "false"); - ConnectUtils.addMetricsContextProperties(producerProps, config, clusterId); Map consumerProps = new HashMap<>(originals); consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName()); ConnectUtils.addMetricsContextProperties(consumerProps, config, clusterId); + if (config.exactlyOnceSourceEnabled()) { + ConnectUtils.ensureProperty( + consumerProps, ConsumerConfig.ISOLATION_LEVEL_CONFIG, IsolationLevel.READ_COMMITTED.name().toLowerCase(Locale.ROOT), + "for the worker's config topic consumer when exactly-once source support is enabled", + true + ); + } Map adminProps = new HashMap<>(originals); ConnectUtils.addMetricsContextProperties(adminProps, config, clusterId); @@ -547,6 +662,34 @@ KafkaBasedLog setupAndCreateKafkaBasedLog(String topic, final Wo return createKafkaBasedLog(topic, producerProps, consumerProps, new ConsumeCallback(), topicDescription, adminSupplier); } + private void maybeSendFencably(String key, byte[] value) { + if (!usesFencableWriter) { + configLog.send(key, value); + return; + } + + if (fencableProducer == null) { + throw new IllegalStateException("Cannot produce to config topic without claiming write privileges first"); + } + + try { + fencableProducer.beginTransaction(); + fencableProducer.send(new ProducerRecord<>(topic, key, value)); + fencableProducer.commitTransaction(); + } catch (Exception e) { + log.warn("Failed to perform fencable send to config topic", e); + relinquishWritePrivileges(); + throw new PrivilegedWriteException("Failed to perform fencable send to config topic", e); + } + } + + private void relinquishWritePrivileges() { + if (fencableProducer != null) { + Utils.closeQuietly(() -> fencableProducer.close(Duration.ZERO), "fencable producer for config topic"); + fencableProducer = null; + } + } + private KafkaBasedLog createKafkaBasedLog(String topic, Map producerProps, Map consumerProps, Callback> consumedCallback, @@ -606,7 +749,7 @@ public void onCompletion(Throwable error, ConsumerRecord record) } Object targetState = ((Map) value.value()).get("state"); if (!(targetState instanceof String)) { - log.error("Invalid data for target state for connector '{}': 'state' field should be a Map but is {}", + log.error("Invalid data for target state for connector '{}': 'state' field should be a String but is {}", connectorName, targetState == null ? null : targetState.getClass()); return; } diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/MemoryConfigBackingStore.java b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/MemoryConfigBackingStore.java index a8b2820404b71..b8f19329ee9b3 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/MemoryConfigBackingStore.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/MemoryConfigBackingStore.java @@ -20,7 +20,6 @@ import org.apache.kafka.connect.runtime.SessionKey; import org.apache.kafka.connect.runtime.TargetState; import org.apache.kafka.connect.runtime.WorkerConfigTransformer; -import org.apache.kafka.connect.runtime.distributed.ClusterConfigState; import org.apache.kafka.connect.util.ConnectorTaskId; import java.util.Collections; diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/PrivilegedWriteException.java b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/PrivilegedWriteException.java new file mode 100644 index 0000000000000..e4900fa9b0e59 --- /dev/null +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/PrivilegedWriteException.java @@ -0,0 +1,32 @@ +/* + * 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.connect.storage; + +import org.apache.kafka.connect.errors.ConnectException; + +/** + * Used when a write that requires {@link ConfigBackingStore#claimWritePrivileges() special privileges} fails + */ +public class PrivilegedWriteException extends ConnectException { + public PrivilegedWriteException(String message) { + super(message); + } + + public PrivilegedWriteException(String message, Throwable cause) { + super(message, cause); + } +} 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 862f31c077087..1f5acb412c52f 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 @@ -31,7 +31,6 @@ import org.apache.kafka.connect.connector.policy.NoneConnectorClientConfigOverridePolicy; import org.apache.kafka.connect.connector.policy.PrincipalConnectorClientConfigOverridePolicy; import org.apache.kafka.connect.errors.NotFoundException; -import org.apache.kafka.connect.runtime.distributed.ClusterConfigState; import org.apache.kafka.connect.runtime.isolation.PluginDesc; import org.apache.kafka.connect.runtime.isolation.Plugins; import org.apache.kafka.connect.runtime.rest.entities.ConfigInfo; @@ -43,6 +42,7 @@ import org.apache.kafka.connect.runtime.rest.errors.BadRequestException; import org.apache.kafka.connect.source.SourceConnector; import org.apache.kafka.connect.source.SourceTask; +import org.apache.kafka.connect.storage.ClusterConfigState; import org.apache.kafka.connect.storage.ConfigBackingStore; import org.apache.kafka.connect.storage.StatusBackingStore; import org.apache.kafka.connect.transforms.Transformation; @@ -73,16 +73,16 @@ import static org.apache.kafka.connect.runtime.AbstractHerder.keysWithVariableValues; import static org.easymock.EasyMock.anyString; -import static org.junit.Assert.assertNull; -import static org.junit.Assert.assertThrows; -import static org.powermock.api.easymock.PowerMock.verifyAll; -import static org.powermock.api.easymock.PowerMock.replayAll; -import static org.easymock.EasyMock.strictMock; import static org.easymock.EasyMock.partialMockBuilder; +import static org.easymock.EasyMock.strictMock; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; +import static org.powermock.api.easymock.PowerMock.replayAll; +import static org.powermock.api.easymock.PowerMock.verifyAll; @RunWith(PowerMockRunner.class) @PrepareForTest({AbstractHerder.class}) diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/ErrorHandlingTaskTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/ErrorHandlingTaskTest.java index a63430d6457dc..e222894945fb1 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/ErrorHandlingTaskTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/ErrorHandlingTaskTest.java @@ -33,7 +33,7 @@ import org.apache.kafka.connect.errors.RetriableException; import org.apache.kafka.connect.integration.MonitorableSourceConnector; import org.apache.kafka.connect.json.JsonConverter; -import org.apache.kafka.connect.runtime.distributed.ClusterConfigState; +import org.apache.kafka.connect.storage.ClusterConfigState; import org.apache.kafka.connect.runtime.errors.ErrorHandlingMetrics; import org.apache.kafka.connect.runtime.errors.ErrorReporter; import org.apache.kafka.connect.runtime.errors.LogReporter; diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskTest.java index 65ab0c7e7cbbb..41d4de5e0a86e 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskTest.java @@ -39,7 +39,7 @@ import org.apache.kafka.connect.errors.ConnectException; import org.apache.kafka.connect.errors.RetriableException; import org.apache.kafka.connect.runtime.ConnectMetrics.MetricGroup; -import org.apache.kafka.connect.runtime.distributed.ClusterConfigState; +import org.apache.kafka.connect.storage.ClusterConfigState; import org.apache.kafka.connect.runtime.WorkerSinkTask.SinkTaskMetricsGroup; import org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperatorTest; import org.apache.kafka.connect.runtime.isolation.PluginClassLoader; diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskThreadedTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskThreadedTest.java index a0c99fb47145f..eef6ca82be63e 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskThreadedTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskThreadedTest.java @@ -29,7 +29,7 @@ import org.apache.kafka.connect.data.Schema; import org.apache.kafka.connect.data.SchemaAndValue; import org.apache.kafka.connect.errors.ConnectException; -import org.apache.kafka.connect.runtime.distributed.ClusterConfigState; +import org.apache.kafka.connect.storage.ClusterConfigState; import org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperatorTest; import org.apache.kafka.connect.runtime.isolation.PluginClassLoader; import org.apache.kafka.connect.runtime.standalone.StandaloneConfig; diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskTest.java index 78db83c7ee3c6..5aeff5e9d87e8 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskTest.java @@ -41,7 +41,7 @@ import org.apache.kafka.connect.integration.MonitorableSourceConnector; import org.apache.kafka.connect.runtime.ConnectMetrics.MetricGroup; import org.apache.kafka.connect.runtime.WorkerSourceTask.SourceTaskMetricsGroup; -import org.apache.kafka.connect.runtime.distributed.ClusterConfigState; +import org.apache.kafka.connect.storage.ClusterConfigState; import org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator; import org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperatorTest; import org.apache.kafka.connect.runtime.isolation.Plugins; diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerTest.java index e410183c7f806..36cec9205861f 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerTest.java @@ -36,7 +36,7 @@ import org.apache.kafka.connect.json.JsonConverter; import org.apache.kafka.connect.runtime.ConnectMetrics.MetricGroup; import org.apache.kafka.connect.runtime.MockConnectMetrics.MockMetricsReporter; -import org.apache.kafka.connect.runtime.distributed.ClusterConfigState; +import org.apache.kafka.connect.storage.ClusterConfigState; import org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader; import org.apache.kafka.connect.runtime.isolation.PluginClassLoader; import org.apache.kafka.connect.runtime.isolation.Plugins; diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerTestUtils.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerTestUtils.java index ed77018f2883b..544507ab64739 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerTestUtils.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerTestUtils.java @@ -16,7 +16,7 @@ */ package org.apache.kafka.connect.runtime; -import org.apache.kafka.connect.runtime.distributed.ClusterConfigState; +import org.apache.kafka.connect.storage.ClusterConfigState; import org.apache.kafka.connect.runtime.distributed.ExtendedAssignment; import org.apache.kafka.connect.runtime.distributed.ExtendedWorkerState; import org.apache.kafka.connect.util.ConnectorTaskId; diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedHerderTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedHerderTest.java index 54dacd6d82362..8f3a8cf3eb418 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedHerderTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedHerderTest.java @@ -56,6 +56,7 @@ import org.apache.kafka.connect.source.ExactlyOnceSupport; import org.apache.kafka.connect.source.SourceConnector; import org.apache.kafka.connect.source.SourceTask; +import org.apache.kafka.connect.storage.ClusterConfigState; import org.apache.kafka.connect.storage.ConfigBackingStore; import org.apache.kafka.connect.storage.StatusBackingStore; import org.apache.kafka.connect.util.Callback; @@ -687,7 +688,7 @@ public void testHaltCleansUpWorker() { public void testCreateConnector() throws Exception { EasyMock.expect(member.memberId()).andStubReturn("leader"); EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0); - expectRebalance(1, Collections.emptyList(), Collections.emptyList()); + expectRebalance(1, Collections.emptyList(), Collections.emptyList(), true); expectPostRebalanceCatchup(SNAPSHOT); member.wakeup(); @@ -742,7 +743,7 @@ public void testCreateConnector() throws Exception { public void testCreateConnectorFailedValidation() throws Exception { EasyMock.expect(member.memberId()).andStubReturn("leader"); EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0); - expectRebalance(1, Collections.emptyList(), Collections.emptyList()); + expectRebalance(1, Collections.emptyList(), Collections.emptyList(), true); expectPostRebalanceCatchup(SNAPSHOT); HashMap config = new HashMap<>(CONN2_CONFIG); @@ -1065,7 +1066,7 @@ public void testCreateConnectorAlreadyExists() throws Exception { return null; }); - expectRebalance(1, Collections.emptyList(), Collections.emptyList()); + expectRebalance(1, Collections.emptyList(), Collections.emptyList(), true); expectPostRebalanceCatchup(SNAPSHOT); member.wakeup(); @@ -1104,7 +1105,7 @@ public void testDestroyConnector() throws Exception { EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0); // Start with one connector EasyMock.expect(worker.getPlugins()).andReturn(plugins); - expectRebalance(1, Arrays.asList(CONN1), Collections.emptyList()); + expectRebalance(1, Arrays.asList(CONN1), Collections.emptyList(), true); expectPostRebalanceCatchup(SNAPSHOT); Capture> onStart = newCapture(); worker.startConnector(EasyMock.eq(CONN1), EasyMock.anyObject(), EasyMock.anyObject(), @@ -1138,8 +1139,8 @@ public void testDestroyConnector() throws Exception { statusBackingStore.deleteTopic(EasyMock.eq(CONN1), EasyMock.eq(BAR_TOPIC)); PowerMock.expectLastCall().times(2); expectRebalance(Arrays.asList(CONN1), Arrays.asList(TASK1), - ConnectProtocol.Assignment.NO_ERROR, 2, - Collections.emptyList(), Collections.emptyList(), 0); + ConnectProtocol.Assignment.NO_ERROR, 2, "leader", "leaderUrl", + Collections.emptyList(), Collections.emptyList(), 0, true); expectPostRebalanceCatchup(ClusterConfigState.EMPTY); member.requestRejoin(); PowerMock.expectLastCall(); @@ -1168,7 +1169,7 @@ public void testRestartConnector() throws Exception { EasyMock.expect(member.memberId()).andStubReturn("leader"); EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0); EasyMock.expect(worker.getPlugins()).andReturn(plugins); - expectRebalance(1, singletonList(CONN1), Collections.emptyList()); + expectRebalance(1, singletonList(CONN1), Collections.emptyList(), true); expectPostRebalanceCatchup(SNAPSHOT); member.poll(EasyMock.anyInt()); PowerMock.expectLastCall(); @@ -1221,7 +1222,7 @@ public void testRestartUnknownConnector() throws Exception { // get the initial assignment EasyMock.expect(member.memberId()).andStubReturn("leader"); EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0); - expectRebalance(1, Collections.emptyList(), Collections.emptyList()); + expectRebalance(1, Collections.emptyList(), Collections.emptyList(), true); expectPostRebalanceCatchup(SNAPSHOT); member.poll(EasyMock.anyInt()); PowerMock.expectLastCall(); @@ -1290,7 +1291,7 @@ public void testRestartConnectorRedirectToOwner() throws Exception { // get the initial assignment EasyMock.expect(member.memberId()).andStubReturn("leader"); EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0); - expectRebalance(1, Collections.emptyList(), Collections.emptyList()); + expectRebalance(1, Collections.emptyList(), Collections.emptyList(), true); expectPostRebalanceCatchup(SNAPSHOT); member.poll(EasyMock.anyInt()); PowerMock.expectLastCall(); @@ -1339,7 +1340,7 @@ public void testRestartConnectorAndTasksUnknownConnector() throws Exception { // get the initial assignment EasyMock.expect(member.memberId()).andStubReturn("leader"); EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0); - expectRebalance(1, Collections.emptyList(), Collections.emptyList()); + expectRebalance(1, Collections.emptyList(), Collections.emptyList(), true); expectPostRebalanceCatchup(SNAPSHOT); member.poll(EasyMock.anyInt()); PowerMock.expectLastCall(); @@ -1408,7 +1409,7 @@ public void testRestartConnectorAndTasksUnknownStatus() throws Exception { // get the initial assignment EasyMock.expect(member.memberId()).andStubReturn("leader"); EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0); - expectRebalance(1, Collections.emptyList(), Collections.emptyList()); + expectRebalance(1, Collections.emptyList(), Collections.emptyList(), true); expectPostRebalanceCatchup(SNAPSHOT); member.poll(EasyMock.anyInt()); PowerMock.expectLastCall(); @@ -1448,7 +1449,7 @@ public void testRestartConnectorAndTasksSuccess() throws Exception { // get the initial assignment EasyMock.expect(member.memberId()).andStubReturn("leader"); EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0); - expectRebalance(1, Collections.emptyList(), Collections.emptyList()); + expectRebalance(1, Collections.emptyList(), Collections.emptyList(), true); expectPostRebalanceCatchup(SNAPSHOT); member.poll(EasyMock.anyInt()); PowerMock.expectLastCall(); @@ -1611,7 +1612,7 @@ public void testRestartTask() throws Exception { // get the initial assignment EasyMock.expect(member.memberId()).andStubReturn("leader"); EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0); - expectRebalance(1, Collections.emptyList(), singletonList(TASK0)); + expectRebalance(1, Collections.emptyList(), singletonList(TASK0), true); expectPostRebalanceCatchup(SNAPSHOT); member.poll(EasyMock.anyInt()); PowerMock.expectLastCall(); @@ -1731,7 +1732,7 @@ public void testRestartTaskRedirectToOwner() throws Exception { // get the initial assignment EasyMock.expect(member.memberId()).andStubReturn("leader"); EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0); - expectRebalance(1, Collections.emptyList(), Collections.emptyList()); + expectRebalance(1, Collections.emptyList(), Collections.emptyList(), true); expectPostRebalanceCatchup(SNAPSHOT); member.poll(EasyMock.anyInt()); PowerMock.expectLastCall(); @@ -2193,8 +2194,8 @@ public void testJoinLeaderCatchUpFails() throws Exception { EasyMock.expect(member.memberId()).andStubReturn("leader"); EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0); expectRebalance(Collections.emptyList(), Collections.emptyList(), - ConnectProtocol.Assignment.CONFIG_MISMATCH, 1, Collections.emptyList(), - Collections.emptyList()); + ConnectProtocol.Assignment.CONFIG_MISMATCH, 1, "leader", "leaderUrl", Collections.emptyList(), + Collections.emptyList(), 0, true); // Reading to end of log times out configBackingStore.refresh(EasyMock.anyLong(), EasyMock.anyObject(TimeUnit.class)); EasyMock.expectLastCall().andThrow(new TimeoutException()); @@ -2203,7 +2204,7 @@ public void testJoinLeaderCatchUpFails() throws Exception { member.requestRejoin(); // After backoff, restart the process and this time succeed - expectRebalance(1, Arrays.asList(CONN1), Arrays.asList(TASK1)); + expectRebalance(1, Arrays.asList(CONN1), Arrays.asList(TASK1), true); expectPostRebalanceCatchup(SNAPSHOT); EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0); @@ -2226,7 +2227,7 @@ public void testJoinLeaderCatchUpFails() throws Exception { PowerMock.expectLastCall(); // one more tick, to make sure we don't keep trying to read to the config topic unnecessarily - expectRebalance(1, Collections.emptyList(), Collections.emptyList()); + expectRebalance(1, Collections.emptyList(), Collections.emptyList(), true); member.poll(EasyMock.anyInt()); PowerMock.expectLastCall(); @@ -2261,7 +2262,7 @@ public void testJoinLeaderCatchUpRetriesForIncrementalCooperative() throws Excep // Join group and as leader fail to do assignment EasyMock.expect(member.memberId()).andStubReturn("leader"); EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V1); - expectRebalance(1, Arrays.asList(CONN1), Arrays.asList(TASK1)); + expectRebalance(1, Arrays.asList(CONN1), Arrays.asList(TASK1), true); expectPostRebalanceCatchup(SNAPSHOT); member.poll(EasyMock.anyInt()); @@ -2270,7 +2271,7 @@ public void testJoinLeaderCatchUpRetriesForIncrementalCooperative() throws Excep // The leader got its assignment expectRebalance(Collections.emptyList(), Collections.emptyList(), ConnectProtocol.Assignment.NO_ERROR, - 1, Arrays.asList(CONN1), Arrays.asList(TASK1), 0); + 1, "leader", "leaderUrl", Arrays.asList(CONN1), Arrays.asList(TASK1), 0, true); EasyMock.expect(worker.getPlugins()).andReturn(plugins); Capture> onStart = newCapture(); @@ -2294,8 +2295,8 @@ public void testJoinLeaderCatchUpRetriesForIncrementalCooperative() throws Excep // Another rebalance is triggered but this time it fails to read to the max offset and // triggers a re-sync expectRebalance(Collections.emptyList(), Collections.emptyList(), - ConnectProtocol.Assignment.CONFIG_MISMATCH, 1, Collections.emptyList(), - Collections.emptyList()); + ConnectProtocol.Assignment.CONFIG_MISMATCH, 1, "leader", "leaderUrl", + Collections.emptyList(), Collections.emptyList(), 0, true); // The leader will retry a few times to read to the end of the config log int retries = 2; @@ -2311,7 +2312,7 @@ public void testJoinLeaderCatchUpRetriesForIncrementalCooperative() throws Excep // After a few retries succeed to read the log to the end expectRebalance(Collections.emptyList(), Collections.emptyList(), ConnectProtocol.Assignment.NO_ERROR, - 1, Arrays.asList(CONN1), Arrays.asList(TASK1), 0); + 1, "leader", "leaderUrl", Arrays.asList(CONN1), Arrays.asList(TASK1), 0, true); expectPostRebalanceCatchup(SNAPSHOT); member.poll(EasyMock.anyInt()); PowerMock.expectLastCall(); @@ -2352,7 +2353,7 @@ public void testJoinLeaderCatchUpFailsForIncrementalCooperative() throws Excepti // Join group and as leader fail to do assignment EasyMock.expect(member.memberId()).andStubReturn("leader"); EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V1); - expectRebalance(1, Arrays.asList(CONN1), Arrays.asList(TASK1)); + expectRebalance(1, Arrays.asList(CONN1), Arrays.asList(TASK1), true); expectPostRebalanceCatchup(SNAPSHOT); member.poll(EasyMock.anyInt()); @@ -2360,8 +2361,8 @@ public void testJoinLeaderCatchUpFailsForIncrementalCooperative() throws Excepti // The leader got its assignment expectRebalance(Collections.emptyList(), Collections.emptyList(), - ConnectProtocol.Assignment.NO_ERROR, - 1, Arrays.asList(CONN1), Arrays.asList(TASK1), 0); + ConnectProtocol.Assignment.NO_ERROR, 1, + "leader", "leaderUrl", Arrays.asList(CONN1), Arrays.asList(TASK1), 0, true); EasyMock.expect(worker.getPlugins()).andReturn(plugins); // and the new assignment started @@ -2386,8 +2387,8 @@ public void testJoinLeaderCatchUpFailsForIncrementalCooperative() throws Excepti // Another rebalance is triggered but this time it fails to read to the max offset and // triggers a re-sync expectRebalance(Collections.emptyList(), Collections.emptyList(), - ConnectProtocol.Assignment.CONFIG_MISMATCH, 1, Collections.emptyList(), - Collections.emptyList()); + ConnectProtocol.Assignment.CONFIG_MISMATCH, 1, "leader", "leaderUrl", + Collections.emptyList(), Collections.emptyList(), 0, true); // The leader will exhaust the retries while trying to read to the end of the config log int maxRetries = 5; @@ -2408,7 +2409,8 @@ public void testJoinLeaderCatchUpFailsForIncrementalCooperative() throws Excepti // The worker gets back the assignment that had given up expectRebalance(Collections.emptyList(), Collections.emptyList(), ConnectProtocol.Assignment.NO_ERROR, - 1, Arrays.asList(CONN1), Arrays.asList(TASK1), 0); + 1, "leader", "leaderUrl", Arrays.asList(CONN1), Arrays.asList(TASK1), + 0, true); expectPostRebalanceCatchup(SNAPSHOT); member.poll(EasyMock.anyInt()); PowerMock.expectLastCall(); @@ -2448,7 +2450,7 @@ public void testAccessors() throws Exception { EasyMock.expect(member.memberId()).andStubReturn("leader"); EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0); EasyMock.expect(worker.getPlugins()).andReturn(plugins).anyTimes(); - expectRebalance(1, Collections.emptyList(), Collections.emptyList()); + expectRebalance(1, Collections.emptyList(), Collections.emptyList(), true); EasyMock.expect(configBackingStore.snapshot()).andReturn(SNAPSHOT).times(2); WorkerConfigTransformer configTransformer = EasyMock.mock(WorkerConfigTransformer.class); @@ -2503,7 +2505,7 @@ public void testAccessors() throws Exception { @Test public void testPutConnectorConfig() throws Exception { EasyMock.expect(member.memberId()).andStubReturn("leader"); - expectRebalance(1, Arrays.asList(CONN1), Collections.emptyList()); + expectRebalance(1, Arrays.asList(CONN1), Collections.emptyList(), true); expectPostRebalanceCatchup(SNAPSHOT); EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0); Capture> onFirstStart = newCapture(); @@ -2615,7 +2617,7 @@ public void testKeyRotationWhenWorkerBecomesLeader() throws Exception { member.poll(Long.MAX_VALUE); EasyMock.expectLastCall(); - expectRebalance(2, Collections.emptyList(), Collections.emptyList(), "member", MEMBER_URL); + expectRebalance(2, Collections.emptyList(), Collections.emptyList(), "member", MEMBER_URL, true); Capture updatedKey = EasyMock.newCapture(); configBackingStore.putSessionKey(EasyMock.capture(updatedKey)); EasyMock.expectLastCall().andAnswer(() -> { @@ -2644,7 +2646,7 @@ public void testKeyRotationDisabledWhenWorkerBecomesFollower() throws Exception EasyMock.expect(member.memberId()).andStubReturn("member"); EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V2); - expectRebalance(1, Collections.emptyList(), Collections.emptyList(), "member", MEMBER_URL); + expectRebalance(1, Collections.emptyList(), Collections.emptyList(), "member", MEMBER_URL, true); SecretKey initialSecretKey = EasyMock.mock(SecretKey.class); EasyMock.expect(initialSecretKey.getAlgorithm()).andReturn(DistributedConfig.INTER_WORKER_KEY_GENERATION_ALGORITHM_DEFAULT).anyTimes(); EasyMock.expect(initialSecretKey.getEncoded()).andReturn(new byte[32]).anyTimes(); @@ -2783,7 +2785,7 @@ public void testFailedToWriteSessionKey() throws Exception { // session key to the config topic, and fail EasyMock.expect(member.memberId()).andStubReturn("leader"); EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V2); - expectRebalance(1, Collections.emptyList(), Collections.emptyList()); + expectRebalance(1, Collections.emptyList(), Collections.emptyList(), true); expectPostRebalanceCatchup(SNAPSHOT); configBackingStore.putSessionKey(anyObject(SessionKey.class)); EasyMock.expectLastCall().andThrow(new ConnectException("Oh no!")); @@ -2825,7 +2827,7 @@ public void testFailedToReadBackNewlyWrittenSessionKey() throws Exception { // to write the key) EasyMock.expect(member.memberId()).andStubReturn("leader"); EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V2); - expectRebalance(1, Collections.emptyList(), Collections.emptyList()); + expectRebalance(1, Collections.emptyList(), Collections.emptyList(), true); expectPostRebalanceCatchup(SNAPSHOT); configBackingStore.putSessionKey(anyObject(SessionKey.class)); EasyMock.expectLastCall().andThrow(new ConnectException("Oh no!")); @@ -2903,16 +2905,22 @@ public void testHerderStopServicesClosesUponShutdown() { private void expectRebalance(final long offset, final List assignedConnectors, final List assignedTasks) { - expectRebalance(Collections.emptyList(), Collections.emptyList(), - ConnectProtocol.Assignment.NO_ERROR, offset, assignedConnectors, assignedTasks, 0); + expectRebalance(offset, assignedConnectors, assignedTasks, false); } private void expectRebalance(final long offset, final List assignedConnectors, final List assignedTasks, - String leader, String leaderUrl) { + final boolean isLeader) { expectRebalance(Collections.emptyList(), Collections.emptyList(), - ConnectProtocol.Assignment.NO_ERROR, offset, leader, leaderUrl, assignedConnectors, assignedTasks, 0); + ConnectProtocol.Assignment.NO_ERROR, offset, "leader", "leaderUrl", assignedConnectors, assignedTasks, 0, isLeader); + } + + private void expectRebalance(final long offset, + final List assignedConnectors, final List assignedTasks, + String leader, String leaderUrl, boolean isLeader) { + expectRebalance(Collections.emptyList(), Collections.emptyList(), + ConnectProtocol.Assignment.NO_ERROR, offset, leader, leaderUrl, assignedConnectors, assignedTasks, 0, isLeader); } // Handles common initial part of rebalance callback. Does not handle instantiation of connectors and tasks. @@ -2924,7 +2932,6 @@ private void expectRebalance(final Collection revokedConnectors, final List assignedTasks) { expectRebalance(revokedConnectors, revokedTasks, error, offset, assignedConnectors, assignedTasks, 0); } - // Handles common initial part of rebalance callback. Does not handle instantiation of connectors and tasks. private void expectRebalance(final Collection revokedConnectors, final List revokedTasks, @@ -2933,7 +2940,7 @@ private void expectRebalance(final Collection revokedConnectors, final List assignedConnectors, final List assignedTasks, int delay) { - expectRebalance(revokedConnectors, revokedTasks, error, offset, "leader", "leaderUrl", assignedConnectors, assignedTasks, delay); + expectRebalance(revokedConnectors, revokedTasks, error, offset, "leader", "leaderUrl", assignedConnectors, assignedTasks, delay, false); } // Handles common initial part of rebalance callback. Does not handle instantiation of connectors and tasks. @@ -2945,7 +2952,8 @@ private void expectRebalance(final Collection revokedConnectors, String leaderUrl, final List assignedConnectors, final List assignedTasks, - int delay) { + int delay, + boolean isLeader) { member.ensureActive(); PowerMock.expectLastCall().andAnswer(() -> { ExtendedAssignment assignment; @@ -2969,6 +2977,11 @@ private void expectRebalance(final Collection revokedConnectors, return null; }); + if (isLeader) { + configBackingStore.claimWritePrivileges(); + EasyMock.expectLastCall(); + } + if (!revokedConnectors.isEmpty()) { for (String connector : revokedConnectors) { worker.stopAndAwaitConnector(connector); diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/IncrementalCooperativeAssignorTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/IncrementalCooperativeAssignorTest.java index bd51456b721cc..bb767139bb910 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/IncrementalCooperativeAssignorTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/IncrementalCooperativeAssignorTest.java @@ -24,6 +24,7 @@ import org.apache.kafka.connect.runtime.TargetState; import org.apache.kafka.connect.runtime.distributed.WorkerCoordinator.ConnectorsAndTasks; import org.apache.kafka.connect.util.ConnectUtils; +import org.apache.kafka.connect.storage.ClusterConfigState; import org.apache.kafka.connect.util.ConnectorTaskId; import org.junit.Before; import org.junit.Test; diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/WorkerCoordinatorIncrementalTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/WorkerCoordinatorIncrementalTest.java index 3a15956ff1a4d..f8cf14200ca41 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/WorkerCoordinatorIncrementalTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/WorkerCoordinatorIncrementalTest.java @@ -26,6 +26,7 @@ import org.apache.kafka.common.requests.RequestTestUtils; import org.apache.kafka.common.utils.LogContext; import org.apache.kafka.common.utils.MockTime; +import org.apache.kafka.connect.storage.ClusterConfigState; import org.apache.kafka.connect.storage.KafkaConfigBackingStore; import org.apache.kafka.connect.util.ConnectorTaskId; import org.junit.After; diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/WorkerCoordinatorTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/WorkerCoordinatorTest.java index 60fbe37ad36ab..e17fb7daf7543 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/WorkerCoordinatorTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/WorkerCoordinatorTest.java @@ -36,6 +36,7 @@ import org.apache.kafka.common.utils.MockTime; import org.apache.kafka.common.utils.Utils; import org.apache.kafka.connect.runtime.TargetState; +import org.apache.kafka.connect.storage.ClusterConfigState; import org.apache.kafka.connect.storage.KafkaConfigBackingStore; import org.apache.kafka.connect.util.ConnectorTaskId; import org.easymock.EasyMock; diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/standalone/StandaloneHerderTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/standalone/StandaloneHerderTest.java index f5ee4ccd310d7..629187a6ef2b4 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/standalone/StandaloneHerderTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/standalone/StandaloneHerderTest.java @@ -41,7 +41,7 @@ import org.apache.kafka.connect.runtime.Worker; import org.apache.kafka.connect.runtime.WorkerConfigTransformer; import org.apache.kafka.connect.runtime.WorkerConnector; -import org.apache.kafka.connect.runtime.distributed.ClusterConfigState; +import org.apache.kafka.connect.storage.ClusterConfigState; import org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader; import org.apache.kafka.connect.runtime.isolation.PluginClassLoader; import org.apache.kafka.connect.runtime.isolation.Plugins; diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/storage/KafkaConfigBackingStoreTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/storage/KafkaConfigBackingStoreTest.java index 726b4ccbbf90d..32719d816cde4 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/storage/KafkaConfigBackingStoreTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/storage/KafkaConfigBackingStoreTest.java @@ -20,7 +20,10 @@ import org.apache.kafka.clients.admin.NewTopic; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.producer.Producer; import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.common.IsolationLevel; +import org.apache.kafka.common.errors.ProducerFencedException; import org.apache.kafka.common.header.internals.RecordHeaders; import org.apache.kafka.common.record.TimestampType; import org.apache.kafka.common.config.ConfigException; @@ -30,7 +33,6 @@ import org.apache.kafka.connect.data.Struct; import org.apache.kafka.connect.runtime.RestartRequest; import org.apache.kafka.connect.runtime.TargetState; -import org.apache.kafka.connect.runtime.distributed.ClusterConfigState; import org.apache.kafka.connect.runtime.distributed.DistributedConfig; import org.apache.kafka.connect.util.Callback; import org.apache.kafka.connect.util.ConnectUtils; @@ -50,22 +52,30 @@ import org.powermock.modules.junit4.PowerMockRunner; import org.powermock.reflect.Whitebox; +import java.time.Duration; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.HashMap; import java.util.LinkedHashMap; import java.util.List; +import java.util.Locale; import java.util.Map; import java.util.Optional; import java.util.concurrent.TimeUnit; import java.util.function.Supplier; +import static org.apache.kafka.clients.consumer.ConsumerConfig.ISOLATION_LEVEL_CONFIG; +import static org.apache.kafka.clients.producer.ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG; +import static org.apache.kafka.clients.producer.ProducerConfig.TRANSACTIONAL_ID_CONFIG; +import static org.apache.kafka.connect.runtime.distributed.DistributedConfig.GROUP_ID_CONFIG; import static org.apache.kafka.connect.storage.KafkaConfigBackingStore.INCLUDE_TASKS_FIELD_NAME; import static org.apache.kafka.connect.storage.KafkaConfigBackingStore.ONLY_FAILED_FIELD_NAME; import static org.apache.kafka.connect.storage.KafkaConfigBackingStore.RESTART_KEY; +import static org.apache.kafka.connect.runtime.distributed.DistributedConfig.EXACTLY_ONCE_SOURCE_SUPPORT_CONFIG; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotSame; import static org.junit.Assert.assertNull; import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; @@ -151,6 +161,8 @@ public class KafkaConfigBackingStoreTest { private ConfigBackingStore.UpdateListener configUpdateListener; @Mock KafkaBasedLog storeLog; + @Mock + Producer fencableProducer; private KafkaConfigBackingStore configStorage; private Capture capturedTopic = EasyMock.newCapture(); @@ -162,15 +174,22 @@ public class KafkaConfigBackingStoreTest { private long logOffset = 0; + private void createStore(DistributedConfig config, KafkaBasedLog storeLog) { + configStorage = PowerMock.createPartialMock( + KafkaConfigBackingStore.class, + new String[]{"createKafkaBasedLog", "createFencableProducer"}, + converter, config, null); + Whitebox.setInternalState(configStorage, "configLog", storeLog); + configStorage.setUpdateListener(configUpdateListener); + } + @Before public void setUp() { PowerMock.mockStaticPartial(ConnectUtils.class, "lookupKafkaClusterId"); EasyMock.expect(ConnectUtils.lookupKafkaClusterId(EasyMock.anyObject())).andReturn("test-cluster").anyTimes(); PowerMock.replay(ConnectUtils.class); - configStorage = PowerMock.createPartialMock(KafkaConfigBackingStore.class, new String[]{"createKafkaBasedLog"}, converter, DEFAULT_DISTRIBUTED_CONFIG, null); - Whitebox.setInternalState(configStorage, "configLog", storeLog); - configStorage.setUpdateListener(configUpdateListener); + createStore(DEFAULT_DISTRIBUTED_CONFIG, storeLog); } @Test @@ -203,6 +222,29 @@ public void testStartStop() throws Exception { PowerMock.verifyAll(); } + @Test + public void testSnapshotCannotMutateInternalState() throws Exception { + expectConfigure(); + expectStart(Collections.emptyList(), Collections.emptyMap()); + expectPartitionCount(1); + PowerMock.replayAll(); + + Map settings = new HashMap<>(DEFAULT_CONFIG_STORAGE_PROPS); + settings.put("config.storage.min.insync.replicas", "3"); + settings.put("config.storage.max.message.bytes", "1001"); + configStorage.setupAndCreateKafkaBasedLog(TOPIC, new DistributedConfig(settings)); + + configStorage.start(); + ClusterConfigState snapshot = configStorage.snapshot(); + assertNotSame(snapshot.connectorTaskCounts, configStorage.connectorTaskCounts); + assertNotSame(snapshot.connectorConfigs, configStorage.connectorConfigs); + assertNotSame(snapshot.connectorTargetStates, configStorage.connectorTargetStates); + assertNotSame(snapshot.taskConfigs, configStorage.taskConfigs); + assertNotSame(snapshot.inconsistentConnectors, configStorage.inconsistent); + + PowerMock.verifyAll(); + } + @Test public void testPutConnectorConfig() throws Exception { expectConfigure(); @@ -266,6 +308,86 @@ public void testPutConnectorConfig() throws Exception { PowerMock.verifyAll(); } + @Test + public void testWritePrivileges() throws Exception { + // With exactly.once.source.support = preparing (or also, "enabled"), we need to use a transactional producer + // to write some types of messages to the config topic + Map workerProps = new HashMap<>(DEFAULT_CONFIG_STORAGE_PROPS); + workerProps.put(EXACTLY_ONCE_SOURCE_SUPPORT_CONFIG, "preparing"); + DistributedConfig config = new DistributedConfig(workerProps); + createStore(config, storeLog); + + expectConfigure(); + expectStart(Collections.emptyList(), Collections.emptyMap()); + + // Try and fail to write a connector config to the config topic without write privileges + expectConvert(KafkaConfigBackingStore.CONNECTOR_CONFIGURATION_V0, CONNECTOR_CONFIG_STRUCTS.get(0), CONFIGS_SERIALIZED.get(1)); + // Claim write privileges + expectFencableProducer(); + + // Try again to write a connector config + expectConvert(KafkaConfigBackingStore.CONNECTOR_CONFIGURATION_V0, CONNECTOR_CONFIG_STRUCTS.get(0), CONFIGS_SERIALIZED.get(1)); + fencableProducer.beginTransaction(); + EasyMock.expectLastCall(); + EasyMock.expect(fencableProducer.send(EasyMock.anyObject())).andReturn(null); + // Get fenced out + fencableProducer.commitTransaction(); + EasyMock.expectLastCall().andThrow(new ProducerFencedException("Better luck next time")); + fencableProducer.close(Duration.ZERO); + EasyMock.expectLastCall(); + // And fail when trying to write again without reclaiming write privileges + expectConvert(KafkaConfigBackingStore.CONNECTOR_CONFIGURATION_V0, CONNECTOR_CONFIG_STRUCTS.get(0), CONFIGS_SERIALIZED.get(1)); + + // In the meantime, write a target state (which doesn't require write privileges) + expectConvert(KafkaConfigBackingStore.TARGET_STATE_V0, TARGET_STATE_PAUSED, CONFIGS_SERIALIZED.get(1)); + storeLog.send("target-state-" + CONNECTOR_IDS.get(1), CONFIGS_SERIALIZED.get(1)); + PowerMock.expectLastCall(); + + // Reclaim write privileges + expectFencableProducer(); + // And successfully write the config + expectConvert(KafkaConfigBackingStore.CONNECTOR_CONFIGURATION_V0, CONNECTOR_CONFIG_STRUCTS.get(0), CONFIGS_SERIALIZED.get(1)); + fencableProducer.beginTransaction(); + EasyMock.expectLastCall(); + EasyMock.expect(fencableProducer.send(EasyMock.anyObject())).andReturn(null); + fencableProducer.commitTransaction(); + EasyMock.expectLastCall(); + expectConvertRead(CONNECTOR_CONFIG_KEYS.get(1), CONNECTOR_CONFIG_STRUCTS.get(0), CONFIGS_SERIALIZED.get(2)); + configUpdateListener.onConnectorConfigUpdate(CONNECTOR_IDS.get(1)); + EasyMock.expectLastCall(); + + expectPartitionCount(1); + expectStop(); + fencableProducer.close(Duration.ZERO); + EasyMock.expectLastCall(); + + PowerMock.replayAll(); + + + configStorage.setupAndCreateKafkaBasedLog(TOPIC, DEFAULT_DISTRIBUTED_CONFIG); + configStorage.start(); + + // Should fail the first time since we haven't claimed write privileges + assertThrows(IllegalStateException.class, () -> configStorage.putConnectorConfig(CONNECTOR_IDS.get(1), SAMPLE_CONFIGS.get(0))); + // Try to recover + configStorage.claimWritePrivileges(); + // Should fail again when we get fenced out + assertThrows(PrivilegedWriteException.class, () -> configStorage.putConnectorConfig(CONNECTOR_IDS.get(1), SAMPLE_CONFIGS.get(0))); + // Should fail if we retry without reclaiming write privileges + assertThrows(IllegalStateException.class, () -> configStorage.putConnectorConfig(CONNECTOR_IDS.get(1), SAMPLE_CONFIGS.get(0))); + + // Should succeed even without write privileges (target states can be written by anyone) + configStorage.putTargetState(CONNECTOR_IDS.get(1), TargetState.PAUSED); + + // Should succeed if we re-claim write privileges + configStorage.claimWritePrivileges(); + configStorage.putConnectorConfig(CONNECTOR_IDS.get(1), SAMPLE_CONFIGS.get(0)); + + configStorage.stop(); + + PowerMock.verifyAll(); + } + @Test public void testPutTaskConfigs() throws Exception { expectConfigure(); @@ -1066,6 +1188,127 @@ public void testExceptionOnStartWhenConfigTopicHasMultiplePartitions() throws Ex PowerMock.verifyAll(); } + @Test + public void testFencableProducerPropertiesInsertedByDefault() throws Exception { + Map workerProps = new HashMap<>(DEFAULT_CONFIG_STORAGE_PROPS); + workerProps.put(EXACTLY_ONCE_SOURCE_SUPPORT_CONFIG, "preparing"); + String groupId = "my-connect-cluster"; + workerProps.put(GROUP_ID_CONFIG, groupId); + workerProps.remove(TRANSACTIONAL_ID_CONFIG); + workerProps.remove(ENABLE_IDEMPOTENCE_CONFIG); + DistributedConfig config = new DistributedConfig(workerProps); + createStore(config, storeLog); + + PowerMock.replayAll(); + + Map fencableProducerProperties = configStorage.fencableProducerProps(config); + assertEquals("connect-cluster-" + groupId, fencableProducerProperties.get(TRANSACTIONAL_ID_CONFIG)); + assertEquals("true", fencableProducerProperties.get(ENABLE_IDEMPOTENCE_CONFIG)); + + PowerMock.verifyAll(); + } + + @Test + public void testFencableProducerPropertiesOverrideUserSuppliedValues() throws Exception { + Map workerProps = new HashMap<>(DEFAULT_CONFIG_STORAGE_PROPS); + workerProps.put(EXACTLY_ONCE_SOURCE_SUPPORT_CONFIG, "preparing"); + String groupId = "my-other-connect-cluster"; + workerProps.put(GROUP_ID_CONFIG, groupId); + workerProps.put(TRANSACTIONAL_ID_CONFIG, "my-custom-transactional-id"); + workerProps.put(ENABLE_IDEMPOTENCE_CONFIG, "false"); + DistributedConfig config = new DistributedConfig(workerProps); + createStore(config, storeLog); + + PowerMock.replayAll(); + + Map fencableProducerProperties = configStorage.fencableProducerProps(config); + assertEquals("connect-cluster-" + groupId, fencableProducerProperties.get(TRANSACTIONAL_ID_CONFIG)); + assertEquals("true", fencableProducerProperties.get(ENABLE_IDEMPOTENCE_CONFIG)); + + PowerMock.verifyAll(); + } + + @Test + public void testConsumerPropertiesInsertedByDefaultWithExactlyOnceSourceEnabled() throws Exception { + Map workerProps = new HashMap<>(DEFAULT_CONFIG_STORAGE_PROPS); + workerProps.put(EXACTLY_ONCE_SOURCE_SUPPORT_CONFIG, "enabled"); + workerProps.remove(ISOLATION_LEVEL_CONFIG); + DistributedConfig config = new DistributedConfig(workerProps); + createStore(config, storeLog); + + expectConfigure(); + PowerMock.replayAll(); + + configStorage.setupAndCreateKafkaBasedLog(TOPIC, config); + + assertEquals( + IsolationLevel.READ_COMMITTED.name().toLowerCase(Locale.ROOT), + capturedConsumerProps.getValue().get(ISOLATION_LEVEL_CONFIG) + ); + + PowerMock.verifyAll(); + } + + @Test + public void testConsumerPropertiesOverrideUserSuppliedValuesWithExactlyOnceSourceEnabled() throws Exception { + Map workerProps = new HashMap<>(DEFAULT_CONFIG_STORAGE_PROPS); + workerProps.put(EXACTLY_ONCE_SOURCE_SUPPORT_CONFIG, "enabled"); + workerProps.put(ISOLATION_LEVEL_CONFIG, IsolationLevel.READ_UNCOMMITTED.name().toLowerCase(Locale.ROOT)); + DistributedConfig config = new DistributedConfig(workerProps); + createStore(config, storeLog); + + expectConfigure(); + PowerMock.replayAll(); + + configStorage.setupAndCreateKafkaBasedLog(TOPIC, config); + + assertEquals( + IsolationLevel.READ_COMMITTED.name().toLowerCase(Locale.ROOT), + capturedConsumerProps.getValue().get(ISOLATION_LEVEL_CONFIG) + ); + + PowerMock.verifyAll(); + } + + @Test + public void testConsumerPropertiesNotInsertedByDefaultWithoutExactlyOnceSourceEnabled() throws Exception { + Map workerProps = new HashMap<>(DEFAULT_CONFIG_STORAGE_PROPS); + workerProps.put(EXACTLY_ONCE_SOURCE_SUPPORT_CONFIG, "preparing"); + workerProps.remove(ISOLATION_LEVEL_CONFIG); + DistributedConfig config = new DistributedConfig(workerProps); + createStore(config, storeLog); + + expectConfigure(); + PowerMock.replayAll(); + + configStorage.setupAndCreateKafkaBasedLog(TOPIC, config); + + assertNull(capturedConsumerProps.getValue().get(ISOLATION_LEVEL_CONFIG)); + + PowerMock.verifyAll(); + } + + @Test + public void testConsumerPropertiesDoNotOverrideUserSuppliedValuesWithoutExactlyOnceSourceEnabled() throws Exception { + Map workerProps = new HashMap<>(DEFAULT_CONFIG_STORAGE_PROPS); + workerProps.put(EXACTLY_ONCE_SOURCE_SUPPORT_CONFIG, "preparing"); + workerProps.put(ISOLATION_LEVEL_CONFIG, IsolationLevel.READ_UNCOMMITTED.name().toLowerCase(Locale.ROOT)); + DistributedConfig config = new DistributedConfig(workerProps); + createStore(config, storeLog); + + expectConfigure(); + PowerMock.replayAll(); + + configStorage.setupAndCreateKafkaBasedLog(TOPIC, config); + + assertEquals( + IsolationLevel.READ_UNCOMMITTED.name().toLowerCase(Locale.ROOT), + capturedConsumerProps.getValue().get(ISOLATION_LEVEL_CONFIG) + ); + + PowerMock.verifyAll(); + } + private void expectConfigure() throws Exception { PowerMock.expectPrivate(configStorage, "createKafkaBasedLog", EasyMock.capture(capturedTopic), EasyMock.capture(capturedProducerProps), @@ -1074,6 +1317,13 @@ private void expectConfigure() throws Exception { .andReturn(storeLog); } + private void expectFencableProducer() throws Exception { + fencableProducer.initTransactions(); + EasyMock.expectLastCall(); + PowerMock.expectPrivate(configStorage, "createFencableProducer") + .andReturn(fencableProducer); + } + private void expectPartitionCount(int partitionCount) { EasyMock.expect(storeLog.partitionCount()) .andReturn(partitionCount); @@ -1116,6 +1366,11 @@ private void expectRead(final String key, final byte[] serializedValue, Struct d expectRead(serializedData, Collections.singletonMap(key, deserializedValue)); } + private void expectConvert(Schema valueSchema, Struct valueStruct, byte[] serialized) { + EasyMock.expect(converter.fromConnectData(EasyMock.eq(TOPIC), EasyMock.eq(valueSchema), EasyMock.eq(valueStruct))) + .andReturn(serialized); + } + // Expect a conversion & write to the underlying log, followed by a subsequent read when the data is consumed back // from the log. Validate the data that is captured when the conversion is performed matches the specified data // (by checking a single field's value) @@ -1136,6 +1391,14 @@ private void expectConvertWriteRead(final String configKey, final Schema valueSc }); } + private void expectConvertRead(final String configKey, final Struct struct, final byte[] serialized) { + EasyMock.expect(converter.toConnectData(EasyMock.eq(TOPIC), EasyMock.aryEq(serialized))) + .andAnswer(() -> new SchemaAndValue(null, serialized == null ? null : structToMap(struct))); + LinkedHashMap recordsToRead = new LinkedHashMap<>(); + recordsToRead.put(configKey, serialized); + expectReadToEnd(recordsToRead); + } + // This map needs to maintain ordering private void expectReadToEnd(final LinkedHashMap serializedConfigs) { EasyMock.expect(storeLog.readToEnd())