jsancio commented on code in PR #22111:
URL: https://github.com/apache/kafka/pull/22111#discussion_r3477056199
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientFetchTest.java:
##########
@@ -765,4 +766,122 @@ void testUpdatedHighWatermarkCompleted() throws Exception
{
assertEquals(localLogEndOffset, partitionResponse.highWatermark());
}
}
+
+ @Test
+ void testObserverFetchesBetweenLeaderAndBootstrapServers() throws
Exception {
+ final var epoch = 2;
+ final var local = KafkaRaftClientTest.replicaKey(
+ KafkaRaftClientTest.randomReplicaId(),
+ true
+ );
+ final var leader = KafkaRaftClientTest.replicaKey(local.id() + 1,
true);
+ final var bootstrapVoter = KafkaRaftClientTest.replicaKey(local.id() +
2, true);
+ final var voters = VoterSet.fromMap(
+ Map.of(
+ leader.id(), VoterSetTest.voterNode(leader),
+ bootstrapVoter.id(), VoterSetTest.voterNode(bootstrapVoter)
+ )
+ );
+
+ final var context = new RaftClientTestContext.Builder(
+ local.id(),
+ local.directoryId().get()
+ )
+ .withStaticVoters(voters)
+ // configure the bootstrap servers to only include the bootstrap
voter
+ // to reliably check the destination of the observer's fetch
requests
+ // alternates between the leader and the bootstrap voter
+ .withBootstrapServers(
+
Optional.of(List.of(RaftClientTestContext.mockAddress(bootstrapVoter.id())))
+ )
+
.withRaftProtocol(RaftClientTestContext.RaftProtocol.KIP_1166_PROTOCOL)
+ .build();
+
+ // The observer initially fetches from the bootstrap servers,
+ // where it will discover the leader's endpoints.
+ final var bootstrapFetch = pollAndCheckObserverFetchRequest(
+ context,
+ true,
+ bootstrapVoter.id()
+ );
+ context.deliverResponse(
+ bootstrapFetch.correlationId(),
+ bootstrapFetch.destination(),
+ context.fetchResponse(
+ epoch,
+ leader.id(),
+ MemoryRecords.EMPTY,
+ 0L,
+ Errors.NOT_LEADER_OR_FOLLOWER
+ )
+ );
+
+ // Subsequent fetch from the observer is sent to the leader
+ // Return a BROKER_NOT_AVAILABLE error, and then advance time past the
fetch timeout,
+ // which should cause the observer to fetch from the bootstrap servers
on the next fetch.
+ final var leaderFetch = pollAndCheckObserverFetchRequest(
+ context,
+ false,
+ leader.id()
+ );
+ context.deliverResponse(
+ leaderFetch.correlationId(),
+ leaderFetch.destination(),
+ RaftUtil.errorResponse(
+ ApiKeys.FETCH,
+ Errors.BROKER_NOT_AVAILABLE
+ )
+ );
Review Comment:
According to the trace I pasted, this response is not deliver before the
fetch timeout. This is misleading when reading the test.
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientFetchTest.java:
##########
@@ -765,4 +766,122 @@ void testUpdatedHighWatermarkCompleted() throws Exception
{
assertEquals(localLogEndOffset, partitionResponse.highWatermark());
}
}
+
+ @Test
+ void testObserverFetchesBetweenLeaderAndBootstrapServers() throws
Exception {
+ final var epoch = 2;
+ final var local = KafkaRaftClientTest.replicaKey(
+ KafkaRaftClientTest.randomReplicaId(),
+ true
+ );
+ final var leader = KafkaRaftClientTest.replicaKey(local.id() + 1,
true);
+ final var bootstrapVoter = KafkaRaftClientTest.replicaKey(local.id() +
2, true);
+ final var voters = VoterSet.fromMap(
+ Map.of(
+ leader.id(), VoterSetTest.voterNode(leader),
+ bootstrapVoter.id(), VoterSetTest.voterNode(bootstrapVoter)
+ )
+ );
+
+ final var context = new RaftClientTestContext.Builder(
+ local.id(),
+ local.directoryId().get()
+ )
+ .withStaticVoters(voters)
Review Comment:
Why use static voters? Why not enable and use all of the latest features?
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientFetchTest.java:
##########
@@ -765,4 +766,122 @@ void testUpdatedHighWatermarkCompleted() throws Exception
{
assertEquals(localLogEndOffset, partitionResponse.highWatermark());
}
}
+
+ @Test
+ void testObserverFetchesBetweenLeaderAndBootstrapServers() throws
Exception {
+ final var epoch = 2;
+ final var local = KafkaRaftClientTest.replicaKey(
+ KafkaRaftClientTest.randomReplicaId(),
+ true
+ );
+ final var leader = KafkaRaftClientTest.replicaKey(local.id() + 1,
true);
+ final var bootstrapVoter = KafkaRaftClientTest.replicaKey(local.id() +
2, true);
+ final var voters = VoterSet.fromMap(
+ Map.of(
+ leader.id(), VoterSetTest.voterNode(leader),
+ bootstrapVoter.id(), VoterSetTest.voterNode(bootstrapVoter)
+ )
+ );
+
+ final var context = new RaftClientTestContext.Builder(
+ local.id(),
+ local.directoryId().get()
+ )
+ .withStaticVoters(voters)
+ // configure the bootstrap servers to only include the bootstrap
voter
+ // to reliably check the destination of the observer's fetch
requests
+ // alternates between the leader and the bootstrap voter
+ .withBootstrapServers(
+
Optional.of(List.of(RaftClientTestContext.mockAddress(bootstrapVoter.id())))
+ )
+
.withRaftProtocol(RaftClientTestContext.RaftProtocol.KIP_1166_PROTOCOL)
+ .build();
+
+ // The observer initially fetches from the bootstrap servers,
+ // where it will discover the leader's endpoints.
+ final var bootstrapFetch = pollAndCheckObserverFetchRequest(
+ context,
+ true,
+ bootstrapVoter.id()
+ );
+ context.deliverResponse(
+ bootstrapFetch.correlationId(),
+ bootstrapFetch.destination(),
+ context.fetchResponse(
+ epoch,
+ leader.id(),
+ MemoryRecords.EMPTY,
+ 0L,
+ Errors.NOT_LEADER_OR_FOLLOWER
+ )
+ );
+
+ // Subsequent fetch from the observer is sent to the leader
+ // Return a BROKER_NOT_AVAILABLE error, and then advance time past the
fetch timeout,
+ // which should cause the observer to fetch from the bootstrap servers
on the next fetch.
+ final var leaderFetch = pollAndCheckObserverFetchRequest(
+ context,
+ false,
+ leader.id()
+ );
+ context.deliverResponse(
+ leaderFetch.correlationId(),
+ leaderFetch.destination(),
+ RaftUtil.errorResponse(
+ ApiKeys.FETCH,
+ Errors.BROKER_NOT_AVAILABLE
+ )
+ );
+
+ // The fetch timeout is much greater than the request manager's
configured backoff, so the
+ // current unreachable connection will no longer be backing off when
the next fetch is sent.
+ // Expire the fetch timeout and check that the next fetch is sent to
the bootstrap server again.
+ context.time.sleep(context.fetchTimeoutMs + 1);
Review Comment:
I made this changes and the test pass. It looks like the issue is that the
leader is in the backoff state because kraft got an error from the leader:
```bash
diff --git
a/raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientFetchTest.java
b/raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientFetchTest.java
index 8d762d6c96..3d873add30 100644
--- a/raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientFetchTest.java
+++ b/raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientFetchTest.java
@@ -836,7 +836,6 @@ public final class KafkaRaftClientFetchTest {
// The fetch timeout is much greater than the request manager's
configured backoff, so the
// current unreachable connection will no longer be backing off
when the next fetch is sent.
// Expire the fetch timeout and check that the next fetch is sent
to the bootstrap server again.
- context.time.sleep(context.fetchTimeoutMs + 1);
final var nextBootstrapFetch = pollAndCheckObserverFetchRequest(
context,
true,
@@ -854,6 +853,8 @@ public final class KafkaRaftClientFetchTest {
)
);
+ context.time.sleep(context.retryBackoffMs);
+
// Discovering the leader from a bootstrap fetch means the observer
resumes fetching from the leader
pollAndCheckObserverFetchRequest(
context,
@@ -871,10 +872,8 @@ public final class KafkaRaftClientFetchTest {
context.pollUntilRequest();
RaftRequest.Outbound fetchRequest =
context.assertSentFetchRequest();
if (isBootstrapFetch) {
- assertTrue(context.client.quorum().isUnattached());
assertTrue(fetchRequest.destination().id() < -1);
} else {
- assertTrue(context.client.quorum().isFollower());
assertEquals(expectedDestinationId,
fetchRequest.destination().id());
}
// only need to check port since the host is always "localhost" for
the mock addresses
```
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientFetchTest.java:
##########
@@ -765,4 +766,122 @@ void testUpdatedHighWatermarkCompleted() throws Exception
{
assertEquals(localLogEndOffset, partitionResponse.highWatermark());
}
}
+
+ @Test
+ void testObserverFetchesBetweenLeaderAndBootstrapServers() throws
Exception {
+ final var epoch = 2;
+ final var local = KafkaRaftClientTest.replicaKey(
+ KafkaRaftClientTest.randomReplicaId(),
+ true
+ );
+ final var leader = KafkaRaftClientTest.replicaKey(local.id() + 1,
true);
+ final var bootstrapVoter = KafkaRaftClientTest.replicaKey(local.id() +
2, true);
+ final var voters = VoterSet.fromMap(
+ Map.of(
+ leader.id(), VoterSetTest.voterNode(leader),
+ bootstrapVoter.id(), VoterSetTest.voterNode(bootstrapVoter)
+ )
+ );
+
+ final var context = new RaftClientTestContext.Builder(
+ local.id(),
+ local.directoryId().get()
+ )
+ .withStaticVoters(voters)
+ // configure the bootstrap servers to only include the bootstrap
voter
+ // to reliably check the destination of the observer's fetch
requests
+ // alternates between the leader and the bootstrap voter
+ .withBootstrapServers(
+
Optional.of(List.of(RaftClientTestContext.mockAddress(bootstrapVoter.id())))
+ )
+
.withRaftProtocol(RaftClientTestContext.RaftProtocol.KIP_1166_PROTOCOL)
+ .build();
+
+ // The observer initially fetches from the bootstrap servers,
+ // where it will discover the leader's endpoints.
+ final var bootstrapFetch = pollAndCheckObserverFetchRequest(
+ context,
+ true,
+ bootstrapVoter.id()
+ );
+ context.deliverResponse(
+ bootstrapFetch.correlationId(),
+ bootstrapFetch.destination(),
+ context.fetchResponse(
+ epoch,
+ leader.id(),
+ MemoryRecords.EMPTY,
+ 0L,
+ Errors.NOT_LEADER_OR_FOLLOWER
+ )
+ );
+
+ // Subsequent fetch from the observer is sent to the leader
+ // Return a BROKER_NOT_AVAILABLE error, and then advance time past the
fetch timeout,
+ // which should cause the observer to fetch from the bootstrap servers
on the next fetch.
+ final var leaderFetch = pollAndCheckObserverFetchRequest(
+ context,
+ false,
+ leader.id()
+ );
+ context.deliverResponse(
+ leaderFetch.correlationId(),
+ leaderFetch.destination(),
+ RaftUtil.errorResponse(
+ ApiKeys.FETCH,
+ Errors.BROKER_NOT_AVAILABLE
+ )
+ );
+
+ // The fetch timeout is much greater than the request manager's
configured backoff, so the
+ // current unreachable connection will no longer be backing off when
the next fetch is sent.
+ // Expire the fetch timeout and check that the next fetch is sent to
the bootstrap server again.
+ context.time.sleep(context.fetchTimeoutMs + 1);
+ final var nextBootstrapFetch = pollAndCheckObserverFetchRequest(
+ context,
+ true,
+ bootstrapVoter.id()
+ );
+ context.deliverResponse(
+ nextBootstrapFetch.correlationId(),
+ nextBootstrapFetch.destination(),
+ context.fetchResponse(
+ epoch,
+ leader.id(),
+ MemoryRecords.EMPTY,
+ 0L,
+ Errors.NOT_LEADER_OR_FOLLOWER
+ )
+ );
+
+ // Discovering the leader from a bootstrap fetch means the observer
resumes fetching from the leader
+ pollAndCheckObserverFetchRequest(
+ context,
+ false,
+ leader.id()
+ );
+ }
+
+ //
+ private RaftRequest.Outbound pollAndCheckObserverFetchRequest(
+ RaftClientTestContext context,
+ boolean isBootstrapFetch,
+ int expectedDestinationId
+ ) throws Exception {
+ context.pollUntilRequest();
+ RaftRequest.Outbound fetchRequest = context.assertSentFetchRequest();
+ if (isBootstrapFetch) {
+ assertTrue(context.client.quorum().isUnattached());
Review Comment:
Let's avoid testing internal kraft state. We should try to test and check
kraft's externalities: RPCs, writes to the log, etc.
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientFetchTest.java:
##########
@@ -765,4 +766,122 @@ void testUpdatedHighWatermarkCompleted() throws Exception
{
assertEquals(localLogEndOffset, partitionResponse.highWatermark());
}
}
+
+ @Test
+ void testObserverFetchesBetweenLeaderAndBootstrapServers() throws
Exception {
Review Comment:
I ran this test against this PR and I got this trace:
```
[2026-06-25 15:02:25,329] INFO Starting request manager with bootstrap
servers: [localhost:10634 (id: -2 rack: null isFenced: false)]
(org.apache.kafka.raft.KafkaRaftClient:331)
[2026-06-25 15:02:25,561] INFO Reading KRaft snapshot and log as part of the
initialization (org.apache.kafka.raft.KafkaRaftClient:509)
[2026-06-25 15:02:25,563] INFO Starting voters are
VoterSet(voters={643=VoterNode(voterKey=ReplicaKey(id=643,
directoryId=<undefined>),
listeners=Endpoints(endpoints={ListenerName(LISTENER)=localhost/<unresolved>:10633}),
supportedKRaftVersion=SupportedVersionRange[min_version:0, max_version:0]),
644=VoterNode(voterKey=ReplicaKey(id=644, directoryId=<undefined>),
listeners=Endpoints(endpoints={ListenerName(LISTENER)=localhost/<unresolved>:10634}),
supportedKRaftVersion=SupportedVersionRange[min_version:0, max_version:0])})
(org.apache.kafka.raft.KafkaRaftClient:511)
[2026-06-25 15:02:25,565] INFO Attempting durable transition to
UnattachedState(epoch=0, leaderId=OptionalInt.empty, votedKey=Optional.empty,
voters=[643, 644], electionTimeoutMs=18985, highWatermark=Optional.empty) from
null (org.apache.kafka.raft.QuorumState:732)
[2026-06-25 15:02:25,568] INFO Completed transition to
UnattachedState(epoch=0, leaderId=OptionalInt.empty, votedKey=Optional.empty,
voters=[643, 644], electionTimeoutMs=18985, highWatermark=Optional.empty) from
null (org.apache.kafka.raft.QuorumState:744)
[2026-06-25 15:02:25,586] TRACE Sent outbound request:
OutboundRequest(correlationId=0,
data=FetchRequestData(clusterId='Xs7d_i8LRIuAcKg9hc0dhw', replicaId=-1,
replicaState=ReplicaState(replicaId=642, replicaEpoch=-1), maxWaitMs=0,
minBytes=0, maxBytes=1048576, isolationLevel=0, sessionId=0, sessionEpoch=-1,
topics=[FetchTopic(topic='metadata', topicId=AAAAAAAAAAAAAAAAAAAAAQ,
partitions=[FetchPartition(partition=0, currentLeaderEpoch=0, fetchOffset=0,
lastFetchedEpoch=0, logStartOffset=-1, partitionMaxBytes=0,
replicaDirectoryId=ezHminGtTAmIQQ3i5JFLUQ, highWatermark=-1)])],
forgottenTopicsData=[], rackId=''), createdTimeMs=1782414145309,
destination=localhost:10634 (id: -2 rack: null isFenced: false))
(org.apache.kafka.raft.KafkaRaftClient:2908)
[2026-06-25 15:02:25,587] INFO Registered the listener
org.apache.kafka.raft.RaftClientTestContext$MockListener@107632469
(org.apache.kafka.raft.KafkaRaftClient:3590)
[2026-06-25 15:02:25,725] TRACE Received inbound message
InboundResponse(correlationId=0, data=FetchResponseData(throttleTimeMs=0,
errorCode=0, sessionId=0, responses=[FetchableTopicResponse(topic='',
topicId=AAAAAAAAAAAAAAAAAAAAAQ, partitions=[PartitionData(partitionIndex=0,
errorCode=6, highWatermark=0, lastStableOffset=-1, logStartOffset=-1,
divergingEpoch=EpochEndOffset(epoch=-1, endOffset=-1),
currentLeader=LeaderIdAndEpoch(leaderId=643, leaderEpoch=2),
snapshotId=SnapshotId(endOffset=-1, epoch=-1), abortedTransactions=[],
preferredReadReplica=-1, records=MemoryRecords(size=0,
buffer=java.nio.HeapByteBuffer[pos=0 lim=0 cap=37]))])],
nodeEndpoints=[NodeEndpoint(nodeId=643, host='localhost', port=10633,
rack=null)]), source=localhost:10634 (id: -2 rack: null isFenced: false))
(org.apache.kafka.raft.KafkaRaftClient:2848)
[2026-06-25 15:02:25,726] INFO Attempting durable transition to
FollowerState(fetchTimeoutMs=50000, epoch=2, leader=643,
leaderEndpoints=Endpoints(endpoints={ListenerName(LISTENER)=localhost/<unresolved>:10633}),
votedKey=Optional.empty, voters=[643, 644], highWatermark=Optional.empty,
fetchingSnapshot=Optional.empty) from UnattachedState(epoch=0,
leaderId=OptionalInt.empty, votedKey=Optional.empty, voters=[643, 644],
electionTimeoutMs=18985, highWatermark=Optional.empty)
(org.apache.kafka.raft.QuorumState:732)
[2026-06-25 15:02:25,727] INFO Completed transition to
FollowerState(fetchTimeoutMs=50000, epoch=2, leader=643,
leaderEndpoints=Endpoints(endpoints={ListenerName(LISTENER)=localhost/<unresolved>:10633}),
votedKey=Optional.empty, voters=[643, 644], highWatermark=Optional.empty,
fetchingSnapshot=Optional.empty) from UnattachedState(epoch=0,
leaderId=OptionalInt.empty, votedKey=Optional.empty, voters=[643, 644],
electionTimeoutMs=18985, highWatermark=Optional.empty)
(org.apache.kafka.raft.QuorumState:744)
[2026-06-25 15:02:25,728] DEBUG Notifying listener
org.apache.kafka.raft.RaftClientTestContext$MockListener@107632469 of leader
change LeaderAndEpoch[leaderId=OptionalInt[643], epoch=2]
(org.apache.kafka.raft.KafkaRaftClient:4121)
[2026-06-25 15:02:25,836] TRACE Sent outbound request:
OutboundRequest(correlationId=1,
data=FetchRequestData(clusterId='Xs7d_i8LRIuAcKg9hc0dhw', replicaId=-1,
replicaState=ReplicaState(replicaId=642, replicaEpoch=-1), maxWaitMs=0,
minBytes=0, maxBytes=1048576, isolationLevel=0, sessionId=0, sessionEpoch=-1,
topics=[FetchTopic(topic='metadata', topicId=AAAAAAAAAAAAAAAAAAAAAQ,
partitions=[FetchPartition(partition=0, currentLeaderEpoch=2, fetchOffset=0,
lastFetchedEpoch=0, logStartOffset=-1, partitionMaxBytes=0,
replicaDirectoryId=ezHminGtTAmIQQ3i5JFLUQ, highWatermark=-1)])],
forgottenTopicsData=[], rackId=''), createdTimeMs=1782414145309,
destination=localhost:10633 (id: 643 rack: null isFenced: false))
(org.apache.kafka.raft.KafkaRaftClient:2908)
[2026-06-25 15:02:25,837] INFO Attempting durable transition to
UnattachedState(epoch=2, leaderId=OptionalInt[643], votedKey=Optional.empty,
voters=[643, 644], electionTimeoutMs=9223372036854775807,
highWatermark=Optional.empty) from FollowerState(fetchTimeoutMs=50000, epoch=2,
leader=643,
leaderEndpoints=Endpoints(endpoints={ListenerName(LISTENER)=localhost/<unresolved>:10633}),
votedKey=Optional.empty, voters=[643, 644], highWatermark=Optional.empty,
fetchingSnapshot=Optional.empty) (org.apache.kafka.raft.QuorumState:732)
[2026-06-25 15:02:25,837] INFO Completed transition to
UnattachedState(epoch=2, leaderId=OptionalInt[643], votedKey=Optional.empty,
voters=[643, 644], electionTimeoutMs=9223372036854775807,
highWatermark=Optional.empty) from FollowerState(fetchTimeoutMs=50000, epoch=2,
leader=643,
leaderEndpoints=Endpoints(endpoints={ListenerName(LISTENER)=localhost/<unresolved>:10633}),
votedKey=Optional.empty, voters=[643, 644], highWatermark=Optional.empty,
fetchingSnapshot=Optional.empty) (org.apache.kafka.raft.QuorumState:744)
[2026-06-25 15:02:25,837] TRACE Received inbound message
InboundResponse(correlationId=1, data=FetchResponseData(throttleTimeMs=0,
errorCode=8, sessionId=0, responses=[], nodeEndpoints=[]),
source=localhost:10633 (id: 643 rack: null isFenced: false))
(org.apache.kafka.raft.KafkaRaftClient:2848)
[2026-06-25 15:02:25,838] DEBUG Ignoring response
InboundResponse(correlationId=1, data=FetchResponseData(throttleTimeMs=0,
errorCode=8, sessionId=0, responses=[], nodeEndpoints=[]),
source=localhost:10633 (id: 643 rack: null isFenced: false)) since it is no
longer needed (org.apache.kafka.raft.KafkaRaftClient:2856)
[2026-06-25 15:02:25,942] TRACE Sent outbound request:
OutboundRequest(correlationId=2,
data=FetchRequestData(clusterId='Xs7d_i8LRIuAcKg9hc0dhw', replicaId=-1,
replicaState=ReplicaState(replicaId=642, replicaEpoch=-1), maxWaitMs=0,
minBytes=0, maxBytes=1048576, isolationLevel=0, sessionId=0, sessionEpoch=-1,
topics=[FetchTopic(topic='metadata', topicId=AAAAAAAAAAAAAAAAAAAAAQ,
partitions=[FetchPartition(partition=0, currentLeaderEpoch=2, fetchOffset=0,
lastFetchedEpoch=0, logStartOffset=-1, partitionMaxBytes=0,
replicaDirectoryId=ezHminGtTAmIQQ3i5JFLUQ, highWatermark=-1)])],
forgottenTopicsData=[], rackId=''), createdTimeMs=1782414195310,
destination=localhost:10634 (id: -2 rack: null isFenced: false))
(org.apache.kafka.raft.KafkaRaftClient:2908)
[2026-06-25 15:02:25,943] TRACE Received inbound message
InboundResponse(correlationId=2, data=FetchResponseData(throttleTimeMs=0,
errorCode=0, sessionId=0, responses=[FetchableTopicResponse(topic='',
topicId=AAAAAAAAAAAAAAAAAAAAAQ, partitions=[PartitionData(partitionIndex=0,
errorCode=6, highWatermark=0, lastStableOffset=-1, logStartOffset=-1,
divergingEpoch=EpochEndOffset(epoch=-1, endOffset=-1),
currentLeader=LeaderIdAndEpoch(leaderId=643, leaderEpoch=2),
snapshotId=SnapshotId(endOffset=-1, epoch=-1), abortedTransactions=[],
preferredReadReplica=-1, records=MemoryRecords(size=0,
buffer=java.nio.HeapByteBuffer[pos=0 lim=0 cap=37]))])],
nodeEndpoints=[NodeEndpoint(nodeId=643, host='localhost', port=10633,
rack=null)]), source=localhost:10634 (id: -2 rack: null isFenced: false))
(org.apache.kafka.raft.KafkaRaftClient:2848)
[2026-06-25 15:02:25,943] INFO Attempting durable transition to
FollowerState(fetchTimeoutMs=50000, epoch=2, leader=643,
leaderEndpoints=Endpoints(endpoints={ListenerName(LISTENER)=localhost/<unresolved>:10633}),
votedKey=Optional.empty, voters=[643, 644], highWatermark=Optional.empty,
fetchingSnapshot=Optional.empty) from UnattachedState(epoch=2,
leaderId=OptionalInt[643], votedKey=Optional.empty, voters=[643, 644],
electionTimeoutMs=9223372036854775807, highWatermark=Optional.empty)
(org.apache.kafka.raft.QuorumState:732)
[2026-06-25 15:02:25,944] INFO Completed transition to
FollowerState(fetchTimeoutMs=50000, epoch=2, leader=643,
leaderEndpoints=Endpoints(endpoints={ListenerName(LISTENER)=localhost/<unresolved>:10633}),
votedKey=Optional.empty, voters=[643, 644], highWatermark=Optional.empty,
fetchingSnapshot=Optional.empty) from UnattachedState(epoch=2,
leaderId=OptionalInt[643], votedKey=Optional.empty, voters=[643, 644],
electionTimeoutMs=9223372036854775807, highWatermark=Optional.empty)
(org.apache.kafka.raft.QuorumState:744)
[2026-06-25 15:02:26,047] TRACE Sent outbound request:
OutboundRequest(correlationId=3,
data=FetchRequestData(clusterId='Xs7d_i8LRIuAcKg9hc0dhw', replicaId=-1,
replicaState=ReplicaState(replicaId=642, replicaEpoch=-1), maxWaitMs=0,
minBytes=0, maxBytes=1048576, isolationLevel=0, sessionId=0, sessionEpoch=-1,
topics=[FetchTopic(topic='metadata', topicId=AAAAAAAAAAAAAAAAAAAAAQ,
partitions=[FetchPartition(partition=0, currentLeaderEpoch=2, fetchOffset=0,
lastFetchedEpoch=0, logStartOffset=-1, partitionMaxBytes=0,
replicaDirectoryId=ezHminGtTAmIQQ3i5JFLUQ, highWatermark=-1)])],
forgottenTopicsData=[], rackId=''), createdTimeMs=1782414195310,
destination=localhost:10633 (id: 643 rack: null isFenced: false))
(org.apache.kafka.raft.KafkaRaftClient:2908)
```
Can we get a TRACE of the actual issue to make sure we are solving the
correct problem? I am having a hard time understanding the actual problem so I
am sure that this change solves that problem.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
