[kafka] branch trunk updated: KAFKA-15059: Remove pending rebalance check when fencing zombie source connector tasks (#13819)

[vi...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

viktor 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 e1d59920f4f KAFKA-15059: Remove pending rebalance check when fencing zombie source connector tasks (#13819)
e1d59920f4f is described below

commit e1d59920f4fed5c4bc890e53e249d3439e148bab
Author: Chris Egerton <ch...@aiven.io>
AuthorDate: Fri Jun 16 02:58:36 2023 -0700

    KAFKA-15059: Remove pending rebalance check when fencing zombie source connector tasks (#13819)
    
    Discovered while researching KAFKA-14718
    
    Currently, we perform a check during zombie fencing that causes the round of zombie fencing to fail when a rebalance is pending (i.e., when we've detected from a background poll of the config topic that a new connector has been created, that an existing connector has been deleted, or that a new set of connector tasks has been generated).
    
    It's possible but not especially likely that this check causes issues when running vanilla Kafka Connect. Even when it does, it's easy enough to restart failed tasks via the REST API.
    
    However, when running MirrorMaker 2 in dedicated mode, this check is more likely to cause issues as we write three connector configs to the config topic in rapid succession on startup. And in that mode, there is no API to restart failed tasks aside from restarting the worker that they are hosted on.
    
    In either case, this check can lead to test flakiness in integration tests for MirrorMaker 2 both in dedicated mode and when deployed onto a vanilla Kafka Connect cluster.
    
    This check is not actually necessary, and we can safely remove it. Copied from Jira:
    
    >If the worker that we forward the zombie fencing request to is a zombie leader (i.e., a worker that believes it is the leader but in reality is not), it will fail to finish the round of zombie fencing because it won't be able to write to the config topic with a transactional producer.
    
    >If the connector has just been deleted, we'll still fail the request since we force a read-to-end of the config topic and refresh our snapshot of its contents before checking to see if the connector exists.
    
    >And regardless, the worker that owns the task will still do a read-to-end of the config topic and verify that (1) no new task configs have been generated for the connector and (2) the worker is still assigned the connector, before allowing the task to process any data.
    
    In addition, while waiting on a fix for KAFKA-14718 that adds more granularity for diagnosing failures in the DedicatedMirrorIntegrationTest suite (#13284), some of the timeouts in that test are bumped to work better on our CI infrastructure.
    
    Reviewers: Mickael Maison <mi...@gmail.com>, Yash Mayya <ya...@gmail.com>, Viktor Somogyi-Vass <vi...@gmail.com>
---
 .../connect/mirror/integration/DedicatedMirrorIntegrationTest.java    | 4 ++--
 .../apache/kafka/connect/runtime/distributed/DistributedHerder.java   | 3 +--
 2 files changed, 3 insertions(+), 4 deletions(-)

diff --git a/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/DedicatedMirrorIntegrationTest.java b/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/DedicatedMirrorIntegrationTest.java
index e2db7b3865e..2c57a4396f1 100644
--- a/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/DedicatedMirrorIntegrationTest.java
+++ b/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/DedicatedMirrorIntegrationTest.java
@@ -47,8 +47,8 @@ public class DedicatedMirrorIntegrationTest {
 
     private static final Logger log = LoggerFactory.getLogger(DedicatedMirrorIntegrationTest.class);
 
-    private static final int TOPIC_CREATION_TIMEOUT_MS = 30_000;
-    private static final int TOPIC_REPLICATION_TIMEOUT_MS = 30_000;
+    private static final int TOPIC_CREATION_TIMEOUT_MS = 120_000;
+    private static final int TOPIC_REPLICATION_TIMEOUT_MS = 120_000;
 
     private Map<String, EmbeddedKafkaCluster> kafkaClusters;
     private Map<String, MirrorMaker> mirrorMakers;
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 b867a8ab13e..dac8e7b9bb7 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
@@ -1246,8 +1246,7 @@ public class DistributedHerder extends AbstractHerder implements Runnable {
     private void doFenceZombieSourceTasks(String connName, Callback<Void> callback) {
         log.trace("Performing zombie fencing request for connector {}", connName);
 
-        if (checkRebalanceNeeded(callback))
-            return;
+        // We don't have to check for a pending rebalance here
 
         if (!isLeader())
             callback.onCompletion(new NotLeaderException("Only the leader may perform zombie fencing.", leaderUrl()), null);