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

Code before:
                // 1. For the first batch which might have offsets prior to the request base
                // offset i.e. cached batch of 10-14 offsets and request batch of 12-13.
                // 2. Skip the offsets which are below the start offset of the share partition
                if (offsetState.getKey() < batch.firstOffset() || offsetState.getKey() < startOffset) {
                    continue;
                }

                if (offsetState.getKey() > batch.lastOffset()) {
                    // No further offsets to process.
                    break;
                }

                if (offsetState.getValue().state() != RecordState.ACQUIRED) {
                    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 offset cannot be acknowledged. The offset has on-going acknowledgement."));
                }

                // 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(),

Review comment (suggestion):
Again, I would use something more generic like "The record state is invalid. The acknowledgement of delivery could not be completed.". I don't think the distinction in why the state is invalid matters to the consuming application.

Code after:
            for (Map.Entry<Long, InFlightState> offsetState : inFlightBatch.offsetState().entrySet()) {

                // 1. For the first batch which might have offsets prior to the request base
                // offset i.e. cached batch of 10-14 offsets and request batch of 12-13.
                // 2. Skip the offsets which are below the start offset of the share partition
                if (offsetState.getKey() < batch.firstOffset() || offsetState.getKey() < startOffset) {
                    continue;
                }

                if (offsetState.getKey() > batch.lastOffset()) {
                    // No further offsets to process.
                    break;
                }

                if (offsetState.getValue().state() != RecordState.ACQUIRED) {
                    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"));
                }

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