jsancio commented on code in PR #15671:
URL: https://github.com/apache/kafka/pull/15671#discussion_r1586825107
##########
core/src/main/scala/kafka/raft/RaftManager.scala:
##########
@@ -181,20 +181,12 @@ class KafkaRaftManager[T](
private val clientDriver = new KafkaRaftClientDriver[T](client,
threadNamePrefix, fatalFaultHandler, logContext)
def startup(): Unit = {
- // Update the voter endpoints (if valid) with what's in RaftConfig
- val voterAddresses: util.Map[Integer, AddressSpec] =
controllerQuorumVotersFuture.get()
- for (voterAddressEntry <- voterAddresses.entrySet.asScala) {
- voterAddressEntry.getValue match {
- case spec: InetAddressSpec =>
- netChannel.updateEndpoint(voterAddressEntry.getKey, spec)
- case _: UnknownAddressSpec =>
- info(s"Skipping channel update for destination ID:
${voterAddressEntry.getKey} " +
- s"because of non-routable endpoint:
${NON_ROUTABLE_ADDRESS.toString}")
- case invalid: AddressSpec =>
- warn(s"Unexpected address spec (type: ${invalid.getClass}) for
channel update for " +
- s"destination ID: ${voterAddressEntry.getKey}")
- }
- }
+ client.initialize(
+ controllerQuorumVotersFuture.get(),
+ config.controllerListenerNames.head,
+ new FileBasedStateStore(new File(dataDir, "quorum-state")),
Review Comment:
Done.
--
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
