jsancio commented on code in PR #15671:
URL: https://github.com/apache/kafka/pull/15671#discussion_r1595916070
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -370,8 +368,52 @@ private void maybeFireLeaderChange() {
}
}
- @Override
- public void initialize() {
+ public void initialize(
+ Map<Integer, InetSocketAddress> voterAddresses,
+ String listenerName,
+ QuorumStateStore quorumStateStore,
+ Metrics metrics
+ ) {
+ partitionState = new KRaftControlRecordStateMachine(
+ Optional.of(VoterSet.fromInetSocketAddresses(listenerName,
voterAddresses)),
+ log,
+ serde,
+ BufferSupplier.create(),
+ MAX_BATCH_SIZE_BYTES,
+ logContext
+ );
+ // Read the entire log
+ logger.info("Reading KRaft snapshot and log as part of the
initialization");
+ partitionState.updateState();
+
+ requestManager = new RequestManager(
+ partitionState.lastVoterSet().voterIds(),
+ quorumConfig.retryBackoffMs(),
+ quorumConfig.requestTimeoutMs(),
+ random
+ );
+
+ quorum = new QuorumState(
+ nodeId,
+ partitionState.lastVoterSet().voterIds(),
+ quorumConfig.electionTimeoutMs(),
+ quorumConfig.fetchTimeoutMs(),
+ quorumStateStore,
+ time,
+ logContext,
+ random
+ );
+
+ kafkaRaftMetrics = new KafkaRaftMetrics(metrics, "raft", quorum);
+ // All Raft voters are statically configured and known at startup
+ // so there are no unknown voter connections. Report this metric as 0.
+ kafkaRaftMetrics.updateNumUnknownVoterConnections(0);
+
+ VoterSet lastVoterSet = partitionState.lastVoterSet();
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
