title: Kafka Connect assigns same task to multiple workers
body: <p>I'm using Kafka Connect in distributed mode. A strange behavior I observed multiple times now is that, after some time (can be hours, can be days), what appears to be a balancing error happens: same tasks get assigned to multiple workers. As a result, they run concurrently and, depending on the nature of the connector, fail or produce "unpredictable" outputs.</p>
  
  <p>The simplest configuration I was able to use to reproduce the behavior is: two Kafka Connect workers, two connectors, each connector with one task only. Kafka Connect is deployed into Kubernetes. Kafka itself is in Confluent Cloud. Both Kafka Connect and Kafka are of the same version (5.3.1).</p>
  
  <p>Relevant messages from the log:</p>
  
  <p>Worker A:</p>
  
  <pre><code>[2019-10-30 12:44:23,925] INFO [Worker clientId=connect-1, groupId=some-kafka-connect-cluster] Successfully joined group with generation 488 (org.apache.kafka.clients.consumer.internals.AbstractCoordinator:469)
  [2019-10-30 12:44:23,926] INFO [Worker clientId=connect-1, groupId=some-kafka-connect-cluster] Joined group at generation 488 and got assignment: Assignment{error=0, leader='connect-1-d5c19893-b33c-4f07-85fb-db9736795759', leaderUrl='http://10.16.0.15:8083/', offset=250, connectorIds=[some-hdfs-sink, some-mqtt-source], taskIds=[some-hdfs-sink-0, some-mqtt-source-0], revokedConnectorIds=[], revokedTaskIds=[], delay=0} (org.apache.kafka.connect.runtime.distributed.DistributedHerder:1397)
  </code></pre>
  
  <p>Worker B:</p>
  
  <pre><code>[2019-10-30 12:44:23,930] INFO [Worker clientId=connect-1, groupId=some-kafka-connect-cluster] Successfully joined group with generation 488 (org.apache.kafka.clients.consumer.internals.AbstractCoordinator:469)
  [2019-10-30 12:44:23,936] INFO [Worker clientId=connect-1, groupId=some-kafka-connect-cluster] Joined group at generation 488 and got assignment: Assignment{error=0, leader='connect-1-d5c19893-b33c-4f07-85fb-db9736795759', leaderUrl='http://10.16.0.15:8083/', offset=250, connectorIds=[some-mqtt-source], taskIds=[some-mqtt-source-0], revokedConnectorIds=[], revokedTaskIds=[], delay=0} (org.apache.kafka.connect.runtime.distributed.DistributedHerder:1397)
  </code></pre>
  
  <p>In above log extracts you can observe that same task (<code>some-mqtt-source-0</code>) is assigned to two workers. After this message, I can also see log messages by the task instances on both workers. </p>
  
  <p>This behavior doesn't depend on the connector (I observed it with other tasks as well). It also doesn't happen immediately after the workers are started, but after some time only. </p>
  
  <p>My question is what can be the cause of this behavior? </p>
  
  <p><strong>EDIT 1</strong>:
  I've tried running 3 workers, instead of two, thinking that it might be a distributed consensus issue. It appears not to be, and having 3 workers doesn't fix the issue.</p>
  
  <p><strong>EDIT 2</strong>:
  I've noticed that just before a worker <em>A</em> is assigned a task that originally ran on worker <em>B</em>, that worker (<em>B</em>) observes an error joining a group. For example, if tasks get "duplicated" in generation N, worker <em>B</em> would not have a "Successfully joined group with generation N" message in logs. More so, between generation N-1 and N+1, worker <em>B</em> typically logs errors like <code>Attempt to heartbeat failed for since member id</code> and <code>Group coordinator bx-xxx-xxxxx.europe-west1.gcp.confluent.cloud:9092 (id: 1234567890 rack: null) is unavailable or invalid</code>. Worker <em>B</em> typically joins the generation N+1 shortly after the generation N (sometimes in as few as just about 3 seconds later). It is now clear what triggers the behavior. However:</p>
  
  <ul>
  <li><p>although I understand that there may be temporary issues like these and they are probably normal in a general case, <strong>why doesn't rebalancing fix the issue <em>after</em> all servers successfully join the next generation</strong>? Although more rebalnce follow - it doesn't correctly distribute the tasks, and keeps the "duplicates" forever (until I restart workers).</p></li>
  <li><p>it appears that in some periods, rebalance happens almost once per several hours, and in other periods it happens every 5 minutes (precisely up to seconds); what could be the reason? what is the normal? </p></li>
  <li><p>what could be the reason for "Group coordinator is unavailable or invalid" errors, given that I use a Confluent Cloud, and <strong>are there any configuration parameters that can be tweaked in Kafka Connect in order to make it more resilient with regards to this error</strong>? I know there are <code>session.timeout.ms</code> and <code>heartbeat.interval.ms</code>, but the <a href="https://docs.confluent.io/current/connect/references/allconfigs.html#distributed-worker-configuration" rel="noreferrer">documentation</a> is so minimalist it is not even clear what is the practical impact of changing these parameters to smaller or bigger values.</p></li>
  </ul>
  
  <p><strong>EDIT 3</strong>:
  I observed that the issue is not critical for sink tasks: although same sink tasks get assigned to multiple workers, corresponding consumers are assigned to different partitions as they normally should, and everything works almost as it should - I simply got more tasks than I originally asked for. However, <strong>in case of source tasks, the behavior is breaking</strong> - tasks run concurrently and compete for resources on the source side. </p>
  
  <p><strong>EDIT 4</strong>: 
  Meanwhile, I downgraded Kafka Connect to version 2.2 (Confluent Platform 5.2.3) - a pre-<a href="https://www.confluent.io/blog/incremental-cooperative-rebalancing-in-kafka" rel="noreferrer">"Incremental Cooperative Rebalancing"</a> version. It works fine for last 2 days. So, I assume the behavior is related to the new rebalancing mechanism. </p>