DL1231 commented on code in PR #19430:
URL: https://github.com/apache/kafka/pull/19430#discussion_r2055365066
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -3003,24 +3014,32 @@ private boolean hasOngoingStateTransition() {
* if the state update is unsuccessful.
*
* @param newState The new state of the records.
- * @param incrementDeliveryCount Whether to increment the delivery
count.
+ * @param ops The behavior on the delivery count.
*
* @return {@code InFlightState} if update succeeds, null otherwise.
Returning state
* helps update chaining.
*/
- private InFlightState tryUpdateState(RecordState newState, boolean
incrementDeliveryCount, int maxDeliveryCount, String newMemberId) {
+ private InFlightState tryUpdateState(RecordState newState,
DeliveryCountOps ops, int maxDeliveryCount, String newMemberId) {
+ boolean decreaseFlag = false;
try {
+ if (ops == DeliveryCountOps.DECREASE && newState !=
RecordState.ARCHIVED) {
+ deliveryCount--;
Review Comment:
Consider a situation where deliveryCount=4, deliveryCountLimit=5, and the
message is delivered and the deliveryCount becomes 5.
The client is then shut down and needs to release the message,
`ops=decrease`. If the operation is performed after `validateTransition`, it
will first go to
[here](https://github.com/apache/kafka/blob/71d08780d11b23ec4e931efaa8ca329c03f161e3/core/src/main/java/kafka/server/share/SharePartition.java#L3014),
and because deliveryCount=5 at this point, newState is set to ARCHIVED.
In order to avoid this, the decrease of the deliveryCount is performed
before `validateTransition`.
--
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]
