Kafka Connect: Coord consumer group on iceberg control topic with phantom consumer after stopping a connector
- Dominant language
- Java
- Stars
- 9.2k
- Forks
- 3.5k
- Avg merge
- 2d 16h
- Merged PRs (30d)
- 129
Description
### Apache Iceberg version
1.11.0 (latest release)
### Query engine
None
### Please describe the bug 🐞
We have repeatedly witnessed a phantom consumer being left behind after stopping a connector.
For example:
```
GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
connect-dbz-iceberg-xxx-coord iceberg-control-topic 0 2392395 2392398 3 7b6c3837-5f3b-4257-87b8-fba8e28c29d7-edc2df8b-1ec3-4217-9cf8-7ab210f9102e /10.104.165.73 7b6c3837-5f3b-4257-87b8-fba8e28c29d7
```
We are currently running a slightly patched CommitterImpl.java, which seems to fix the issue:
```java
private void stopCoordinator() {
if (coordinatorThread != null) {
coordinatorThread.terminate();
+ try {
+ LOG.info("Asking coordinator thread to join for task {}", taskId);
+ coordinatorThread.join(30000);
+ LOG.info("Coordinator thread joined for task {}", taskId);
+ } catch (InterruptedException e) {
+ LOG.warn("Coordinator thread join interrupted for task {}", taskId, e);
+ }
coordinatorThread = null;
}
}
```
This gives the CoordniatorThread:run() method a proper chance to finish and close resources:
```java
@Override
public void run() {
...
try {
LOG.info("Stopping coordinator");
coordinator.stop();
} catch (Exception e) {
LOG.error("Coordinator error during stop, ignoring", e);
}
}
```
The Channel:stop() method (i.e. coordinator.stop();):
```java
void stop() {
LOG.info("Channel stopping");
producer.close();
consumer.close();
admin.close();
}
```
### Willingness to contribute
- [ ] I can contribute a fix for this bug independently
- [ ] I would be willing to contribute a fix for this bug with guidance from the Iceberg community
- [ ] I cannot contribute a fix for this bug at this time
Contributor guide
Research direction
Start by reading CommitterImpl.java, especially stopCoordinator(), CoordinatorThread.run(), and Channel.stop(). Trace how stopping a connector terminates the coordinator and closes its producer, consumer, and admin resources, then reproduce or test the shutdown path. Done means stopping a connector no longer leaves a phantom consumer in the iceberg-control topic consumer group.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java, kafka
- Domain
- distributed-systems, stream-processing
- Issue type
- Bug
- Difficulty
- 3/5
- Estimated time
- 1-2 days
- Activity status
- Quiet
- Clarity
- Mostly clear
- Newbie friendliness
- 68/100