Re: [I] Kafka Connect: `cg-control` consumer group fails to stabilize on idle/low-volume topics after PR #14395 [iceberg]

["vbhanuchander-lang (via GitHub)" <gi...@apache.org>]

vbhanuchander-lang commented on issue #17193:
URL: https://github.com/apache/iceberg/issues/17193#issuecomment-5274641634

   I looked into this because the symptom is corroborated (#11818 reports the same "constant hanging
   lag for low-volume topics"), but I do not think the stated mechanism can be what is happening.
   Four things in the chain do not match the code. I am posting them because they change what needs
   fixing, not to argue the bug away.
   
   **1. #14395 did not remove a `cg-control` readiness check.** The check it removed was on
   `config.connectGroupId()` — the *Connect* consumer group for the data topic — inside
   `CommitterImpl.hasLeaderPartition`, and its purpose was coordinator **leader election** (deciding
   which task owns the first partition):
   
   ```java
   -    if (groupDesc.state() == ConsumerGroupState.STABLE) {
   -      Collection<MemberDescription> members = groupDesc.members();
   -      if (containsFirstPartition(members, currentAssignedPartitions)) { ... }
   -    }
   ```
   
   It never inspected the control group, so it cannot have been "ensuring the Worker's `cg-control`
   consumer had joined".
   
   **2. `put()` is still called when the data topic is idle.** Step 3 assumes the control consumer's
   `poll()` is not called during the block. It is: in `WorkerSinkTask.poll()`, `deliverMessages()` —
   and therefore `task.put(...)` — is invoked **unconditionally** after `convertMessages(msgs)`, with
   no non-empty guard:
   
   ```java
   convertMessages(msgs);
   deliverMessages();     // -> task.put(new ArrayList<>(messageBatch)), even when empty
   ```
   
   So `Channel.consumeAvailable` still runs on every Connect poll cycle on an idle topic.
   
   **3. The 60s figure is `offset.flush.interval.ms`, not `max.poll.interval.ms`.**
   `WorkerSinkTask.iteration()` polls with the time remaining until the next offset commit:
   
   ```java
   long timeoutMs = Math.max(nextCommit - now, 0);
   poll(timeoutMs);
   ```
   
   `nextCommit` advances by `OFFSET_COMMIT_INTERVAL_MS_CONFIG` (`offset.flush.interval.ms`), whose
   default is `60000L` in `WorkerConfig`. `max.poll.interval.ms` defaults to **300000**, not 60000
   (`ConsumerConfig`, line 607). So the idle gap between `put()` calls is ~60s and bounded by a Connect
   worker setting.
   
   **4. A 60s gap between polls does not trip `session.timeout.ms`.** Since KIP-62 the consumer
   heartbeats from a background thread, so `session.timeout.ms` (default 45000, confirmed in
   `ConsumerConfig`) covers heartbeat loss, not time between `poll()` calls. The config that evicts a
   member for not polling is `max.poll.interval.ms` — 300s here, which a 60s gap does not breach. That
   is also why raising `heartbeat.interval.ms`, as suggested, would not be the lever even if the timing
   were as described.
   
   **What I could not explain, and what would pin it down**
   
   The Worker's control group is created per worker instance as
   `config.controlGroupIdPrefix() + UUID.randomUUID()` (`Worker.java:53`), i.e. a fresh
   **single-member** group named `cg-control-<uuid>`. A single-member group that owns its own group id
   reaching a non-STABLE state indefinitely is genuinely odd, and the `UnknownMemberIdException` you
   saw is real evidence of *something*, so I would rather find the actual cause than hand-wave.
   
   Two things would settle it:
   
   1. The exact group id in the `UnknownMemberIdException` — is it `cg-control-<uuid>`, the
      `<connectGroupId>-coord` group the Coordinator uses (`Coordinator.java:97`), or the Connect group
      itself? These are three different groups and the fix differs for each.
   2. Whether a fresh `cg-control-<uuid>` group is created on every task restart. If restarts are
      frequent, abandoned single-member groups accumulate, and it is worth checking whether the group
      being described as "not STABLE" is a previous incarnation rather than the live one.
   
   If the real problem turns out to be that the coordinator can now start a commit cycle before workers
   have joined — which is a plausible consequence of #14395, just not via `cg-control` liveness — then
   the fix belongs in the leader-election/commit-start path, and I am happy to work on it. But I did
   not want to write a patch against a mechanism I had just shown does not hold.
   
   *(Config defaults and the `WorkerSinkTask` behaviour above were read from the Kafka 3.9 sources,
   matching the 3.9.2 runtime in your environment section.)*
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org