You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by sh...@apache.org on 2022/06/07 07:17:07 UTC
[kafka] branch trunk updated: KAFKA-10000: Use transactional producer for leader-only writes to the config topic (#11778)
This is an automated email from the ASF dual-hosted git repository.
showuon pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 603502bf5f KAFKA-10000: Use transactional producer for leader-only writes to the config topic (#11778)
603502bf5f is described below
commit 603502bf5fb78983434a1a44ccc15a49ef6942b0
Author: Chris Egerton <ch...@confluent.io>
AuthorDate: Tue Jun 7 00:16:48 2022 -0700
KAFKA-10000: Use transactional producer for leader-only writes to the config topic (#11778)
Implements the behavior described in KIP-618: using a transactional producer for writes to the config topic that should only be performed by the leader of the cluster.
Reviewers: Luke Chen <sh...@gmail.com>, Tom Bentley <tb...@redhat.com>
---
checkstyle/suppressions.xml | 2 +-
.../kafka/connect/runtime/AbstractHerder.java | 2 +-
.../org/apache/kafka/connect/runtime/Worker.java | 2 +-
.../kafka/connect/runtime/WorkerSinkTask.java | 2 +-
.../connect/runtime/WorkerSinkTaskContext.java | 2 +-
.../kafka/connect/runtime/WorkerSourceTask.java | 2 +-
.../connect/runtime/WorkerSourceTaskContext.java | 2 +-
.../runtime/distributed/DistributedHerder.java | 83 +++++--
.../connect/runtime/distributed/EagerAssignor.java | 1 +
.../IncrementalCooperativeAssignor.java | 1 +
.../runtime/distributed/WorkerCoordinator.java | 1 +
.../runtime/standalone/StandaloneHerder.java | 2 +-
.../ClusterConfigState.java | 12 +-
.../kafka/connect/storage/ConfigBackingStore.java | 14 +-
.../connect/storage/KafkaConfigBackingStore.java | 237 ++++++++++++++----
.../connect/storage/MemoryConfigBackingStore.java | 1 -
.../connect/storage/PrivilegedWriteException.java | 32 +++
.../kafka/connect/runtime/AbstractHerderTest.java | 12 +-
.../connect/runtime/ErrorHandlingTaskTest.java | 2 +-
.../kafka/connect/runtime/WorkerSinkTaskTest.java | 2 +-
.../runtime/WorkerSinkTaskThreadedTest.java | 2 +-
.../connect/runtime/WorkerSourceTaskTest.java | 2 +-
.../apache/kafka/connect/runtime/WorkerTest.java | 2 +-
.../kafka/connect/runtime/WorkerTestUtils.java | 2 +-
.../runtime/distributed/DistributedHerderTest.java | 97 ++++----
.../IncrementalCooperativeAssignorTest.java | 1 +
.../WorkerCoordinatorIncrementalTest.java | 1 +
.../runtime/distributed/WorkerCoordinatorTest.java | 1 +
.../runtime/standalone/StandaloneHerderTest.java | 2 +-
.../storage/KafkaConfigBackingStoreTest.java | 271 ++++++++++++++++++++-
30 files changed, 657 insertions(+), 138 deletions(-)
diff --git a/checkstyle/suppressions.xml b/checkstyle/suppressions.xml
index a111ce74cf..9d48741edd 100644
--- a/checkstyle/suppressions.xml
+++ b/checkstyle/suppressions.xml
@@ -130,7 +130,7 @@
<suppress checks="ClassFanOutComplexity"
files="Worker(|Test).java"/>
<suppress checks="MethodLength"
- files="(KafkaConfigBackingStore|Values|IncrementalCooperativeAssignor).java"/>
+ files="(DistributedHerder|KafkaConfigBackingStore|Values|IncrementalCooperativeAssignor).java"/>
<suppress checks="ParameterNumber"
files="Worker(SinkTask|SourceTask|Coordinator).java"/>
<suppress checks="ParameterNumber"
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractHerder.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractHerder.java
index 166ac9f05c..90adde67a6 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractHerder.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractHerder.java
@@ -30,7 +30,6 @@ import org.apache.kafka.connect.connector.Connector;
import org.apache.kafka.connect.connector.policy.ConnectorClientConfigOverridePolicy;
import org.apache.kafka.connect.connector.policy.ConnectorClientConfigRequest;
import org.apache.kafka.connect.errors.NotFoundException;
-import org.apache.kafka.connect.runtime.distributed.ClusterConfigState;
import org.apache.kafka.connect.runtime.isolation.PluginType;
import org.apache.kafka.connect.runtime.isolation.Plugins;
import org.apache.kafka.connect.runtime.rest.entities.ActiveTopicsInfo;
@@ -44,6 +43,7 @@ import org.apache.kafka.connect.runtime.rest.entities.ConnectorType;
import org.apache.kafka.connect.runtime.rest.errors.BadRequestException;
import org.apache.kafka.connect.sink.SinkConnector;
import org.apache.kafka.connect.source.SourceConnector;
+import org.apache.kafka.connect.storage.ClusterConfigState;
import org.apache.kafka.connect.storage.ConfigBackingStore;
import org.apache.kafka.connect.storage.Converter;
import org.apache.kafka.connect.storage.HeaderConverter;
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java
index 96e2bf9b98..d309ac93d2 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java
@@ -35,7 +35,7 @@ import org.apache.kafka.connect.health.ConnectorType;
import org.apache.kafka.connect.json.JsonConverter;
import org.apache.kafka.connect.json.JsonConverterConfig;
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.errors.DeadLetterQueueReporter;
import org.apache.kafka.connect.runtime.errors.ErrorHandlingMetrics;
import org.apache.kafka.connect.runtime.errors.ErrorReporter;
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java
index c2566ef9cf..01303e308d 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java
@@ -40,7 +40,7 @@ import org.apache.kafka.connect.errors.RetriableException;
import org.apache.kafka.connect.header.ConnectHeaders;
import org.apache.kafka.connect.header.Headers;
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.errors.RetryWithToleranceOperator;
import org.apache.kafka.connect.runtime.errors.Stage;
import org.apache.kafka.connect.runtime.errors.WorkerErrantRecordReporter;
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTaskContext.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTaskContext.java
index d91c09896d..f242ef4fe5 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTaskContext.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTaskContext.java
@@ -19,7 +19,7 @@ package org.apache.kafka.connect.runtime;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.connect.errors.IllegalWorkerStateException;
-import org.apache.kafka.connect.runtime.distributed.ClusterConfigState;
+import org.apache.kafka.connect.storage.ClusterConfigState;
import org.apache.kafka.connect.sink.ErrantRecordReporter;
import org.apache.kafka.connect.sink.SinkTaskContext;
import org.slf4j.Logger;
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java
index 9ce2b8dbb8..43bf64c6af 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java
@@ -37,7 +37,7 @@ import org.apache.kafka.connect.header.Header;
import org.apache.kafka.connect.header.Headers;
import org.apache.kafka.connect.runtime.ConnectMetrics.MetricGroup;
import org.apache.kafka.connect.runtime.SubmittedRecords.SubmittedRecord;
-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.Stage;
import org.apache.kafka.connect.runtime.errors.ToleranceType;
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTaskContext.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTaskContext.java
index fe1409b282..5c1451fff1 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTaskContext.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTaskContext.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.source.SourceTaskContext;
import org.apache.kafka.connect.storage.OffsetStorageReader;
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java
index 49d3a5278b..b189d2e642 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java
@@ -49,6 +49,7 @@ import org.apache.kafka.connect.runtime.SourceConnectorConfig;
import org.apache.kafka.connect.runtime.TargetState;
import org.apache.kafka.connect.runtime.TaskStatus;
import org.apache.kafka.connect.runtime.Worker;
+import org.apache.kafka.connect.storage.PrivilegedWriteException;
import org.apache.kafka.connect.runtime.rest.InternalRequestSignature;
import org.apache.kafka.connect.runtime.rest.RestClient;
import org.apache.kafka.connect.runtime.rest.entities.ConnectorInfo;
@@ -61,6 +62,7 @@ import org.apache.kafka.connect.source.ConnectorTransactionBoundaries;
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;
@@ -189,6 +191,7 @@ public class DistributedHerder extends AbstractHerder implements Runnable {
// herder's main thread.
private Set<String> 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 class DistributedHerder extends AbstractHerder implements Runnable {
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 class DistributedHerder extends AbstractHerder implements Runnable {
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 @@ public class DistributedHerder extends AbstractHerder implements Runnable {
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 @@ public class DistributedHerder extends AbstractHerder implements Runnable {
+ "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 class DistributedHerder extends AbstractHerder implements Runnable {
callback.onCompletion(new NotFoundException("Connector " + connName + " not found"), null);
} else {
log.trace("Removing connector config {} {}", connName, configState.connectors());
- configBackingStore.removeConnectorConfig(connName);
+ writeToConfigTopicAsLeader(() -> configBackingStore.removeConnectorConfig(connName));
callback.onCompletion(null, new Created<>(false, null));
}
return null;
@@ -1008,7 +1032,7 @@ public class DistributedHerder extends AbstractHerder implements Runnable {
}
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 class DistributedHerder extends AbstractHerder implements Runnable {
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 @@ public class DistributedHerder extends AbstractHerder implements Runnable {
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 @@ public class DistributedHerder extends AbstractHerder implements Runnable {
if (changed) {
List<Map<String, String>> 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 class DistributedHerder extends AbstractHerder implements Runnable {
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 d86feaaaaf..f4edb98ed6 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 @@ package org.apache.kafka.connect.runtime.distributed;
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 8ed80903c5..57e7b00485 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.common.utils.Time;
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 77fb16eaa3..1ab3543361 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.requests.JoinGroupRequest;
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 dac389ba0e..eb3c3393ce 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.SinkConnectorConfig;
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 717120d850..7a766905b8 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<String, Integer> connectorTaskCounts;
- private final Map<String, Map<String, String>> connectorConfigs;
- private final Map<String, TargetState> connectorTargetStates;
- private final Map<ConnectorTaskId, Map<String, String>> taskConfigs;
- private final Set<String> inconsistentConnectors;
private final WorkerConfigTransformer configTransformer;
+ final Map<String, Integer> connectorTaskCounts;
+ final Map<String, Map<String, String>> connectorConfigs;
+ final Map<String, TargetState> connectorTargetStates;
+ final Map<ConnectorTaskId, Map<String, String>> taskConfigs;
+ final Set<String> 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 826f934ffd..24539076ad 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 @@ package org.apache.kafka.connect.storage;
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 94b98cb9eb..d17864458b 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 @@ package org.apache.kafka.connect.storage;
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.SessionKey;
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.Logger;
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.Collections;
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 class KafkaConfigBackingStore implements ConfigBackingStore {
// initialization as long as we always read the volatile variable "started" before we access the listener.
private UpdateListener updateListener;
+ private final Map<String, Object> 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<String, byte[]> configLog;
// Connector -> # of tasks
- private final Map<String, Integer> connectorTaskCounts = new HashMap<>();
+ final Map<String, Integer> connectorTaskCounts = new HashMap<>();
// Connector and task configs: name or id -> config map
- private final Map<String, Map<String, String>> connectorConfigs = new HashMap<>();
- private final Map<ConnectorTaskId, Map<String, String>> taskConfigs = new HashMap<>();
+ final Map<String, Map<String, String>> connectorConfigs = new HashMap<>();
+ final Map<ConnectorTaskId, Map<String, String>> taskConfigs = new HashMap<>();
private final Supplier<TopicAdmin> 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<String> inconsistent = new HashSet<>();
+ final Set<String> 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 class KafkaConfigBackingStore implements ConfigBackingStore {
// Connector -> Map[ConnectorTaskId -> Configs]
private final Map<String, Map<ConnectorTaskId, Map<String, String>>> deferredTaskUpdates = new HashMap<>();
- private final Map<String, TargetState> connectorTargetStates = new HashMap<>();
+ final Map<String, TargetState> connectorTargetStates = new HashMap<>();
private final WorkerConfigTransformer configTransformer;
+ private final boolean usesFencableWriter;
+ private volatile Producer<String, byte[]> fencableProducer;
+ private final Map<String, Object> 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<TopicAdmin> adminSupplier) {
+ public KafkaConfigBackingStore(Converter converter, DistributedConfig config, WorkerConfigTransformer configTransformer, Supplier<TopicAdmin> 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 class KafkaConfigBackingStore implements ConfigBackingStore {
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 class KafkaConfigBackingStore implements ConfigBackingStore {
@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<String, Object> baseProducerProps(WorkerConfig workerConfig) {
+ Map<String, Object> 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<String, Object> fencableProducerProps(DistributedConfig workerConfig) {
+ Map<String, Object> 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<String, byte[]> createFencableProducer() {
+ return new KafkaProducer<>(fencableProducerProps);
+ }
+
/**
* Get a snapshot of the current state of the cluster.
*/
@@ -349,10 +432,15 @@ public class KafkaConfigBackingStore implements ConfigBackingStore {
/**
* 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<String, String> properties) {
@@ -360,19 +448,30 @@ public class KafkaConfigBackingStore implements ConfigBackingStore {
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 class KafkaConfigBackingStore implements ConfigBackingStore {
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<Map<String, String>> configs) {
@@ -425,7 +520,7 @@ public class KafkaConfigBackingStore implements ConfigBackingStore {
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 class KafkaConfigBackingStore implements ConfigBackingStore {
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 class KafkaConfigBackingStore implements ConfigBackingStore {
}
}
+ /**
+ * 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 class KafkaConfigBackingStore implements ConfigBackingStore {
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 class KafkaConfigBackingStore implements ConfigBackingStore {
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 class KafkaConfigBackingStore implements ConfigBackingStore {
}
}
+ /**
+ * 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 class KafkaConfigBackingStore implements ConfigBackingStore {
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 class KafkaConfigBackingStore implements ConfigBackingStore {
// package private for testing
KafkaBasedLog<String, byte[]> setupAndCreateKafkaBasedLog(String topic, final WorkerConfig config) {
+ Map<String, Object> producerProps = new HashMap<>(baseProducerProps);
+
String clusterId = ConnectUtils.lookupKafkaClusterId(config);
Map<String, Object> originals = config.originals();
- Map<String, Object> 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<String, Object> 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<String, Object> adminProps = new HashMap<>(originals);
ConnectUtils.addMetricsContextProperties(adminProps, config, clusterId);
@@ -547,6 +662,34 @@ public class KafkaConfigBackingStore implements ConfigBackingStore {
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<String, byte[]> createKafkaBasedLog(String topic, Map<String, Object> producerProps,
Map<String, Object> consumerProps,
Callback<ConsumerRecord<String, byte[]>> consumedCallback,
@@ -606,7 +749,7 @@ public class KafkaConfigBackingStore implements ConfigBackingStore {
}
Object targetState = ((Map<String, Object>) 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 a8b2820404..b8f19329ee 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.RestartRequest;
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 0000000000..e4900fa9b0
--- /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 862f31c077..1f5acb412c 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.ConnectorClientConfigOverridePo
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.entities.ConnectorType;
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 java.util.stream.Collectors;
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 a63430d645..e222894945 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.data.Struct;
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 65ab0c7e7c..41d4de5e0a 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.data.SchemaAndValue;
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 a0c99fb471..eef6ca82be 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.common.utils.Time;
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 78db83c7ee..5aeff5e9d8 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.header.ConnectHeaders;
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 e410183c7f..36cec92058 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.errors.ConnectException;
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 ed77018f28..544507ab64 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 54dacd6d82..8f3a8cf3eb 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.ConnectorTransactionBoundaries;
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 class DistributedHerderTest {
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 class DistributedHerderTest {
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<String, String> config = new HashMap<>(CONN2_CONFIG);
@@ -1065,7 +1066,7 @@ public class DistributedHerderTest {
return null;
});
- expectRebalance(1, Collections.emptyList(), Collections.emptyList());
+ expectRebalance(1, Collections.emptyList(), Collections.emptyList(), true);
expectPostRebalanceCatchup(SNAPSHOT);
member.wakeup();
@@ -1104,7 +1105,7 @@ public class DistributedHerderTest {
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<Callback<TargetState>> onStart = newCapture();
worker.startConnector(EasyMock.eq(CONN1), EasyMock.anyObject(), EasyMock.anyObject(),
@@ -1138,8 +1139,8 @@ public class DistributedHerderTest {
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 class DistributedHerderTest {
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 class DistributedHerderTest {
// 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 class DistributedHerderTest {
// 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 class DistributedHerderTest {
// 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 class DistributedHerderTest {
// 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 class DistributedHerderTest {
// 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 class DistributedHerderTest {
// 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 class DistributedHerderTest {
// 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 class DistributedHerderTest {
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 class DistributedHerderTest {
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 class DistributedHerderTest {
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 class DistributedHerderTest {
// 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 class DistributedHerderTest {
// 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<Callback<TargetState>> onStart = newCapture();
@@ -2294,8 +2295,8 @@ public class DistributedHerderTest {
// 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 class DistributedHerderTest {
// 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 class DistributedHerderTest {
// 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 class DistributedHerderTest {
// 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 class DistributedHerderTest {
// 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 class DistributedHerderTest {
// 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 class DistributedHerderTest {
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 class DistributedHerderTest {
@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<Callback<TargetState>> onFirstStart = newCapture();
@@ -2615,7 +2617,7 @@ public class DistributedHerderTest {
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<SessionKey> updatedKey = EasyMock.newCapture();
configBackingStore.putSessionKey(EasyMock.capture(updatedKey));
EasyMock.expectLastCall().andAnswer(() -> {
@@ -2644,7 +2646,7 @@ public class DistributedHerderTest {
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 class DistributedHerderTest {
// 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 class DistributedHerderTest {
// 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 class DistributedHerderTest {
private void expectRebalance(final long offset,
final List<String> assignedConnectors,
final List<ConnectorTaskId> 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<String> assignedConnectors,
final List<ConnectorTaskId> 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<String> assignedConnectors, final List<ConnectorTaskId> 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 @@ public class DistributedHerderTest {
final List<ConnectorTaskId> 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<String> revokedConnectors,
final List<ConnectorTaskId> revokedTasks,
@@ -2933,7 +2940,7 @@ public class DistributedHerderTest {
final List<String> assignedConnectors,
final List<ConnectorTaskId> 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 @@ public class DistributedHerderTest {
String leaderUrl,
final List<String> assignedConnectors,
final List<ConnectorTaskId> assignedTasks,
- int delay) {
+ int delay,
+ boolean isLeader) {
member.ensureActive();
PowerMock.expectLastCall().andAnswer(() -> {
ExtendedAssignment assignment;
@@ -2969,6 +2977,11 @@ public class DistributedHerderTest {
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 bd51456b72..bb767139bb 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.common.utils.Time;
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 3a15956ff1..f8cf14200c 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.metrics.Metrics;
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 60fbe37ad3..e17fb7daf7 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.LogContext;
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 f5ee4ccd31..629187a6ef 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.TaskStatus;
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 726b4ccbbf..32719d816c 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.CommonClientConfigs;
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.SchemaAndValue;
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.core.classloader.annotations.PrepareForTest;
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<String, byte[]> storeLog;
+ @Mock
+ Producer<String, byte[]> fencableProducer;
private KafkaConfigBackingStore configStorage;
private Capture<String> capturedTopic = EasyMock.newCapture();
@@ -162,15 +174,22 @@ public class KafkaConfigBackingStoreTest {
private long logOffset = 0;
+ private void createStore(DistributedConfig config, KafkaBasedLog<String, byte[]> 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 class KafkaConfigBackingStoreTest {
PowerMock.verifyAll();
}
+ @Test
+ public void testSnapshotCannotMutateInternalState() throws Exception {
+ expectConfigure();
+ expectStart(Collections.emptyList(), Collections.emptyMap());
+ expectPartitionCount(1);
+ PowerMock.replayAll();
+
+ Map<String, String> 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 class KafkaConfigBackingStoreTest {
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<String, String> 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 class KafkaConfigBackingStoreTest {
PowerMock.verifyAll();
}
+ @Test
+ public void testFencableProducerPropertiesInsertedByDefault() throws Exception {
+ Map<String, String> 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<String, Object> 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<String, String> 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<String, Object> 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<String, String> 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<String, String> 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<String, String> 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<String, String> 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 @@ public class KafkaConfigBackingStoreTest {
.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 @@ public class KafkaConfigBackingStoreTest {
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 @@ public class KafkaConfigBackingStoreTest {
});
}
+ 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<String, byte[]> recordsToRead = new LinkedHashMap<>();
+ recordsToRead.put(configKey, serialized);
+ expectReadToEnd(recordsToRead);
+ }
+
// This map needs to maintain ordering
private void expectReadToEnd(final LinkedHashMap<String, byte[]> serializedConfigs) {
EasyMock.expect(storeLog.readToEnd())