apache / apache/iceberg

Improve coordinator election logging in Iceberg Kafka Connect Sink

Open
#12,610 7 comments 1 reaction 0 assignees View on GitHub
improvement stale
Dominant language
Java
Stars
9.2k
Forks
3.5k
Avg merge
2d 11h
Merged PRs (30d)
132

Description

### Feature Request / Improvement

# Improve coordinator election logging in Iceberg Kafka Connect Sink

## Problem Description
The Iceberg Kafka Connect Sink connector fails silently when there's a mismatch between:
- The Kafka Connect consumer group ID (`consumer.group.id`)
- The Iceberg connector control topic group ID (`iceberg.connect.group-id`)

When this mismatch occurs, no coordinator is elected and no commits happen, but there are no clear error messages to help users identify the issue.

## Impact
When group IDs don't match:
1. No coordinator is elected
2. No START_COMMIT events are sent
3. Data is consumed but not committed to Iceberg tables
4. Users have limited visibility into why coordination isn't working

## Steps to Reproduce
The issue can be observed in these scenarios:

### Scenario 1: Explicit Different Group IDs
Configure Kafka Connect sink connector with:
```properties
consumer.group.id=connect-sink-group
iceberg.connect.group-id=different-group-name
```

### Scenario 2: Using Default Group ID (More Common)
Configure Kafka Connect sink connector with only:
```properties
consumer.group.id=connect-sink-group
# iceberg.connect.group-id not set, will use default value 'connect-iceberg-sink'
```

Key code references:
```java
// In CommitterImpl.java - Coordinator election process
private boolean hasLeaderPartition(Collection currentAssignedPartitions) {
ConsumerGroupDescription groupDesc;
try (Admin admin = clientFactory.createAdmin()) {
groupDesc = KafkaUtils.consumerGroupDescription(config.connectGroupId(), admin);
}
...

// In IcebergSinkConfig.java - Default group ID configuration
public static final String CONNECT_GROUP_ID = "iceberg.connect.group-id";
public static final String CONNECT_GROUP_ID_DEFAULT = "connect-iceberg-sink";
```

In both scenarios:
1. Start the connector
2. Observe that:
- Data is being consumed
- No commits are happening
- No clear error messages about group ID mismatch

## Root Cause
The `hasLeaderPartition` method in `CommitterImpl` checks for the control topic's consumer group existence but doesn't validate if it matches the actual Kafka Connect consumer group ID. This leads to a situation where:
1. The connector looks for members in the wrong consumer group (either explicitly configured or default `connect-iceberg-sink`)
2. No coordinator is elected because the group appears empty
3. The actual consumer group (where data is being consumed) is different

## Proposed Solution
1. Add appropriate logging when consumer group is not found
2. Provide clear messages to help users identify potential group ID mismatches

## Compatibility
This change is backwards compatible as it only improves logging without changing any behavior.

### Query engine

None

### Willingness to contribute

- [x] I can contribute this improvement/feature independently
- [x] I would be willing to contribute this improvement/feature with guidance from the Iceberg community
- [ ] I cannot contribute this improvement/feature at this time

Contributor guide

Open the contributing guide

Research direction

Start in CommitterImpl at hasLeaderPartition and trace how KafkaUtils.consumerGroupDescription uses config.connectGroupId(). Read IcebergSinkConfig.java for CONNECT_GROUP_ID and its default, then inspect nearby coordinator-election logging. Done means users receive a clear message when the expected consumer group is absent or mismatched, with no behavior change.

Written by the indexing model from the issue text.

Assessment

Tech stack
java, kafka
Domain
data-engineering, distributed-systems
Issue type
Feature
Difficulty
3/5
Estimated time
1-2 days
Activity status
Active
Clarity
Mostly clear
Newbie friendliness
68/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.