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

Code before:
        } finally {
            lock.writeLock().unlock();
        }

        writeShareGroupState(stateBatches).whenComplete((result, exception) -> {
            // There can be a pending delayed share fetch requests for the share partition which are waiting
            // on the startOffset to move ahead, hence track if the state is updated in the cache. If
            // yes, then notify the delayed share fetch purgatory to complete the pending requests.
            boolean cacheStateUpdated = false;
            lock.writeLock().lock();
            try {
                if (exception != null) {
                    log.debug("Failed to write state to persister for the share partition: {}-{}",
                        groupId, topicIdPartition, exception);
                    updatedStates.forEach(state -> state.completeStateTransition(false));
                    future.completeExceptionally(exception);
                    return;
                }

                log.trace("State change request successful for share partition: {}-{}",
                    groupId, topicIdPartition);
                updatedStates.forEach(state -> {
                    state.completeStateTransition(true);
                    // Cancel the acquisition lock timeout task for the state since it is acknowledged/released successfully.
                    state.cancelAndClearAcquisitionLockTimeoutTask();
                    if (state.state != RecordState.ARCHIVED) {
                        findNextFetchOffset.set(true);
                    }
                });
                // Update the cached state and start and end offsets after acknowledging/releasing the acquired records.
                cacheStateUpdated  = maybeUpdateCachedStateAndOffsets();
                future.complete(null);
            } finally {
                lock.writeLock().unlock();
                // Maybe complete the delayed share fetch request if the state has been changed in cache
                // which might have moved start offset ahead. Hence, the pending delayed share fetch
                // request can be completed. The call should be made outside the lock to avoid deadlock.
                maybeCompleteDelayedShareFetchRequest(cacheStateUpdated);
            }
        });
    }

    private boolean maybeUpdateCachedStateAndOffsets() {
        lock.writeLock().lock();
        try {
            if (!canMoveStartOffset()) {
                return false;
            }

            // This will help to find the next position for the startOffset.
            // The new position of startOffset will be lastOffsetAcknowledged + 1

Review comment (suggestion):
Should it be:
```suggestion
                    if (state.state == AVAILABLE) {
```

Code after:
        } finally {
            lock.writeLock().unlock();
        }

        writeShareGroupState(stateBatches).whenComplete((result, exception) -> {
            // There can be a pending delayed share fetch requests for the share partition which are waiting
            // on the startOffset to move ahead, hence track if the state is updated in the cache. If
            // yes, then notify the delayed share fetch purgatory to complete the pending requests.
            boolean cacheStateUpdated = false;
            lock.writeLock().lock();
            try {
                if (exception != null) {
                    log.debug("Failed to write state to persister for the share partition: {}-{}",
                        groupId, topicIdPartition, exception);
                    updatedStates.forEach(state -> state.completeStateTransition(false));
                    future.completeExceptionally(exception);
                    return;
                }

                log.trace("State change request successful for share partition: {}-{}",
                    groupId, topicIdPartition);
                updatedStates.forEach(state -> {
                    state.completeStateTransition(true);
                    // Cancel the acquisition lock timeout task for the state since it is acknowledged/released successfully.
                    state.cancelAndClearAcquisitionLockTimeoutTask();
                    if (state.state == RecordState.AVAILABLE) {
                        findNextFetchOffset.set(true);
                    }
                });
                // Update the cached state and start and end offsets after acknowledging/releasing the acquired records.
                cacheStateUpdated  = maybeUpdateCachedStateAndOffsets();
                future.complete(null);
            } finally {
                lock.writeLock().unlock();
                // Maybe complete the delayed share fetch request if the state has been changed in cache
                // which might have moved start offset ahead. Hence, the pending delayed share fetch
                // request can be completed. The call should be made outside the lock to avoid deadlock.
                maybeCompleteDelayedShareFetchRequest(cacheStateUpdated);
            }
        });
    }

    private boolean maybeUpdateCachedStateAndOffsets() {
        lock.writeLock().lock();
        try {
            if (!canMoveStartOffset()) {
                return false;
            }

            // This will help to find the next position for the startOffset.
            // The new position of startOffset will be lastOffsetAcknowledged + 1