Repository: apache/kafka
File: core/src/main/java/kafka/server/share/SharePartition.java (Java)

Code before:
                    log.debug("The offset is not acquired, offset: {} batch: {} for the share"
                            + " partition: {}-{}", offsetState.getKey(), inFlightBatch, groupId,
                        topicIdPartition);
                    return Optional.of(new InvalidRecordStateException(
                        "The offset cannot be acknowledged. The offset is not acquired."));
                }

                if (offsetState.getValue().hasOngoingStateTransition()) {
                    log.debug("The offset has on-going transition, offset: {} batch: {} for the share"
                            + " partition: {}-{}", offsetState.getKey(), inFlightBatch, groupId,
                        topicIdPartition);
                    return Optional.of(new InvalidRecordStateException(
                        "The record state is invalid. The acknowledgement of delivery could not be completed."));
                }

                // Check if member id is the owner of the offset.
                if (!offsetState.getValue().memberId().equals(memberId)) {
                    log.debug("Member {} is not the owner of offset: {} in batch: {} for the share"
                            + " partition: {}-{}", memberId, offsetState.getKey(), inFlightBatch,
                        groupId, topicIdPartition);
                    return Optional.of(
                        new InvalidRecordStateException("Member is not the owner of offset"));
                }

                // 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);

                if (ackType == AcknowledgeType.RENEW.id) {
                    // If RENEW, renew the acquisition lock timer for this offset and continue without changing state.
                    // We do not care about recordState map here.
                    // Only valid for ACQUIRED offsets; the check above ensures this.
                    long key = offsetState.getKey();
                    InFlightState state = offsetState.getValue();
                    log.debug("Renewing acq lock for {}-{} with offsets {}-{} for member {}.",
                        groupId, topicIdPartition, key, key, memberId);
                    state.cancelAndClearAcquisitionLockTimeoutTask();
                    AcquisitionLockTimerTask renewalTask = scheduleAcquisitionLockTimeout(memberId, key, key);
                    state.updateAcquisitionLockTimeoutTask(renewalTask);
                } else {
                    // 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. This will always be present as it is a static
                    // mapping between bytes and record state type. All ack types have been added except for RENEW which
                    // has been handled above.
                    RecordState recordState = ACK_TYPE_TO_RECORD_STATE.get(ackType);
                    Objects.requireNonNull(recordState);

                    InFlightState updateResult = offsetState.getValue().startStateTransition(
                        recordState,
                        DeliveryCountOps.NO_OP,
                        this.maxDeliveryCount,
                        EMPTY_MEMBER_ID

Review comment (performance):
just confirming, we could have used `ackTypeMap.get(batch.firstOffset())` instead of `batch.acknowledgeTypes().get(0)` as well, right?

Code after:
                    log.debug("The offset is not acquired, offset: {} batch: {} for the share"
                            + " partition: {}-{}", offsetState.getKey(), inFlightBatch, groupId,
                        topicIdPartition);
                    return Optional.of(new InvalidRecordStateException(
                        "The offset cannot be acknowledged. The offset is not acquired."));
                }

                if (offsetState.getValue().hasOngoingStateTransition()) {
                    log.debug("The offset has on-going transition, offset: {} batch: {} for the share"
                            + " partition: {}-{}", offsetState.getKey(), inFlightBatch, groupId,
                        topicIdPartition);
                    return Optional.of(new InvalidRecordStateException(
                        "The record state is invalid. The acknowledgement of delivery could not be completed."));
                }

                // Check if member id is the owner of the offset.
                if (!offsetState.getValue().memberId().equals(memberId)) {
                    log.debug("Member {} is not the owner of offset: {} in batch: {} for the share"
                            + " partition: {}-{}", memberId, offsetState.getKey(), inFlightBatch,
                        groupId, topicIdPartition);
                    return Optional.of(
                        new InvalidRecordStateException("Member is not the owner of offset"));
                }

                // 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);

                if (ackType == AcknowledgeType.RENEW.id) {
                    // If RENEW, renew the acquisition lock timer for this offset and continue without changing state.
                    // We do not care about recordState map here.
                    // Only valid for ACQUIRED offsets; the check above ensures this.
                    long key = offsetState.getKey();
                    InFlightState state = offsetState.getValue();
                    log.debug("Renewing acq lock for {}-{} with offset {} in batch {} for member {}.",
                        groupId, topicIdPartition, key, inFlightBatch, memberId);
                    state.cancelAndClearAcquisitionLockTimeoutTask();
                    AcquisitionLockTimerTask renewalTask = scheduleAcquisitionLockTimeout(memberId, key, key);
                    state.updateAcquisitionLockTimeoutTask(renewalTask);
                } else {
                    // 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. This will always be present as it is a static
                    // mapping between bytes and record state type. All ack types have been added except for RENEW which
                    // has been handled above.
                    RecordState recordState = ACK_TYPE_TO_RECORD_STATE.get(ackType);
                    Objects.requireNonNull(recordState);

                    InFlightState updateResult = offsetState.getValue().startStateTransition(
                        recordState,
                        DeliveryCountOps.NO_OP,
                        this.maxDeliveryCount,
                        EMPTY_MEMBER_ID