Versions Compared

Old Version 24

changes.mady.by.user Shekhar Rajak

Saved on

compared with

New Version 25

changes.mady.by.user Shekhar Rajak

Saved on

Key

  • This line was added.
  • This line was removed.
  • Formatting was changed.

...

Note: The share groups are only suitable for connectors with idempotent, order-independent processing.

2. Scope

2.1 In Scope (What we are building)

  • New Task Type: Introducing WorkerShareSinkTask to handle Share Group logic without changing existing connector code.

  • Flexible Activation: Toggle queue semantics globally or per-connector via consumer.override.group.protocol=share.

  • Delivery Guarantees: * At-least-once: Standard support for all sink types.

    • Exactly-once: Supported for same-cluster Kafka-to-Kafka paths (requires KIP-1289).

  • Observability: New Share Group metrics (acquisition, release, and rejection rates) integrated into the existing sink-task-metrics group.

2.2 Out of Scope (Future/Separate efforts)

  • Source Connectors: Share Group support is currently for Sinks only (Source support and MirrorMaker 2 are excluded).

  • Cross-Cluster EOS: Exactly-once delivery between different Kafka clusters is not supported in this phase.

  • API Changes: No modifications will be made to the public SinkTask Java API or individual connector codebases.

  • Complex Transactions: External 2PC coordinators and cross-cluster transactional protocols are not addressed.


3. Public Interfaces

3.1 New Configuration Properties

...

Worker-

...

Level (

...

connect-distributed.

...

properties)

...

...

  • consumer.group.protocol

...

  • : Set to share to enable KafkaShareConsumer globally for all sink tasks (Default: consumer).

Connector-Level (Per

...

-connector JSON)

Property
TypePer-connector override.
DefaultDescription
consumer.override.group.protocol
string(inherited from worker)
Inherited
Set to share to opt a
single
specific connector into queue semantics.
share.group.id
string
connect-
<connector-name>
<name>
The
Custom Share Group ID
. Defaults to the same naming convention as consumer groups
; follows standard naming conventions.
share.acknowledgement.mode
string
explicitexplicit:
worker calls acknowledge(ACCEPT)
Acknowledge after task.put()
succeeds
. implicit:
acknowledgments are sent
Acknowledge on the next poll
() (simpler, lower latency, weaker guarantee)
.
share.acquisition.lock.timeout.ms
int
30000
Maximum
Max time a record
remains in ACQUIRED state before the broker releases it for
stays acquired before re-delivery. Must
be greater than the expected
exceed task.put() latency.
share.delivery.semantics
string
at-least-onceToggle between at-least-once
or
and exactly-once
. Exactly-once
(requires KIP-1289
and a transactional producer
).
share.max.delivery.count
int
5
Maximum number of times a record can be
Max re-
delivered
delivery attempts before
being sent
sending to
the
a Dead Letter Queue
(if configured). Maps to Share Group's group.share.record.lock.partition.limit
.

3.2 New / Modified Java Interfaces

...