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

Code before:

    private Optional<Throwable> acknowledgeCompleteBatch(
        ShareAcknowledgementBatch batch,
        InFlightBatch inFlightBatch,
        byte ackType,
        List<PersisterBatch> persisterBatches,
        String memberId
    ) {
        lock.writeLock().lock();
        try {
            // The in-flight batch is a full match hence change the state of the complete.
            log.trace("Acknowledging complete batch record {} for the share partition: {}-{}",
                batch, groupId, topicIdPartition);
            if (inFlightBatch.batchState() != RecordState.ACQUIRED) {
                log.debug("The batch is not in the acquired state: {} for share partition: {}-{}",
                    inFlightBatch, groupId, topicIdPartition);
                return Optional.of(new InvalidRecordStateException(
                    "The batch cannot be acknowledged. The batch is not in the acquired state."));
            }

            // If the request is a full-batch RENEW acknowledgement (ack type 4), then renew the
            // acquisition lock without changing the state or persisting anything.
            // Before reaching this point, it should be verified that it is full batch ack and
            // not per offset ack as well as startOffset not moved.
            if (ackType == AcknowledgeType.RENEW.id) {
                if (inFlightBatch.batchState() != RecordState.ACQUIRED) {
                    log.debug("The batch with RENEW ack is not in the acquired state: {} for share partition: {}-{}",
                        inFlightBatch, groupId, topicIdPartition);
                    return Optional.of(new InvalidRecordStateException(
                        "The batch cannot be RENEW acknowledged. The batch is not in the acquired state."));
                }
                // Renew the acquisition lock timer for the complete batch.
                log.debug("Renewing acq lock for {}-{} with offsets {}-{} for member {}.",
                    groupId, topicIdPartition, inFlightBatch.firstOffset(), inFlightBatch.lastOffset(), memberId);
                inFlightBatch.cancelAndClearAcquisitionLockTimeoutTask();
                AcquisitionLockTimerTask renewalTask = scheduleAcquisitionLockTimeout(memberId,
                    inFlightBatch.firstOffset(), inFlightBatch.lastOffset());
                inFlightBatch.updateAcquisitionLockTimeout(renewalTask);
                // Nothing to persist.
                return Optional.empty();
            }

            // Change the state of complete batch since the same state exists for the entire inFlight batch.
            // The member id is reset to EMPTY_MEMBER_ID irrespective of the ack type as the batch is
            // either released or moved to a state where member id existence is not important. The member id
            // is only important when the batch is acquired.
            RecordState recordState = ACK_TYPE_TO_RECORD_STATE.get(ackType);
            Objects.requireNonNull(recordState);

            InFlightState updateResult = inFlightBatch.startBatchStateTransition(
                recordState,

Review comment (performance):
This condition is redundant because of the check on line 2150. We know the the batch state is ACQUIRED if we have reached this point.

Code after:

    private Optional<Throwable> acknowledgeCompleteBatch(
        ShareAcknowledgementBatch batch,
        InFlightBatch inFlightBatch,
        byte ackType,
        List<PersisterBatch> persisterBatches,
        String memberId
    ) {
        lock.writeLock().lock();
        try {
            // The in-flight batch is a full match hence change the state of the complete.
            log.trace("Acknowledging complete batch record {} for the share partition: {}-{}",
                batch, groupId, topicIdPartition);
            if (inFlightBatch.batchState() != RecordState.ACQUIRED) {
                log.debug("The batch is not in the acquired state: {} for share partition: {}-{}",
                    inFlightBatch, groupId, topicIdPartition);
                return Optional.of(new InvalidRecordStateException(
                    "The batch cannot be acknowledged. The batch is not in the acquired state."));
            }

            // If the request is a full-batch RENEW acknowledgement (ack type 4), then renew the
            // acquisition lock without changing the state or persisting anything.
            // Before reaching this point, it should be verified that it is full batch ack and
            // not per offset ack as well as startOffset not moved.
            if (ackType == AcknowledgeType.RENEW.id) {
                // Renew the acquisition lock timer for the complete batch. We have already
                // checked that the batchState is ACQUIRED above.
                log.debug("Renewing acq lock for {}-{} with batch {}-{} for member {}.",
                    groupId, topicIdPartition, inFlightBatch.firstOffset(), inFlightBatch.lastOffset(), memberId);
                inFlightBatch.cancelAndClearAcquisitionLockTimeoutTask();
                AcquisitionLockTimerTask renewalTask = scheduleAcquisitionLockTimeout(memberId,
                    inFlightBatch.firstOffset(), inFlightBatch.lastOffset());
                inFlightBatch.updateAcquisitionLockTimeout(renewalTask);
                // Nothing to persist.
                return Optional.empty();
            }

            // Change the state of complete batch since the same state exists for the entire inFlight batch.
            // The member id is reset to EMPTY_MEMBER_ID irrespective of the ack type as the batch is
            // either released or moved to a state where member id existence is not important. The member id
            // is only important when the batch is acquired.
            RecordState recordState = ACK_TYPE_TO_RECORD_STATE.get(ackType);
            Objects.requireNonNull(recordState);

            InFlightState updateResult = inFlightBatch.startBatchStateTransition(
                recordState,
                DeliveryCountOps.NO_OP,
                this.maxDeliveryCount,
                EMPTY_MEMBER_ID
            );
            if (updateResult == null) {