yixiaodan opened a new pull request, #23995: URL: https://github.com/apache/kafka/pull/23995
### JIRA https://issues.apache.org/jira/browse/KAFKA-21380 ### Problem When the DLQ flow is enabled for a share group (KIP-1191), records whose delivery count reaches the limit are moved to `ARCHIVING` on the normal release path so they are written to the DLQ topic (`InFlightState.tryUpdateState` with `dlqSupportEnabled=true`). However, the rollback path in `InFlightState.completeStateTransition(false)` - taken when a two-phase state transition fails (e.g. the persister write fails) and the acquisition lock has expired - archives the records directly: ```java state = previousState.deliveryCount() >= rollbackState.maxDeliveryCount ? RecordState.ARCHIVED : RecordState.AVAILABLE; ``` This bypasses the DLQ flow entirely: records that should have been dead-lettered are silently archived and never appear in the DLQ topic. Note that `completeStateTransition(boolean)` has no access to the `dlqSupportEnabled` flag, so the rollback path cannot currently reach `ARCHIVING`. Two related inconsistencies: 1. `ARCHIVING` (normal path) vs `ARCHIVED` (rollback path) for the same limit-reached condition. 2. The normal path evaluates the limit with the delivery count including the current delivery (`deliveryCount` incremented before the check), while the rollback path uses `previousState.deliveryCount()` - the counts differ by one. ### Fix - Thread the DLQ-support flag through `completeStateTransition`; on rollback with an expired lock and the delivery count at the limit, records move to `ARCHIVING` when DLQ is enabled (mirroring `tryUpdateState`), otherwise to `ARCHIVED`. - Start the DLQ flow (`initiateDLQAndArchive`) for batches that move to `ARCHIVING` in both the local rollback (`rollbackOrProcessStateUpdates`) and the persister-failure path, outside the partition lock as in the successful path. - Derive the DLQ cause from the delivery count when the batch has no recorded cause. ### Testing - New `InFlightStateTest` (5 cases), which also fills the existing coverage gap for `completeStateTransition`: - normal release path at the limit -> `ARCHIVING` (baseline) - rollback with expired lock at the limit, DLQ enabled -> `ARCHIVING` - rollback with expired lock at the limit, DLQ disabled -> `ARCHIVED` - rollback with expired lock below the limit -> `AVAILABLE` - rollback without an expired lock restores the previous state and member - The DLQ-enabled rollback case fails on trunk and passes with this change. - Regression: `SharePartitionTest` (289), `SharePartitionManagerTest` (60), `ShareGroupDLQStateManagerTest` (71), `ShareGroupDLQRecordFetcherTest` (29) - all pass with the change. - `checkstyle` passes for the changed modules. -- 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]
