You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "smjn (via GitHub)" <gi...@apache.org> on 2026/04/17 09:26:26 UTC
[PR] KAFKA-20245: DLQ records exceeding max delivery count. [kafka]
smjn opened a new pull request, #22080:
URL: https://github.com/apache/kafka/pull/22080
* Changes to support DLQ of offsets and batches which have exceeded max delivery count.
* The Phase 2 part of the transition to DLQ (->ARCHIVED (in mem and persist)) has been refactored into a separate method.
* The current code does not transactionally handle state transitions on max delivery count exceeded. ARCHIVING follows a similar approach.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20245: DLQ records exceeding max delivery count. [3/N] [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #22080:
URL: https://github.com/apache/kafka/pull/22080#discussion_r3136593929
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -2567,75 +2584,14 @@ void rollbackOrProcessStateUpdates(
dlqBatches.forEach(persisterBatch -> {
persisterBatch.updatedState.completeStateTransition(true);
- shareGroupDLQ.enqueue(new ShareGroupDLQRecordParameter(
Review Comment:
Can we please move this as well out of the nested iterations in a separate method.
##########
server/src/main/java/org/apache/kafka/server/share/fetch/InFlightState.java:
##########
@@ -161,7 +165,7 @@ public InFlightState tryUpdateState(RecordState newState, DeliveryCountOps ops,
}
if (newState == RecordState.AVAILABLE && ops != DeliveryCountOps.DECREASE && deliveryCount >= maxDeliveryCount) {
- newState = RecordState.ARCHIVED;
+ newState = shareGroupDlqEnableSupplier.get() ? RecordState.ARCHIVING : RecordState.ARCHIVED;
Review Comment:
What happens when the share version for DLQ is toggled from true to false i.e. can there be some hanging archiving states?
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -2567,75 +2584,14 @@ void rollbackOrProcessStateUpdates(
dlqBatches.forEach(persisterBatch -> {
persisterBatch.updatedState.completeStateTransition(true);
Review Comment:
Hmmm, so we track the non-dlq batches and dlq batches but then completeStateTransition for all the batches so why do we need to track them separately? Don't we only need to know the dlq batches and start shareGroupDLQ.enqueue, for rest all persister batches we complete the state transition? This is as per current code written.
##########
server/src/main/java/org/apache/kafka/server/share/fetch/InFlightState.java:
##########
@@ -53,17 +54,20 @@ public class InFlightState {
// The boolean determines if the record has achieved a terminal state of ARCHIVED from which it cannot transition
// to any other state. This could happen because of LSO movement etc.
private boolean isTerminalState = false;
+ // Supplier indicating share group DLQ support.
+ private final Supplier<Boolean> shareGroupDlqEnableSupplier;
Review Comment:
Can we avoid having a supplier here as the most instances in share partition are created are for InflightState hence the idea is to always keep this class as lean as possible?
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -2567,75 +2584,14 @@ void rollbackOrProcessStateUpdates(
dlqBatches.forEach(persisterBatch -> {
persisterBatch.updatedState.completeStateTransition(true);
- shareGroupDLQ.enqueue(new ShareGroupDLQRecordParameter(
- groupId,
- topicIdPartition,
+ processDlqPhase2(
+ persisterBatch.updatedState,
Review Comment:
Shouldn't dlq processing be post completion of current inflight request i.e. once the persister for persisting state is successful the future should be completed. The calls for DLQ whether successful or error do not alter the result of share fetch/acknowledge calls hence they can be triggered post current ongoing request i.e. future completion. What do you think?
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -2567,75 +2584,14 @@ void rollbackOrProcessStateUpdates(
dlqBatches.forEach(persisterBatch -> {
persisterBatch.updatedState.completeStateTransition(true);
- shareGroupDLQ.enqueue(new ShareGroupDLQRecordParameter(
- groupId,
- topicIdPartition,
+ processDlqPhase2(
Review Comment:
Can we name it better or write in comments what are the phases please.
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -1180,8 +1181,10 @@ private Optional<Throwable> releaseAcquiredRecordsForCompleteBatch(String member
}
// Successfully updated the state of the batch and created a persister state batch for write to persister.
+ // If DLQ support is enabled, the update
Review Comment:
Seems the comment is not completed. Can you please write the details here please.
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -2567,75 +2584,14 @@ void rollbackOrProcessStateUpdates(
dlqBatches.forEach(persisterBatch -> {
persisterBatch.updatedState.completeStateTransition(true);
- shareGroupDLQ.enqueue(new ShareGroupDLQRecordParameter(
- groupId,
- topicIdPartition,
+ processDlqPhase2(
Review Comment:
Do you need to call this method while in a write lock, if not please take it out of the lock.
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -3038,8 +3008,11 @@ && checkForStartOffsetWithinBatch(inFlightBatch.firstOffset(), inFlightBatch.las
};
}
+ private record DlqTimeoutBatch(InFlightState updatedState, long firstOffset, long lastOffset, short deliveryCount) { }
Review Comment:
Please move it after `private record LastOffsetAndMaxRecords` where all other record classes are defined for readability.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20245: DLQ records exceeding max delivery count. [3/N] [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #22080:
URL: https://github.com/apache/kafka/pull/22080#discussion_r3159651205
##########
server/src/main/java/org/apache/kafka/server/share/fetch/InFlightState.java:
##########
@@ -161,7 +165,7 @@ public InFlightState tryUpdateState(RecordState newState, DeliveryCountOps ops,
}
if (newState == RecordState.AVAILABLE && ops != DeliveryCountOps.DECREASE && deliveryCount >= maxDeliveryCount) {
- newState = RecordState.ARCHIVED;
+ newState = shareGroupDlqEnableSupplier.get() ? RecordState.ARCHIVING : RecordState.ARCHIVED;
Review Comment:
Do we ever persist Archiving state in Persister? If yes then this could happen, right?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20245: DLQ records exceeding max delivery count. [3/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on PR #22080:
URL: https://github.com/apache/kafka/pull/22080#issuecomment-4279806122
Case where client continuously RELEASEs records
### Broker
```
[2026-04-20 15:47:53,364] WARN The offset 0 is on last delivery attempt in share partition: gs1-3muqorVvQoaaAcYfittiVA:t1-0, should be delivered alone in this fetch (kafka.server.share.SharePartition)
[2026-04-20 15:47:55,087] WARN Enqueuing share group dlq record parameter: ShareGroupDLQRecordParameter[groupId=gs1, topicIdPartition=3muqorVvQoaaAcYfittiVA:t1-0, firstOffset=0, lastOffset=0, deliveryCount=Optional[5], cause=Optional[org.apache.kafka.server.share.dlq.ShareGroupDLQ$ShareGroupDLQThrowable: Offset delivery count exceeded the threshold.], preserveRecordData=false] (org.apache.kafka.server.share.dlq.NoOpShareGroupDLQManager)
[2026-04-20 15:47:56,354] WARN The offset 1 is on last delivery attempt in share partition: gs1-3muqorVvQoaaAcYfittiVA:t1-0, should be delivered alone in this fetch (kafka.server.share.SharePartition)
```
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20245: DLQ records exceeding max delivery count. [3/N] [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 merged PR #22080:
URL: https://github.com/apache/kafka/pull/22080
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20245: DLQ records exceeding max delivery count. [3/N] [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #22080:
URL: https://github.com/apache/kafka/pull/22080#discussion_r3194640438
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -3336,6 +3325,83 @@ private RecordState recordStateWithDlq(byte ackType) {
return ACK_TYPE_TO_RECORD_STATE.get(ackType);
}
+ // Visible for testing.
+
+ /**
+ * The DLQ flow comprises 2 phases:
+ * Phase 1: State transitions to ARCHIVING (happens in the normal acknowledge/release/timeout path) and persists ARCHIVING to the persister
+ * Phase 2: Enqueues to DLQ, then transitions ARCHIVING → ARCHIVED and persists ARCHIVED to the persister
+ * This method handles the complete phase 2 flow.
+ */
+ void initiateDLQAndArchive(InFlightState updatedState, long firstOffset,
+ long lastOffset, short deliveryCount, Throwable dlqCause,
+ boolean isTimeout) {
+ // Step 1: Enqueue to DLQ
+ shareGroupDLQ.enqueue(new ShareGroupDLQRecordParameter(
+ groupId, topicIdPartition, firstOffset, lastOffset,
+ Optional.of(deliveryCount), Optional.ofNullable(dlqCause), false
+ )).whenComplete((v1, dlqException) -> {
+ if (dlqException != null) {
+ log.error("Failed to write to DLQ ... Proceeding to ARCHIVED regardless.");
+ }
+
+ // Step 2: Transition ARCHIVING → ARCHIVED
+ // For timeout path, use tryUpdateState (no rollback) to match non-DLQ behavior
Review Comment:
I think we should re-visit this once we have more handling related to read is added. I will merge the PR for now and then let's iterate on it again.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20245: DLQ records exceeding max delivery count. [3/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #22080:
URL: https://github.com/apache/kafka/pull/22080#discussion_r3187501763
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -3336,6 +3325,83 @@ private RecordState recordStateWithDlq(byte ackType) {
return ACK_TYPE_TO_RECORD_STATE.get(ackType);
}
+ // Visible for testing.
+
+ /**
+ * The DLQ flow comprises 2 phases:
+ * Phase 1: State transitions to ARCHIVING (happens in the normal acknowledge/release/timeout path) and persists ARCHIVING to the persister
+ * Phase 2: Enqueues to DLQ, then transitions ARCHIVING → ARCHIVED and persists ARCHIVED to the persister
+ * This method handles the complete phase 2 flow.
+ */
+ void initiateDLQAndArchive(InFlightState updatedState, long firstOffset,
+ long lastOffset, short deliveryCount, Throwable dlqCause,
+ boolean isTimeout) {
+ // Step 1: Enqueue to DLQ
+ shareGroupDLQ.enqueue(new ShareGroupDLQRecordParameter(
+ groupId, topicIdPartition, firstOffset, lastOffset,
+ Optional.of(deliveryCount), Optional.ofNullable(dlqCause), false
+ )).whenComplete((v1, dlqException) -> {
+ if (dlqException != null) {
+ log.error("Failed to write to DLQ ... Proceeding to ARCHIVED regardless.");
+ }
+
+ // Step 2: Transition ARCHIVING → ARCHIVED
+ // For timeout path, use tryUpdateState (no rollback) to match non-DLQ behavior
+ // where ARCHIVED is never reverted on persist failure.
+ // For acknowledge/release path, use startStateTransition (with rollback) so the
+ // client can retry on persist failure.
+ PersisterStateBatch stateBatch;
+ lock.writeLock().lock();
+ try {
+ InFlightState updateResult = isTimeout
+ ? updatedState.tryUpdateState(RecordState.ARCHIVED, DeliveryCountOps.NO_OP, maxDeliveryCount(), EMPTY_MEMBER_ID, shareGroupDlqEnableSupplier.get())
+ : updatedState.startStateTransition(RecordState.ARCHIVED, DeliveryCountOps.NO_OP, maxDeliveryCount(), EMPTY_MEMBER_ID, shareGroupDlqEnableSupplier.get());
+ if (updateResult == null) {
+ log.error("Unable to transition ARCHIVING → ARCHIVED ...");
+ return;
+ }
+ stateBatch = new PersisterStateBatch(
+ firstOffset, lastOffset,
+ updateResult.state().id(), (short) updateResult.deliveryCount());
+ deliveryCompleteCount.addAndGet(
+ numInFlightRecordsInBatch(firstOffset, lastOffset));
+ } finally {
+ lock.writeLock().unlock();
+ }
+
+ // Step 3: Persist ARCHIVED
+ writeShareGroupState(List.of(stateBatch))
+ .whenComplete((v2, phase2Exception) -> {
+ boolean cacheUpdated = false;
+ lock.writeLock().lock();
+ try {
+ if (phase2Exception != null) {
+ if (!isTimeout) {
+ // For acknowledge/release path, rollback the state transition.
+ updatedState.completeStateTransition(false);
+ if (isStateTerminal(RecordState.forId(stateBatch.deliveryState()))
+ && !isStateTerminal(updatedState.state())) {
+ deliveryCompleteCount.addAndGet(
+ -numInFlightRecordsInBatch(firstOffset, lastOffset));
+ }
+ return;
+ }
+ // For timeout path, no rollback — ARCHIVED stays in memory,
+ // consistent with non-DLQ timeout behavior. Fall through to
+ // update cached state so the batch can be evicted.
+ }
+ if (!isTimeout) {
+ updatedState.completeStateTransition(true);
+ }
+ cacheUpdated = maybeUpdateCachedStateAndOffsets();
+ } finally {
+ lock.writeLock().unlock();
+ maybeCompleteDelayedShareFetchRequest(cacheUpdated);
Review Comment:
maybeUpdateCachedStateAndOffsets:
- So that ARCHIVED records can be evicted and LSO moved (this call is happening in async context whenComplete from persister)
maybeCompleteDelayedShareFetchRequest
- So that we can start honoring any calls blocked on the LSO movement.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20245: DLQ records exceeding max delivery count. [3/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #22080:
URL: https://github.com/apache/kafka/pull/22080#discussion_r3152593706
##########
server/src/main/java/org/apache/kafka/server/share/fetch/InFlightState.java:
##########
@@ -161,7 +165,7 @@ public InFlightState tryUpdateState(RecordState newState, DeliveryCountOps ops,
}
if (newState == RecordState.AVAILABLE && ops != DeliveryCountOps.DECREASE && deliveryCount >= maxDeliveryCount) {
- newState = RecordState.ARCHIVED;
+ newState = shareGroupDlqEnableSupplier.get() ? RecordState.ARCHIVING : RecordState.ARCHIVED;
Review Comment:
Hi @apoorvmittal10 ,
Thanks for the insightful comments. In this case, the hang case does not arise.
Consider the possibilities:
* dlq is enabled -> newState = ARCHIVING -> phase 2 will move ARCHIVING to ARCHIVED irrespective of dlq status (this was deliberate)
* dlq is disabled -> newState = ARCHIVED, so DLQ flow is not triggered
* dlq is enabled -> newState = ARCHIVING -> dlq is disabled -> the phase 2 (move to ARCHIVED) will still run for this batch/record.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20245: DLQ records exceeding max delivery count. [3/N] [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #22080:
URL: https://github.com/apache/kafka/pull/22080#discussion_r3183796096
##########
server/src/main/java/org/apache/kafka/server/share/fetch/InFlightState.java:
##########
@@ -161,7 +165,7 @@ public InFlightState tryUpdateState(RecordState newState, DeliveryCountOps ops,
}
if (newState == RecordState.AVAILABLE && ops != DeliveryCountOps.DECREASE && deliveryCount >= maxDeliveryCount) {
- newState = RecordState.ARCHIVED;
+ newState = shareGroupDlqEnableSupplier.get() ? RecordState.ARCHIVING : RecordState.ARCHIVED;
Review Comment:
Sounds good, probably will discuss again when we add the readState handling.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20245: DLQ records exceeding max delivery count. [3/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #22080:
URL: https://github.com/apache/kafka/pull/22080#discussion_r3165969784
##########
server/src/main/java/org/apache/kafka/server/share/fetch/InFlightState.java:
##########
@@ -161,7 +165,7 @@ public InFlightState tryUpdateState(RecordState newState, DeliveryCountOps ops,
}
if (newState == RecordState.AVAILABLE && ops != DeliveryCountOps.DECREASE && deliveryCount >= maxDeliveryCount) {
- newState = RecordState.ARCHIVED;
+ newState = shareGroupDlqEnableSupplier.get() ? RecordState.ARCHIVING : RecordState.ARCHIVED;
Review Comment:
We persist - but above check is only when we transition to some(state) => ARCHIVING. The logic is not applicable to records which are already in ARCHIVING (persisted). If something is already in ARCHIVING then the flag does not affect.
Additional code to handle ARCHIVING state records received in leadership change is pending.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20245: DLQ records exceeding max delivery count. [3/N] [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on PR #22080:
URL: https://github.com/apache/kafka/pull/22080#issuecomment-4387766833
@apoorvmittal10 Could you take a look https://github.com/apache/kafka/pull/22222 ? Thanks
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20245: DLQ records exceeding max delivery count. [3/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #22080:
URL: https://github.com/apache/kafka/pull/22080#discussion_r3165969784
##########
server/src/main/java/org/apache/kafka/server/share/fetch/InFlightState.java:
##########
@@ -161,7 +165,7 @@ public InFlightState tryUpdateState(RecordState newState, DeliveryCountOps ops,
}
if (newState == RecordState.AVAILABLE && ops != DeliveryCountOps.DECREASE && deliveryCount >= maxDeliveryCount) {
- newState = RecordState.ARCHIVED;
+ newState = shareGroupDlqEnableSupplier.get() ? RecordState.ARCHIVING : RecordState.ARCHIVED;
Review Comment:
We persist - but above check is only when we transition to some(state) => ARCHIVING. The logic is not applicable to records which are already in ARCHIVING (persisted). If something is already in ARCHIVING then the flag does not affect.
Additional code to handle ARCHIVING state records received in leadership change is pending.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20245: DLQ records exceeding max delivery count. [3/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #22080:
URL: https://github.com/apache/kafka/pull/22080#discussion_r3187570800
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -3336,6 +3325,83 @@ private RecordState recordStateWithDlq(byte ackType) {
return ACK_TYPE_TO_RECORD_STATE.get(ackType);
}
+ // Visible for testing.
+
+ /**
+ * The DLQ flow comprises 2 phases:
+ * Phase 1: State transitions to ARCHIVING (happens in the normal acknowledge/release/timeout path) and persists ARCHIVING to the persister
+ * Phase 2: Enqueues to DLQ, then transitions ARCHIVING → ARCHIVED and persists ARCHIVED to the persister
+ * This method handles the complete phase 2 flow.
+ */
+ void initiateDLQAndArchive(InFlightState updatedState, long firstOffset,
+ long lastOffset, short deliveryCount, Throwable dlqCause,
+ boolean isTimeout) {
+ // Step 1: Enqueue to DLQ
+ shareGroupDLQ.enqueue(new ShareGroupDLQRecordParameter(
+ groupId, topicIdPartition, firstOffset, lastOffset,
+ Optional.of(deliveryCount), Optional.ofNullable(dlqCause), false
+ )).whenComplete((v1, dlqException) -> {
+ if (dlqException != null) {
+ log.error("Failed to write to DLQ ... Proceeding to ARCHIVED regardless.");
+ }
+
+ // Step 2: Transition ARCHIVING → ARCHIVED
+ // For timeout path, use tryUpdateState (no rollback) to match non-DLQ behavior
Review Comment:
The issue with ARCHIVING -> ARCHIVED is that it is not fetchable or timeoutable. Once the state moves from REJECT/ACQUIRED -> ARCHIVING, the acquisitionlocktimeout is also cancelled. So, we do not get the opportunity to retry on client request or timeouts.
Hence, the behavior in both cases should be the same, which is non-transactional state transition to ARCHIVED. Now, if ARCHIVED persistance fails - 2 things can happen
1. At some point the LSO movement evicts the batch
2. The leadership changes and the record is read as ARCHIVING from the persister and DLQ will be retried.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20245: DLQ records exceeding max delivery count. [3/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on PR #22080:
URL: https://github.com/apache/kafka/pull/22080#issuecomment-4378414719
@apoorvmittal10 Thanks for the additional comments, incorporated.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20245: DLQ records exceeding max delivery count. [3/N] [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #22080:
URL: https://github.com/apache/kafka/pull/22080#discussion_r3183850801
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -3336,6 +3325,83 @@ private RecordState recordStateWithDlq(byte ackType) {
return ACK_TYPE_TO_RECORD_STATE.get(ackType);
}
+ // Visible for testing.
+
+ /**
+ * The DLQ flow comprises 2 phases:
+ * Phase 1: State transitions to ARCHIVING (happens in the normal acknowledge/release/timeout path) and persists ARCHIVING to the persister
+ * Phase 2: Enqueues to DLQ, then transitions ARCHIVING → ARCHIVED and persists ARCHIVED to the persister
+ * This method handles the complete phase 2 flow.
+ */
+ void initiateDLQAndArchive(InFlightState updatedState, long firstOffset,
+ long lastOffset, short deliveryCount, Throwable dlqCause,
+ boolean isTimeout) {
+ // Step 1: Enqueue to DLQ
+ shareGroupDLQ.enqueue(new ShareGroupDLQRecordParameter(
+ groupId, topicIdPartition, firstOffset, lastOffset,
+ Optional.of(deliveryCount), Optional.ofNullable(dlqCause), false
+ )).whenComplete((v1, dlqException) -> {
+ if (dlqException != null) {
+ log.error("Failed to write to DLQ ... Proceeding to ARCHIVED regardless.");
Review Comment:
```suggestion
log.error("Failed to write to DLQ, proceeding to ARCHIVED regardless.", dlqException);
```
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -3336,6 +3325,83 @@ private RecordState recordStateWithDlq(byte ackType) {
return ACK_TYPE_TO_RECORD_STATE.get(ackType);
}
+ // Visible for testing.
+
+ /**
+ * The DLQ flow comprises 2 phases:
+ * Phase 1: State transitions to ARCHIVING (happens in the normal acknowledge/release/timeout path) and persists ARCHIVING to the persister
+ * Phase 2: Enqueues to DLQ, then transitions ARCHIVING → ARCHIVED and persists ARCHIVED to the persister
+ * This method handles the complete phase 2 flow.
+ */
+ void initiateDLQAndArchive(InFlightState updatedState, long firstOffset,
+ long lastOffset, short deliveryCount, Throwable dlqCause,
+ boolean isTimeout) {
+ // Step 1: Enqueue to DLQ
+ shareGroupDLQ.enqueue(new ShareGroupDLQRecordParameter(
+ groupId, topicIdPartition, firstOffset, lastOffset,
+ Optional.of(deliveryCount), Optional.ofNullable(dlqCause), false
+ )).whenComplete((v1, dlqException) -> {
+ if (dlqException != null) {
+ log.error("Failed to write to DLQ ... Proceeding to ARCHIVED regardless.");
+ }
+
+ // Step 2: Transition ARCHIVING → ARCHIVED
+ // For timeout path, use tryUpdateState (no rollback) to match non-DLQ behavior
+ // where ARCHIVED is never reverted on persist failure.
+ // For acknowledge/release path, use startStateTransition (with rollback) so the
+ // client can retry on persist failure.
+ PersisterStateBatch stateBatch;
+ lock.writeLock().lock();
+ try {
+ InFlightState updateResult = isTimeout
+ ? updatedState.tryUpdateState(RecordState.ARCHIVED, DeliveryCountOps.NO_OP, maxDeliveryCount(), EMPTY_MEMBER_ID, shareGroupDlqEnableSupplier.get())
+ : updatedState.startStateTransition(RecordState.ARCHIVED, DeliveryCountOps.NO_OP, maxDeliveryCount(), EMPTY_MEMBER_ID, shareGroupDlqEnableSupplier.get());
+ if (updateResult == null) {
+ log.error("Unable to transition ARCHIVING → ARCHIVED ...");
+ return;
+ }
+ stateBatch = new PersisterStateBatch(
+ firstOffset, lastOffset,
+ updateResult.state().id(), (short) updateResult.deliveryCount());
+ deliveryCompleteCount.addAndGet(
+ numInFlightRecordsInBatch(firstOffset, lastOffset));
+ } finally {
+ lock.writeLock().unlock();
+ }
+
+ // Step 3: Persist ARCHIVED
+ writeShareGroupState(List.of(stateBatch))
+ .whenComplete((v2, phase2Exception) -> {
+ boolean cacheUpdated = false;
+ lock.writeLock().lock();
+ try {
+ if (phase2Exception != null) {
+ if (!isTimeout) {
+ // For acknowledge/release path, rollback the state transition.
+ updatedState.completeStateTransition(false);
+ if (isStateTerminal(RecordState.forId(stateBatch.deliveryState()))
+ && !isStateTerminal(updatedState.state())) {
+ deliveryCompleteCount.addAndGet(
+ -numInFlightRecordsInBatch(firstOffset, lastOffset));
+ }
+ return;
+ }
+ // For timeout path, no rollback — ARCHIVED stays in memory,
+ // consistent with non-DLQ timeout behavior. Fall through to
+ // update cached state so the batch can be evicted.
+ }
+ if (!isTimeout) {
+ updatedState.completeStateTransition(true);
+ }
+ cacheUpdated = maybeUpdateCachedStateAndOffsets();
+ } finally {
+ lock.writeLock().unlock();
+ maybeCompleteDelayedShareFetchRequest(cacheUpdated);
Review Comment:
Why this call is required here?
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -3336,6 +3325,83 @@ private RecordState recordStateWithDlq(byte ackType) {
return ACK_TYPE_TO_RECORD_STATE.get(ackType);
}
+ // Visible for testing.
+
+ /**
+ * The DLQ flow comprises 2 phases:
+ * Phase 1: State transitions to ARCHIVING (happens in the normal acknowledge/release/timeout path) and persists ARCHIVING to the persister
+ * Phase 2: Enqueues to DLQ, then transitions ARCHIVING → ARCHIVED and persists ARCHIVED to the persister
+ * This method handles the complete phase 2 flow.
+ */
+ void initiateDLQAndArchive(InFlightState updatedState, long firstOffset,
+ long lastOffset, short deliveryCount, Throwable dlqCause,
+ boolean isTimeout) {
+ // Step 1: Enqueue to DLQ
+ shareGroupDLQ.enqueue(new ShareGroupDLQRecordParameter(
+ groupId, topicIdPartition, firstOffset, lastOffset,
+ Optional.of(deliveryCount), Optional.ofNullable(dlqCause), false
+ )).whenComplete((v1, dlqException) -> {
+ if (dlqException != null) {
+ log.error("Failed to write to DLQ ... Proceeding to ARCHIVED regardless.");
+ }
+
+ // Step 2: Transition ARCHIVING → ARCHIVED
+ // For timeout path, use tryUpdateState (no rollback) to match non-DLQ behavior
+ // where ARCHIVED is never reverted on persist failure.
+ // For acknowledge/release path, use startStateTransition (with rollback) so the
+ // client can retry on persist failure.
+ PersisterStateBatch stateBatch;
+ lock.writeLock().lock();
+ try {
+ InFlightState updateResult = isTimeout
+ ? updatedState.tryUpdateState(RecordState.ARCHIVED, DeliveryCountOps.NO_OP, maxDeliveryCount(), EMPTY_MEMBER_ID, shareGroupDlqEnableSupplier.get())
+ : updatedState.startStateTransition(RecordState.ARCHIVED, DeliveryCountOps.NO_OP, maxDeliveryCount(), EMPTY_MEMBER_ID, shareGroupDlqEnableSupplier.get());
+ if (updateResult == null) {
+ log.error("Unable to transition ARCHIVING → ARCHIVED ...");
+ return;
+ }
+ stateBatch = new PersisterStateBatch(
+ firstOffset, lastOffset,
+ updateResult.state().id(), (short) updateResult.deliveryCount());
+ deliveryCompleteCount.addAndGet(
+ numInFlightRecordsInBatch(firstOffset, lastOffset));
+ } finally {
+ lock.writeLock().unlock();
+ }
+
+ // Step 3: Persist ARCHIVED
+ writeShareGroupState(List.of(stateBatch))
+ .whenComplete((v2, phase2Exception) -> {
+ boolean cacheUpdated = false;
+ lock.writeLock().lock();
+ try {
+ if (phase2Exception != null) {
+ if (!isTimeout) {
+ // For acknowledge/release path, rollback the state transition.
Review Comment:
Should there be an error log message here with exception?
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -3336,6 +3325,83 @@ private RecordState recordStateWithDlq(byte ackType) {
return ACK_TYPE_TO_RECORD_STATE.get(ackType);
}
+ // Visible for testing.
+
+ /**
+ * The DLQ flow comprises 2 phases:
+ * Phase 1: State transitions to ARCHIVING (happens in the normal acknowledge/release/timeout path) and persists ARCHIVING to the persister
+ * Phase 2: Enqueues to DLQ, then transitions ARCHIVING → ARCHIVED and persists ARCHIVED to the persister
+ * This method handles the complete phase 2 flow.
+ */
+ void initiateDLQAndArchive(InFlightState updatedState, long firstOffset,
+ long lastOffset, short deliveryCount, Throwable dlqCause,
+ boolean isTimeout) {
+ // Step 1: Enqueue to DLQ
+ shareGroupDLQ.enqueue(new ShareGroupDLQRecordParameter(
+ groupId, topicIdPartition, firstOffset, lastOffset,
+ Optional.of(deliveryCount), Optional.ofNullable(dlqCause), false
+ )).whenComplete((v1, dlqException) -> {
+ if (dlqException != null) {
+ log.error("Failed to write to DLQ ... Proceeding to ARCHIVED regardless.");
+ }
+
+ // Step 2: Transition ARCHIVING → ARCHIVED
+ // For timeout path, use tryUpdateState (no rollback) to match non-DLQ behavior
Review Comment:
There is a reason why in case of timeout the persister failure is not important when records are marked back as available. The worst you lose is the delivery count bump.
Hence do we need to differentiate in timeout vs non-timeout path here for DLQ batches? What's the impact when persister calls are failed to mark the record as archived in both cases?
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -3336,6 +3325,83 @@ private RecordState recordStateWithDlq(byte ackType) {
return ACK_TYPE_TO_RECORD_STATE.get(ackType);
}
+ // Visible for testing.
+
+ /**
+ * The DLQ flow comprises 2 phases:
+ * Phase 1: State transitions to ARCHIVING (happens in the normal acknowledge/release/timeout path) and persists ARCHIVING to the persister
+ * Phase 2: Enqueues to DLQ, then transitions ARCHIVING → ARCHIVED and persists ARCHIVED to the persister
+ * This method handles the complete phase 2 flow.
+ */
+ void initiateDLQAndArchive(InFlightState updatedState, long firstOffset,
+ long lastOffset, short deliveryCount, Throwable dlqCause,
+ boolean isTimeout) {
+ // Step 1: Enqueue to DLQ
+ shareGroupDLQ.enqueue(new ShareGroupDLQRecordParameter(
+ groupId, topicIdPartition, firstOffset, lastOffset,
+ Optional.of(deliveryCount), Optional.ofNullable(dlqCause), false
+ )).whenComplete((v1, dlqException) -> {
+ if (dlqException != null) {
+ log.error("Failed to write to DLQ ... Proceeding to ARCHIVED regardless.");
+ }
+
+ // Step 2: Transition ARCHIVING → ARCHIVED
+ // For timeout path, use tryUpdateState (no rollback) to match non-DLQ behavior
+ // where ARCHIVED is never reverted on persist failure.
+ // For acknowledge/release path, use startStateTransition (with rollback) so the
+ // client can retry on persist failure.
+ PersisterStateBatch stateBatch;
+ lock.writeLock().lock();
+ try {
+ InFlightState updateResult = isTimeout
+ ? updatedState.tryUpdateState(RecordState.ARCHIVED, DeliveryCountOps.NO_OP, maxDeliveryCount(), EMPTY_MEMBER_ID, shareGroupDlqEnableSupplier.get())
+ : updatedState.startStateTransition(RecordState.ARCHIVED, DeliveryCountOps.NO_OP, maxDeliveryCount(), EMPTY_MEMBER_ID, shareGroupDlqEnableSupplier.get());
+ if (updateResult == null) {
+ log.error("Unable to transition ARCHIVING → ARCHIVED ...");
Review Comment:
```suggestion
log.error("Unable to transition record from ARCHIVING to ARCHIVED. <Add some additional information like group, topic, partition, offset, etc. which can be helpfule to find the issue.>");
```
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20245: DLQ records exceeding max delivery count. [3/N] [kafka]
Posted by "soarez (via GitHub)" <gi...@apache.org>.
soarez commented on PR #22080:
URL: https://github.com/apache/kafka/pull/22080#issuecomment-4389539618
* May 5th ~2pm BST - Build check for this PR finished.
* May 6th ~1am BST - bbd81e8d00 (#22209) merged, ranaming `ShareGroupDLQ` to `ShareGroupDLQManager`
* May 6th ~3pm BST - This PR was merged
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20245: DLQ records exceeding max delivery count. [3/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #22080:
URL: https://github.com/apache/kafka/pull/22080#discussion_r3166039933
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -2567,75 +2584,14 @@ void rollbackOrProcessStateUpdates(
dlqBatches.forEach(persisterBatch -> {
persisterBatch.updatedState.completeStateTransition(true);
- shareGroupDLQ.enqueue(new ShareGroupDLQRecordParameter(
Review Comment:
This is already done.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20245: DLQ records exceeding max delivery count. [3/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #22080:
URL: https://github.com/apache/kafka/pull/22080#discussion_r3165969784
##########
server/src/main/java/org/apache/kafka/server/share/fetch/InFlightState.java:
##########
@@ -161,7 +165,7 @@ public InFlightState tryUpdateState(RecordState newState, DeliveryCountOps ops,
}
if (newState == RecordState.AVAILABLE && ops != DeliveryCountOps.DECREASE && deliveryCount >= maxDeliveryCount) {
- newState = RecordState.ARCHIVED;
+ newState = shareGroupDlqEnableSupplier.get() ? RecordState.ARCHIVING : RecordState.ARCHIVED;
Review Comment:
We persist - but above check is only when we transition to some(state) => ARCHIVING. The logic is not applicable to records which are already in ARCHIVING (persisted). If something is already in ARCHIVING then the flag does not affect.
Additional code to handle ARCHIVING state records received in leadership change is pending - to handle records fetched in `readState` call. We could decide to directly move such records to ARCHIVE without invoking DLQ.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20245: DLQ records exceeding max delivery count. [3/N] [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on PR #22080:
URL: https://github.com/apache/kafka/pull/22080#issuecomment-4350476221
@apoorvmittal10 Thanks for the comments, have incorporated changes.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org