Repository: apache/kafka
File: server-common/src/main/java/org/apache/kafka/server/share/persister/PersisterStateManager.java (Java)

Code before:
                    coordinatorNode = new Node(coordinatorData.nodeId(), coordinatorData.host(), coordinatorData.port());
                    // now we want the actual share state RPC call to happen
                    if (this.isBatchable()) {
                        addRequestToNodeMap(coordinatorNode, this);
                    } else {
                        enqueue(this);
                    }
                    break;

                case COORDINATOR_NOT_AVAILABLE: // retriable error codes
                case COORDINATOR_LOAD_IN_PROGRESS:
                case NOT_COORDINATOR:
                case UNKNOWN_TOPIC_OR_PARTITION:
                    log.debug("Received retriable error in find coordinator for {} using key {}: {}", name(), partitionKey(), errMessage);
                    if (!findCoordBackoff.canAttempt()) {
                        log.error("Exhausted max retries to find coordinator for {} using key {} without success.", name(), partitionKey());
                        findCoordinatorErrorResponse(error, new Exception("Exhausted max retries to find coordinator without success."));
                        break;
                    }
                    resetCoordinatorNode();
                    timer.add(new PersisterTimerTask(findCoordBackoff.backOff(), this));
                    break;

                default:
                    log.error("Unable to find coordinator for {} using key {}.", name(), partitionKey());
                    findCoordinatorErrorResponse(error, new Exception(errMessage));
            }
        }

        // Visible for testing
        public Node getCoordinatorNode() {
            return coordinatorNode;
        }

        protected abstract boolean isBatchable();

        /**
         * This method can be called by child class objects to register a callback
         * which will be called when the onComplete cb is called on request completion.
         *
         * @param callback
         */
        protected void setOnCompleteCallback(Consumer<ClientResponse> callback) {
            this.onCompleteCallback = callback;
        }
    }

    public class InitializeStateHandler extends PersisterStateManagerHandler {
        private final int stateEpoch;
        private final long startOffset;
        private final CompletableFuture<InitializeShareGroupStateResponse> result;

Review comment (bug):
Why did you choose to pass the error message for this case but not the other RPCs? I would have expected consistent failing in all cases.

Code after:
                    coordinatorNode = new Node(coordinatorData.nodeId(), coordinatorData.host(), coordinatorData.port());
                    // now we want the actual share state RPC call to happen
                    if (this.isBatchable()) {
                        addRequestToNodeMap(coordinatorNode, this);
                    } else {
                        enqueue(this);
                    }
                    break;

                case COORDINATOR_NOT_AVAILABLE: // retriable error codes
                case COORDINATOR_LOAD_IN_PROGRESS:
                case NOT_COORDINATOR:
                case UNKNOWN_TOPIC_OR_PARTITION:
                    log.debug("Received retriable error in find coordinator for {} using key {}: {}", name(), partitionKey(), errorMessage);
                    if (!findCoordBackoff.canAttempt()) {
                        log.error("Exhausted max retries to find coordinator for {} using key {} without success.", name(), partitionKey());
                        findCoordinatorErrorResponse(error, new Exception("Exhausted max retries to find coordinator without success."));
                        break;
                    }
                    resetCoordinatorNode();
                    timer.add(new PersisterTimerTask(findCoordBackoff.backOff(), this));
                    break;

                default:
                    log.error("Unable to find coordinator for {} using key {}.", name(), partitionKey());
                    findCoordinatorErrorResponse(error, new Exception(errorMessage));
            }
        }

        // Visible for testing
        public Node getCoordinatorNode() {
            return coordinatorNode;
        }

        protected abstract boolean isBatchable();

        /**
         * This method can be called by child class objects to register a callback
         * which will be called when the onComplete cb is called on request completion.
         *
         * @param callback
         */
        protected void setOnCompleteCallback(Consumer<ClientResponse> callback) {
            this.onCompleteCallback = callback;
        }
    }

    public class InitializeStateHandler extends PersisterStateManagerHandler {
        private final int stateEpoch;
        private final long startOffset;
        private final CompletableFuture<InitializeShareGroupStateResponse> result;