Savonitar opened a new pull request, #321:
URL: https://github.com/apache/flink-connector-kafka/pull/321
Imagine Flink writes one record, **A**, to Kafka using:
- Transactional ID: `orders-7`
- Producer ID: `42`
- Producer epoch: `32766`
The **transactional ID stays the same**. The producer ID and epoch let Kafka
distinguish current producers from stale ones.
1. **Flink begins the transaction.**
The producer starts a transaction using identity `(42, 32766)`.
2. **Flink writes record A and flushes it.**
Kafka has received the record, but the transaction is still open. A
consumer using `read_committed` cannot see A yet. **Flushing does not commit
the transaction.**
3. **Flink snapshots the information needed to commit it.**
The checkpoint contains a pending committable:
```text
transactionalId = orders-7
producerId = 42
epoch = 32766
```
The original connector does **not** save that this transaction uses
Kafka’s V2 protocol.
4. **The checkpoint completes, and Flink asks Kafka to commit.**
The original, live producer knows it uses V2, so it sends the appropriate
commit request.
5. **Kafka commits transaction A and advances the producer identity.**
A becomes visible to `read_committed` consumers. Normally, V2 advances
the epoch after completion. At this boundary, Kafka allocates a new producer ID
and resets the epoch:
```text
Completed transaction: producer 42, epoch 32766
Identity for future work: producer 99, epoch 0
```
This does not start another transaction automatically.
6. **Flink crashes before a later checkpoint records that this commit was
processed.**
Kafka has committed A, but the checkpoint Flink restores still contains
the pending committable `(42, 32766)`. This can happen even if Flink received
Kafka’s successful response.
7. **Flink restores the checkpoint and retries the commit.**
This is necessary: checkpoint recovery must handle uncertainty about
whether the earlier commit succeeded. Retrying a completed commit should be
safe.
8. **The recovered producer sends the wrong protocol.**
Flink restores `(42, 32766)` into a fresh producer, but does not restore
the V2 flag. It therefore sends an old-protocol commit request.
Kafka’s V2 handling could recognize the previous identity as a retry. The
old-protocol handling instead rejects producer `42`, because `orders-7` now
belongs to producer `99`:
```text
InvalidPidMappingException
```
9. **Checkpoint recovery fails.**
A subsequent restart can restore the same checkpoint, retry the same old
identity using the same wrong protocol, and fail again.
**Record A is already committed. The failure is that Flink cannot
successfully recover and continue processing.**
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
