apache / apache/iceberg

Kafka Connect: Coord consumer group on iceberg control topic with phantom consumer after stopping a connector

Open
#17,349 0 comments 0 reactions 0 assignees View on GitHub
bug
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.