...

2.1 Consume-Transform-Produce (CTP)

Ensures that output records and source acknowledgments are committed as a single atomic unit within a Kafka-to-Kafka pipeline.

• The Flow: beginTransaction() → send(output) → sendShareAcksToTransaction() → commitTransaction().

• Result: Output is visible and source records are finalized only if both operations succeed.

2.2 Source-Only Frameworks (No Kafka Producer)

For applications writing to external systems (Databases, S3) that require transactional acknowledgments coordinated with their own internal checkpoints.

• Tool: KIP-1302 General Transaction Session

• The Flow: Aggregates acks and commits them via a background transaction once the external sink confirms the write.

2.3 End-to-End Exactly-Once (Flink/Spark/OLAP)

Integrates Share Groups into the two-phase commit (2PC) lifecycle of streaming engines.

• Pre-commit: Sink records are flushed and share acks are added to the transaction.

• Snapshot: Transactional metadata is saved to the framework state.

• Commit: On checkpoint completion, the transaction is finalized.

• Recovery: If the framework fails, the transaction is aborted; output is rolled back, and the broker automatically reverts share records to AVAILABLE for redelivery.

3 Public Interfaces

A new method is added to KafkaProducer to mirror traditional offset commits:

• Method: sendShareAcksToTransaction(Map<TopicPartition, ShareAcknowledgements>, ShareGroupMetadata)

• Why a new method? Unlike offsets, share acks target __share_group_state (not __consumer_offsets) and are managed by the ShareCoordinator (not the GroupCoordinator).

KIP-1310: General Transaction Session#4.5Custom2PCCoordinators

...

7. Compatibility, Deprecation, and Migration Plan

Impact on Existing Users

• If older broker doesn’t support TxnShareAcknowledgeRequest, fallback to current explicit/implicit ack with warning.
• Clients must fail fast on unsupported brokers
• No breaking changes for existing Share Group users
• `transactional` mode is opt-in via `share.acknowledgement.mode` config
• Existing `implicit` and `explicit` modes continue to work unchanged

Migration Path

1. Phase 1: Add `transactional` acknowledgement mode (backward compatible)
2. Phase 2: Streaming frameworks (Flink, Spark) implement transactional sources
3. Phase 3: Documentation and best practices for exactly-once semantics

Deprecation

- No deprecation of existing modes planned
- `transactional` mode recommended for exactly-once use cases

8. Test Plan

Unit Tests

- Transaction state machine transitions
- Idempotency of all transaction operations
- Timeout handling and auto-abort
- Record state transitions (ACQUIRED → ACKNOWLEDGED on commit, ACQUIRED → AVAILABLE on abort/timeout)

Integration Tests

- End-to-end transaction commit/abort flows
- Multi-consumer transactions within same group
- Coordinator failure and recovery
- Network partition scenarios

System Tests

- Flink checkpoint integration with transactional Share Groups
- Exactly-once delivery verification under failures
- Performance benchmarks comparing transactional vs non-transactional modes
- Stress testing with concurrent transactions

Chaos Tests

- Broker failure during PREPARE state
- Coordinator crash before/after prepare
- Consumer crash during transaction
- Network partitions between coordinator and brokers

Include exactly‑once recovery tests similar to Flink e2e:

...

• Zero Breaking Changes: Current implicit and explicit modes remain the default and continue to function as-is.

• Opt-in Requirement: Users must explicitly enable the new mode via share.acknowledgement.mode = transactional.

• Version Safety: * Clients will "fail fast" if the connected broker does not support the new TxnShareAcknowledgeRequest.

  • Fallback logic is included to issue warnings if a downgrade to non-transactional acks is necessary.

8. Test Plan

The verification strategy focuses on state machine integrity and fault tolerance under high-concurrency and failure scenarios.

• Unit Tests: Validates state transitions (e.g., ACQUIRED to ACKNOWLEDGED on commit vs. AVAILABLE on abort), idempotency of operations, and transaction timeout/auto-abort logic.

• Integration Tests: Focuses on end-to-end commit/abort flows, coordinator recovery, and multi-consumer behavior within a single group during network partitions.

• System & Performance Tests: Benchmarks transactional vs. non-transactional modes and verifies exactly-once delivery within Flink/Spark checkpointing cycles.

• Chaos Tests: Simulates broker and coordinator crashes specifically during critical phases like PREPARE_COMMIT to ensure protocol durability

...

• .

Rejected Alternatives

If there are alternative ways of accomplishing the same thing, what were they? The purpose of this section is to motivate why the design is the way it is and not some other way.