private boolean handleBeginQuorumEpochResponse(
        RaftResponse.Inbound responseMetadata,
        long currentTimeMs
    ) {
        int remoteNodeId = responseMetadata.source().id();
        BeginQuorumEpochResponseData response = (BeginQuorumEpochResponseData) responseMetadata.data();
        Errors topLevelError = Errors.forCode(response.errorCode());
        if (topLevelError != Errors.NONE) {
            return handleTopLevelError(topLevelError, responseMetadata);
        }

        if (!hasValidTopicPartition(response, log.topicPartition())) {
            return false;
        }

        BeginQuorumEpochResponseData.PartitionData partitionResponse =
            response.topics().get(0).partitions().get(0);

        Errors partitionError = Errors.forCode(partitionResponse.errorCode());
        OptionalInt responseLeaderId = optionalLeaderId(partitionResponse.leaderId());
        int responseEpoch = partitionResponse.leaderEpoch();

        final Endpoints leaderEndpoints;
        if (responseLeaderId.isPresent()) {
            if (response.nodeEndpoints().isEmpty()) {
                leaderEndpoints = partitionState.lastVoterSet().listeners(responseLeaderId.getAsInt());
            } else {
                leaderEndpoints = Endpoints.fromBeginQuorumEpochResponse(
                    channel.listenerName(),
                    responseLeaderId.getAsInt(),
                    response.nodeEndpoints()
                );
            }
        } else {
            leaderEndpoints = Endpoints.empty();
        }

        Optional<Boolean> handled = maybeHandleCommonResponse(
            partitionError,
            responseLeaderId,
            responseEpoch,
            leaderEndpoints,
            responseMetadata.source(),
            currentTimeMs
        );
        if (handled.isPresent()) {
            return handled.get();
        } else if (partitionError == Errors.NONE) {
            if (quorum.isLeader()) {
                LeaderState<T> state = quorum.leaderStateOrThrow();
                state.addAcknowledgementFrom(remoteNodeId);
            } else {
                logger.debug("Ignoring BeginQuorumEpoch response {} since " +
                    "this node is not the leader anymore", response);
            }
            return true;
        } else {
            return handleUnexpectedError(partitionError, responseMetadata);
        }
    }
Handle a BeginEpoch request. This API may return the following errors:

- {@link Errors#INCONSISTENT_CLUSTER_ID} if the cluster id is presented in request
     but different from this node
- {@link Errors#BROKER_NOT_AVAILABLE} if this node is currently shutting down
- {@link Errors#FENCED_LEADER_EPOCH} if the epoch is smaller than this node's epoch
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java