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

Code before:
                    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"));
                }

                // 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())) {
                    deliveryCompleteCount.incrementAndGet();
                }
                // Do not update the nextFetchOffset as the offset has not completed the transition yet.
            }
        } finally {
            lock.writeLock().unlock();
        }
        return Optional.empty();
    }

    private Optional<Throwable> acknowledgeCompleteBatch(
        ShareAcknowledgementBatch batch,
        InFlightBatch inFlightBatch,
        RecordState recordState,
        List<PersisterBatch> persisterBatches
    ) {

Review comment (bug):
nit
```suggestion
                    log.debug("Renewing acquisition lock for {}-{} with offset {} in batch {} for member {}.",
```

Code after:
                    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
                    );

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