You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by kk...@apache.org on 2021/01/26 19:17:41 UTC
[kafka] branch 2.6 updated: KAFKA-10763: Fix incomplete cooperative
rebalances preventing connector/task revocations (#9765)
This is an automated email from the ASF dual-hosted git repository.
kkarantasis pushed a commit to branch 2.6
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/2.6 by this push:
new 7298e44 KAFKA-10763: Fix incomplete cooperative rebalances preventing connector/task revocations (#9765)
7298e44 is described below
commit 7298e44b4c68bebac5a92b04b7921dc0b36f8655
Author: Greg Harris <gr...@confluent.io>
AuthorDate: Tue Jan 26 10:17:05 2021 -0800
KAFKA-10763: Fix incomplete cooperative rebalances preventing connector/task revocations (#9765)
When two cooperative rebalances take place soon after one another, a prior rebalance may not complete before the next rebalance is started.
Under Eager rebalancing, no tasks would have been started, so the subsequent onRevoked call is intentionally skipped whenever rebalanceResolved was false.
Under Incremental Cooperative rebalancing, the same logic causes the DistributedHerder to skip stopping all of the connector/task revocations which occur in the second rebalance.
The DistributedHerder still removes the revoked connectors/tasks from its assignment, so that the DistributedHerder and Worker have different knowledge of running connectors/tasks.
This causes the connector/task instances that would have been stopped to disappear from the rebalance protocol, and left running until their workers are halted, or they fail.
Connectors/Tasks which were then reassigned to other workers by the rebalance protocol would be duplicated, and run concurrently with zombie connectors/tasks.
Connectors/Tasks which were reassigned back to the same worker would encounter exceptions in Worker, indicating that the connector/task existed and was already running.
* Added test for revoking and then reassigning a connector under normal circumstances
* Added test for revoking and then reassigning a connector following an incomplete cooperative rebalance
* Changed expectRebalance to make assignment fields mutable before passing them into the DistributedHerder
* Only skip revocation for the Eager protocol, and never skip revocation for incremental cooperative protocols
Reviewers: Konstantine Karantasis <k....@gmail.com>
---
.../runtime/distributed/DistributedHerder.java | 3 +-
.../runtime/distributed/DistributedHerderTest.java | 108 ++++++++++++++++++++-
2 files changed, 109 insertions(+), 2 deletions(-)
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 e59025d..266427e 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
@@ -93,6 +93,7 @@ import java.util.stream.Collectors;
import static org.apache.kafka.connect.runtime.WorkerConfig.TOPIC_TRACKING_ENABLE_CONFIG;
import static org.apache.kafka.connect.runtime.distributed.ConnectProtocol.CONNECT_PROTOCOL_V0;
import static org.apache.kafka.connect.runtime.distributed.ConnectProtocolCompatibility.EAGER;
+import static org.apache.kafka.connect.runtime.distributed.IncrementalCooperativeConnectProtocol.CONNECT_PROTOCOL_V1;
import static org.apache.kafka.connect.runtime.distributed.IncrementalCooperativeConnectProtocol.CONNECT_PROTOCOL_V2;
/**
@@ -1746,7 +1747,7 @@ public class DistributedHerder extends AbstractHerder implements Runnable {
// Note that since we don't reset the assignment, we don't revoke leadership here. During a rebalance,
// it is still important to have a leader that can write configs, offsets, etc.
- if (rebalanceResolved) {
+ if (rebalanceResolved || currentProtocolVersion >= CONNECT_PROTOCOL_V1) {
List<Callable<Void>> callables = new ArrayList<>();
for (final String connectorName : connectors) {
callables.add(getConnectorStoppingCallable(connectorName));
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 8cf485a..bea062e 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
@@ -567,6 +567,112 @@ public class DistributedHerderTest {
}
@Test
+ public void testRevoke() throws TimeoutException {
+ revokeAndReassign(false);
+ }
+
+ @Test
+ public void testIncompleteRebalanceBeforeRevoke() throws TimeoutException {
+ revokeAndReassign(true);
+ }
+
+ public void revokeAndReassign(boolean incompleteRebalance) throws TimeoutException {
+ connectProtocolVersion = CONNECT_PROTOCOL_V1;
+ int configOffset = 1;
+
+ // Join group and get initial assignment
+ EasyMock.expect(member.memberId()).andStubReturn("member");
+ EasyMock.expect(member.currentProtocolVersion()).andStubReturn(connectProtocolVersion);
+ EasyMock.expect(worker.getPlugins()).andReturn(plugins);
+ expectRebalance(configOffset, Arrays.asList(CONN1), Arrays.asList(TASK1));
+ expectPostRebalanceCatchup(SNAPSHOT);
+ Capture<Callback<TargetState>> onFirstStart = newCapture();
+ worker.startConnector(EasyMock.eq(CONN1), EasyMock.<Map<String, String>>anyObject(), EasyMock.<CloseableConnectorContext>anyObject(),
+ EasyMock.eq(herder), EasyMock.eq(TargetState.STARTED), capture(onFirstStart));
+ PowerMock.expectLastCall().andAnswer(new IAnswer<Boolean>() {
+ @Override
+ public Boolean answer() throws Throwable {
+ onFirstStart.getValue().onCompletion(null, TargetState.STARTED);
+ return true;
+ }
+ });
+ member.wakeup();
+ PowerMock.expectLastCall();
+ EasyMock.expect(worker.isRunning(CONN1)).andReturn(true);
+ EasyMock.expect(worker.connectorTaskConfigs(CONN1, conn1SinkConfig)).andReturn(TASK_CONFIGS);
+ worker.startTask(EasyMock.eq(TASK1), EasyMock.<ClusterConfigState>anyObject(), EasyMock.<Map<String, String>>anyObject(), EasyMock.<Map<String, String>>anyObject(),
+ EasyMock.eq(herder), EasyMock.eq(TargetState.STARTED));
+ PowerMock.expectLastCall().andReturn(true);
+ member.poll(EasyMock.anyInt());
+ PowerMock.expectLastCall();
+
+ // worker is stable with an existing set of tasks
+
+ if (incompleteRebalance) {
+ // Perform a partial re-balance just prior to the revocation
+ // bump the configOffset to trigger reading the config topic to the end
+ configOffset++;
+ expectRebalance(configOffset, Arrays.<String>asList(), Arrays.<ConnectorTaskId>asList());
+ // give it the wrong snapshot, as if we're out of sync/can't reach the broker
+ expectPostRebalanceCatchup(SNAPSHOT);
+ member.requestRejoin();
+ PowerMock.expectLastCall();
+ // tick exits early because we failed, and doesn't do the poll at the end of the method
+ // the worker did not startWork or reset the rebalanceResolved flag
+ }
+
+ // Revoke the connector in the next rebalance
+ expectRebalance(Arrays.asList(CONN1), Arrays.<ConnectorTaskId>asList(),
+ ConnectProtocol.Assignment.NO_ERROR, configOffset, Arrays.<String>asList(),
+ Arrays.<ConnectorTaskId>asList());
+
+ if (incompleteRebalance) {
+ // Same as SNAPSHOT, except with an updated offset
+ // Allow the task to read to the end of the topic and complete the rebalance
+ ClusterConfigState secondSnapshot = new ClusterConfigState(
+ configOffset, null, Collections.singletonMap(CONN1, 3),
+ Collections.singletonMap(CONN1, CONN1_CONFIG), Collections.singletonMap(CONN1, TargetState.STARTED),
+ TASK_CONFIGS_MAP, Collections.<String>emptySet());
+ expectPostRebalanceCatchup(secondSnapshot);
+ }
+ member.requestRejoin();
+ PowerMock.expectLastCall();
+
+ // re-assign the connector back to the same worker to ensure state was cleaned up
+ expectRebalance(configOffset, Arrays.asList(CONN1), Arrays.<ConnectorTaskId>asList());
+ EasyMock.expect(worker.getPlugins()).andReturn(plugins);
+ Capture<Callback<TargetState>> onSecondStart = newCapture();
+ worker.startConnector(EasyMock.eq(CONN1), EasyMock.<Map<String, String>>anyObject(),
+ EasyMock.<CloseableConnectorContext>anyObject(),
+ EasyMock.eq(herder), EasyMock.eq(TargetState.STARTED), capture(onSecondStart));
+ PowerMock.expectLastCall().andAnswer(new IAnswer<Boolean>() {
+ @Override
+ public Boolean answer() throws Throwable {
+ onSecondStart.getValue().onCompletion(null, TargetState.STARTED);
+ return true;
+ }
+ });
+ member.wakeup();
+ PowerMock.expectLastCall();
+ EasyMock.expect(worker.isRunning(CONN1)).andReturn(true);
+ EasyMock.expect(worker.connectorTaskConfigs(CONN1, conn1SinkConfig))
+ .andReturn(TASK_CONFIGS);
+ member.poll(EasyMock.anyInt());
+ PowerMock.expectLastCall();
+
+ PowerMock.replayAll();
+
+ herder.tick();
+ if (incompleteRebalance) {
+ herder.tick();
+ }
+ herder.tick();
+ herder.tick();
+
+ PowerMock.verifyAll();
+ }
+
+ @Test
public void testHaltCleansUpWorker() {
EasyMock.expect(worker.connectorNames()).andReturn(Collections.singleton(CONN1));
worker.stopAndAwaitConnector(CONN1);
@@ -2241,7 +2347,7 @@ public class DistributedHerderTest {
} else {
assignment = new ExtendedAssignment(
connectProtocolVersion, error, "leader", "leaderUrl", offset,
- assignedConnectors, assignedTasks,
+ new ArrayList<>(assignedConnectors), new ArrayList<>(assignedTasks),
new ArrayList<>(revokedConnectors), new ArrayList<>(revokedTasks), delay);
}
rebalanceListener.onAssigned(assignment, 3);