[jira] [Resolved] (KAFKA-16927) Handle expanding leader endpoints

["José Armando García Sancio (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/KAFKA-16927?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

José Armando García Sancio resolved KAFKA-16927.
------------------------------------------------
    Resolution: Fixed

> Handle expanding leader endpoints
> ---------------------------------
>
>                 Key: KAFKA-16927
>                 URL: https://issues.apache.org/jira/browse/KAFKA-16927
>             Project: Kafka
>          Issue Type: Sub-task
>          Components: kraft
>            Reporter: Alyssa Huang
>            Assignee: José Armando García Sancio
>            Priority: Blocker
>             Fix For: 3.9.0
>
>
> Restarting an inactive controller fails to start if the active leader has more endpoint than the latest voter set. The easiest to reproduce is with the following configuration
> {code:java}
> cat kafka.properties | grep controller.quorum
> controller.quorum.voters=0@controller-0:1234,1@controller-1:1234,2@controller-2:1234{code}
> This is what is executed in the QuorumState loading code:
> {code:java}
>               if (leaderEndpoints.isEmpty()) {
>                   ...
>               } else {
>                   initialState = new FollowerState(
>                       time,
>                       election.epoch(),
>                       election.leaderId(),
>                       leaderEndpoints,
>                       voters.voterIds(),
>                       Optional.empty(),
>                       fetchTimeoutMs,
>                       logContext
>                   );
>               }{code}
> If the leader has two endpoints it will send the following BEGIN_QUORUM_EPOCH request:
> {code:java}
>  "leaderEndpoints":[{"name":"CONTROLLER_PLAINTEXT","host":"controller-0","port":1234},{"name":"CONTROLLER","host":"controller-0","port":4321}]{code}
> And this code doesn't handle that correctly:
> {code:java}
>           } else if (
>                   leaderId.isPresent() &&
>                   (!quorum.hasLeader() || leaderEndpoints.size() > quorum.leaderEndpoints().size())
>           ) {
>               // The request or response indicates the leader of the current epoch
>               // which are currently unknown or the replica has discovered more endpoints
>               transitionToFollower(epoch, leaderId.getAsInt(), leaderEndpoints, currentTimeMs);
>           }{code}
> After adding a test for this, the test fails with the following:
> {code:java}
> Gradle Test Run :raft:test > Gradle Test Executor 44 > KafkaRaftClientTest > testHandleBeginQuorumRequestMoreEndpoints() FAILED
>     java.lang.AssertionError: Assertion failed with an exception
>         at org.apache.kafka.test.TestUtils.retryOnExceptionWithTimeout(TestUtils.java:453)
>         at org.apache.kafka.test.TestUtils.waitForCondition(TestUtils.java:393)
>         at org.apache.kafka.test.TestUtils.waitForCondition(TestUtils.java:377)
>         at org.apache.kafka.test.TestUtils.waitForCondition(TestUtils.java:367)
>         at org.apache.kafka.raft.RaftClientTestContext.pollUntil(RaftClientTestContext.java:617)
>         at org.apache.kafka.raft.RaftClientTestContext.pollUntilResponse(RaftClientTestContext.java:624)
>         at org.apache.kafka.raft.KafkaRaftClientTest.testHandleBeginQuorumRequestMoreEndpoints(KafkaRaftClientTest.java:993)
>         Caused by:
>         java.lang.IllegalStateException: Cannot transition to Follower with leader 829 and epoch 3 from state FollowerState(fetchTimeoutMs=50000, epoch=3, leader=829, leaderEndpoints=Endpoints(endpoints={ListenerName(LISTENER)=localhost/<unresolved>:10819}), voters=[828, 829], highWatermark=Optional.empty, fetchingSnapshot=Optional.empty)
>             at org.apache.kafka.raft.QuorumState.transitionToFollower(QuorumState.java:480)
>             at org.apache.kafka.raft.KafkaRaftClient.transitionToFollower(KafkaRaftClient.java:732)
>             at org.apache.kafka.raft.KafkaRaftClient.maybeTransition(KafkaRaftClient.java:2434)
>             at org.apache.kafka.raft.KafkaRaftClient.handleBeginQuorumEpochRequest(KafkaRaftClient.java:1018)
>             at org.apache.kafka.raft.KafkaRaftClient.handleRequest(KafkaRaftClient.java:2565)
>             at org.apache.kafka.raft.KafkaRaftClient.handleInboundMessage(KafkaRaftClient.java:2613)
>             at org.apache.kafka.raft.KafkaRaftClient.poll(KafkaRaftClient.java:3314)
>             at org.apache.kafka.raft.RaftClientTestContext.lambda$pollUntil$1(RaftClientTestContext.java:618)
>             at org.apache.kafka.test.TestUtils.lambda$waitForCondition$3(TestUtils.java:396)
>             at org.apache.kafka.test.TestUtils.retryOnExceptionWithTimeout(TestUtils.java:444)
>             ... 6 more
> 1 test completed, 1 failed {code}



--
This message was sent by Atlassian Jira
(v8.20.10#820010)