smjn commented on code in PR #20815:
URL: https://github.com/apache/kafka/pull/20815#discussion_r2487123211
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -2081,31 +2084,50 @@ private Optional<Throwable>
acknowledgePerOffsetBatchRecords(
new InvalidRecordStateException("Member is not the
owner of offset"));
}
- // Determine the record state for the offset. If the per
offset record state is not provided
- // by the client, then use the batch record state.
- RecordState recordState =
- recordStateMap.size() > 1 ?
recordStateMap.get(offsetState.getKey()) :
- recordStateDefault;
- InFlightState updateResult =
offsetState.getValue().startStateTransition(
- recordState,
- DeliveryCountOps.NO_OP,
- this.maxDeliveryCount,
- EMPTY_MEMBER_ID
- );
- if (updateResult == null) {
- log.debug("Unable to acknowledge records for the offset:
{} in batch: {}"
- + " for the share partition: {}-{}",
offsetState.getKey(),
- inFlightBatch, groupId, topicIdPartition);
- return Optional.of(new InvalidRecordStateException(
- "Unable to acknowledge records for the batch"));
- }
- // Successfully updated the state of the offset and created a
persister state batch for write to persister.
- persisterBatches.add(new PersisterBatch(updateResult, new
PersisterStateBatch(offsetState.getKey(),
- offsetState.getKey(), updateResult.state().id(), (short)
updateResult.deliveryCount())));
- if (isStateTerminal(updateResult.state())) {
- inFlightTerminalRecords.incrementAndGet();
+ // In case of 0 size ackTypeMap, we have already validated the
batch.acknowledgeTypes.
+ byte ackType = ackTypeMap.size() > 1 ?
ackTypeMap.get(offsetState.getKey()) : batch.acknowledgeTypes().get(0);
Review Comment:
@adixitconfluent Yes correct - just avoiding a lookup
--
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]
