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

Code before:
                PartitionErrorData partitionData = state.partitions().get(0);
                if (partitionData.errorCode() != Errors.NONE.code()) {
                    KafkaException ex = fetchPersisterError(partitionData.errorCode(), partitionData.errorMessage());
                    maybeLogError(String.format("Failed to write the share group state for share partition: %s-%s due to exception",
                        groupId, topicIdPartition), Errors.forCode(partitionData.errorCode()), ex);
                    future.completeExceptionally(ex);
                    return;
                }
                future.complete(null);
            });
        return future;
    }

    private KafkaException fetchPersisterError(short errorCode, String errorMessage) {
        Errors error = Errors.forCode(errorCode);
        return switch (error) {
            case NOT_COORDINATOR, COORDINATOR_NOT_AVAILABLE, COORDINATOR_LOAD_IN_PROGRESS ->
                new CoordinatorNotAvailableException(errorMessage);
            case GROUP_ID_NOT_FOUND ->
                new GroupIdNotFoundException(errorMessage);
            case UNKNOWN_TOPIC_OR_PARTITION ->
                new UnknownTopicOrPartitionException(errorMessage);
            case FENCED_LEADER_EPOCH, FENCED_STATE_EPOCH ->
                new NotLeaderOrFollowerException(errorMessage);
            case SASL_AUTHENTICATION_FAILED, UNSUPPORTED_VERSION ->
                new UnknownServerException("Unable to complete operation due to internal error.");
            default ->
                new UnknownServerException(errorMessage);
        };
    }

    // Visible for testing
    AcquisitionLockTimerTask scheduleAcquisitionLockTimeout(String memberId, long firstOffset, long lastOffset) {
        // The recordLockDuration value would depend on whether the dynamic config SHARE_RECORD_LOCK_DURATION_MS in
        // GroupConfig.java is set or not. If dynamic config is set, then that is used, otherwise the value of
        // SHARE_GROUP_RECORD_LOCK_DURATION_MS_CONFIG defined in ShareGroupConfig is used
        int recordLockDurationMs = recordLockDurationMsOrDefault(groupConfigManager, groupId, defaultRecordLockDurationMs);
        return scheduleAcquisitionLockTimeout(memberId, firstOffset, lastOffset, recordLockDurationMs);
    }

    /**
     * Apply acquisition lock to acquired records.
     *
     * @param memberId The member id of the client that is putting the acquisition lock.
     * @param firstOffset The first offset of the acquired records.
     * @param lastOffset The last offset of the acquired records.
     * @param delayMs The delay in milliseconds after which the acquisition lock will be released.
     */
    private AcquisitionLockTimerTask scheduleAcquisitionLockTimeout(
        String memberId,
        long firstOffset,

Review comment (suggestion):
```suggestion
                new UnknownServerException("Unable to complete operation due to server error.");
```

Code after:
                PartitionErrorData partitionData = state.partitions().get(0);
                if (partitionData.errorCode() != Errors.NONE.code()) {
                    KafkaException ex = fetchPersisterError(partitionData.errorCode(), partitionData.errorMessage());
                    maybeLogError(String.format("Failed to write the share group state for share partition: %s-%s due to exception",
                        groupId, topicIdPartition), Errors.forCode(partitionData.errorCode()), ex);
                    future.completeExceptionally(ex);
                    return;
                }
                future.complete(null);
            });
        return future;
    }

    private KafkaException fetchPersisterError(short errorCode, String errorMessage) {
        Errors error = Errors.forCode(errorCode);
        return switch (error) {
            case NOT_COORDINATOR, COORDINATOR_NOT_AVAILABLE, COORDINATOR_LOAD_IN_PROGRESS ->
                new CoordinatorNotAvailableException(errorMessage);
            case GROUP_ID_NOT_FOUND ->
                new GroupIdNotFoundException(errorMessage);
            case UNKNOWN_TOPIC_OR_PARTITION ->
                new UnknownTopicOrPartitionException(errorMessage);
            case FENCED_LEADER_EPOCH, FENCED_STATE_EPOCH ->
                new NotLeaderOrFollowerException(errorMessage);
            case SASL_AUTHENTICATION_FAILED, UNSUPPORTED_VERSION ->
                new UnknownServerException("Unable to complete operation due to server error.");
            default ->
                new UnknownServerException(errorMessage);
        };
    }

    // Visible for testing
    AcquisitionLockTimerTask scheduleAcquisitionLockTimeout(String memberId, long firstOffset, long lastOffset) {
        // The recordLockDuration value would depend on whether the dynamic config SHARE_RECORD_LOCK_DURATION_MS in
        // GroupConfig.java is set or not. If dynamic config is set, then that is used, otherwise the value of
        // SHARE_GROUP_RECORD_LOCK_DURATION_MS_CONFIG defined in ShareGroupConfig is used
        int recordLockDurationMs = recordLockDurationMsOrDefault(groupConfigManager, groupId, defaultRecordLockDurationMs);
        return scheduleAcquisitionLockTimeout(memberId, firstOffset, lastOffset, recordLockDurationMs);
    }

    /**
     * Apply acquisition lock to acquired records.
     *
     * @param memberId The member id of the client that is putting the acquisition lock.
     * @param firstOffset The first offset of the acquired records.
     * @param lastOffset The last offset of the acquired records.
     * @param delayMs The delay in milliseconds after which the acquisition lock will be released.
     */
    private AcquisitionLockTimerTask scheduleAcquisitionLockTimeout(
        String memberId,
        long firstOffset,