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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
