private void updateLeaderEndOffsetAndTimestamp(
        LeaderState<T> state,
        long currentTimeMs
    ) {
        final LogOffsetMetadata endOffsetMetadata = log.endOffset();

        if (state.updateLocalState(endOffsetMetadata, partitionState.lastVoterSet())) {
            onUpdateLeaderHighWatermark(state, currentTimeMs);
        }

        fetchPurgatory.maybeComplete(endOffsetMetadata.offset(), currentTimeMs);
    }
Create a new instance.

Note that if the node ID is empty, then the client will behave as a
non-participating observer.
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java