DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
...
`WorkerShareSinkTask` Lifecycle
Initialization
The setup phase is simplified because Share Groups eliminate partition-level management.
Consumer Creation: Instantiates
KafkaShareConsumer.Subscription: Subscribes to topics directly (no
RebalanceListenerneeded).Task Startup: Calls
task.initialize()andtask.start(). Unlike traditional sinks,task.open()is called once at startup for all topics since there are no rebalances.
Main Loop (Iteration)
The worker drives the at-least-once delivery guarantee by following this execution flow:
Poll: Pulls records from the share group.
Convert: Transforms messages into
SinkRecords.Deliver: Passes records to the connector via
task.put().Acknowledge:
Success: Marks all records as
ACCEPT.Retriable Error: Marks records as
RELEASEfor immediate broker re-delivery.Fatal Error: Marks records as
REJECT(routes to Dead Letter Queue if configured).
Commit: Calls
shareConsumer.commitSync()to durably persist the acknowledgments on the broker.
Ensuring No Data Loss (At-Least-Once)
he system guarantees that no data is lost by ensuring a record is only acknowledged (ACCEPT) after task.put() returns successfully.
Failure Recovery: If a task crashes before acknowledging, the record remains in an
ACQUIREDstate until theshare.acquisition.lock.timeout.msexpires.The broker then makes the record
AVAILABLEfor another task.Durable Commits: The default explicit mode uses
commitSync()to ensure acknowledgments are persistent. If a commit fails, the record is re-delivered.Duplicates: Occasional duplicates may occur if a crash happens after
task.put()but before acknowledgment. This is standard for at-least-once delivery and is best handled by idempotent sinks (e.g., upserts).
Exactly-Once Semantics (Future Phase, requires KIP-1289)
For Kafka-to-Kafka pipelines, exactly-once delivery is achieved by atomizing the producer's records and the consumer's acknowledgments within a single transaction via KIP-1289.
| Step | Action |
| 1. Begin | Start producer transaction. |
| 2. Produce | Send transformed records to output topics. |
| 3. Bind | Call producer.sendShareAcksToTransaction() to link share acks to the transaction. |
| 4. Commit | Atomically commit both output records and source acknowledgments. |
Configuration Resolution Order
The proposal reuses the existing consumer.override.* mechanism in Kafka Connect for a seamless transition.
Resolution Order:
Worker Config:
consumer.group.protocol=share(Global default).Connector Config:
consumer.override.group.protocol=share(Per-connector override).
Important Note on Group IDs: Share groups use a specific state topic (
__share_group_state).To avoid membership conflicts, users must ensure that a Share Group ID does not match an existing Consumer Group ID. A validation check will be implemented to prevent this collision.
4. Compatibility, Deprecation, and Migration Plan
...
Impact on
...
Users
...
: KIP-1302 is opt-in only
...
4.2 Migration Path
...
and purely additive. Standard consumer groups remain the default, and zero code changes are required for existing connectors.
Migration Path:
Ensure brokers are version 4.0+.
Set
group.protocol=shareat the worker level (all connectors) or per-connector JSON.Tune
share.acquisition.lock.timeout.ms
...
to exceed your highest expected
task.put()
...
latency
...
4.3 Rollback
...
.
...
Rollback: Removing the
shareconfig reverts the connector to standard consumer groups.Note
...
(which may be behind the Share Group's position).
4.4 Deprecation
...
: Consumer groups and Share Groups track progress separately; a rollback may result in processing records that were already handled by the Share Group.
5. Test Plan
5.1 Unit Tests
...
| Test Level | Key Objectives |
| Unit | Verify the poll-put-acknowledge loop |
...
(ACCEPT/RELEASE/REJECT) and config resolution. | |
| Integration | Test elastic scaling (adding/removing tasks), task failure recovery, and interoperability between share and traditional connectors. |
| System | Performance benchmarking against traditional consumer groups and chaos testing to verify zero data loss. |
6. Rejected Alternatives
2. `WorkerTest` (modified): Verify that `baseConsumerConfigs()` returns correct configs for `group.protocol=share`.
3. `SinkConnectorConfigTest` (modified): Validate the new configuration properties and their defaults.
5.2 Integration Tests
1. Basic Share Group Sink: Deploy a sink connector with `group.protocol=share` and verify all records are delivered.
2. Elastic Scaling: Start with 2 tasks, scale to 6, verify no records are lost and throughput increases.
3. Task Failure and Re-delivery: Kill a task mid-processing, verify records are re-delivered to surviving tasks within `acquisition.lock.timeout.ms`.
4. No Duplicate Loss: Produce N records, consume with at-least-once Share Group sink, verify received count >= N.
5. Interoperability: Verify that standard consumer group connectors and Share Group connectors can coexist in the same Connect cluster.
5.3 System Tests
...
6. Rejected Alternatives
| Alternative |
...
| Reason for Rejection | |
Modify SinkTask API | Adding explicit acknowledge() methods would break backward compatibility |
...
| for all existing |
...
| connectors |
...
Alternative 2: Use Share Groups Only for MirrorMaker2
...
| . |
...
Alternative 3: Exactly-Once from Day One
...
| Limit to MirrorMaker 2 | Generic sinks (S3, JDBC, etc.) benefit just as much from elastic scaling as Kafka-to-Kafka pipelines. |
| Require EOS Initially |
| At-least-once is sufficient for |
...
| most use cases |
...
| , and exactly-once |
...
| is blocked by the pending KIP-1289/KIP-1310. |