jsancio commented on code in PR #22111:
URL: https://github.com/apache/kafka/pull/22111#discussion_r3422608931
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientFetchTest.java:
##########
@@ -789,52 +788,99 @@ void
testObserverFetchesBetweenLeaderAndBootstrapServers() throws Exception {
local.directoryId().get()
)
.withStaticVoters(voters)
-
.withBootstrapServers(Optional.of(List.of(RaftClientTestContext.mockAddress(otherVoter.id()))))
+ .withBootstrapServers(
+
Optional.of(List.of(RaftClientTestContext.mockAddress(bootstrapVoter.id())))
+ )
.withRaftProtocol(RaftClientTestContext.RaftProtocol.KIP_1166_PROTOCOL)
.build();
- for (int i = 0; i < 10; ++i) {
- // The observer initially fetches from the bootstrap servers,
where it will discover the leader's endpoints.
- context.pollUntilRequest();
- final var bootstrapFetch = context.assertSentFetchRequest();
- assertEquals(-2, bootstrapFetch.destination().id());
-
assertEquals(RaftClientTestContext.mockAddress(otherVoter.id()).getHostName(),
bootstrapFetch.destination().host());
-
assertEquals(RaftClientTestContext.mockAddress(otherVoter.id()).getPort(),
bootstrapFetch.destination().port());
-
- 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
- context.pollUntilRequest();
- final var leaderFetch = context.assertSentFetchRequest();
- assertEquals(leader.id(), leaderFetch.destination().id());
-
assertEquals(RaftClientTestContext.mockAddress(leader.id()).getHostName(),
leaderFetch.destination().host());
-
assertEquals(RaftClientTestContext.mockAddress(leader.id()).getPort(),
leaderFetch.destination().port());
-
- // 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.
-
- // 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.
- context.deliverResponse(
- leaderFetch.correlationId(),
- leaderFetch.destination(),
- RaftUtil.errorResponse(
- ApiKeys.FETCH,
- Errors.BROKER_NOT_AVAILABLE
- )
- );
- context.client.poll();
- context.time.sleep(context.fetchTimeoutMs + 1);
+ // 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());
+ assertEquals(-2, fetchRequest.destination().id());
Review Comment:
This is an implementation detail that I am not sure that we should check in
the protocol. Maybe it is enough to check that id is less than -1 (network
client hack to capture bootstrap node) and check the destination endpoint is
the correct endpoint (leader endpoint vs bootstrap endpoint).
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientFetchTest.java:
##########
@@ -789,52 +788,99 @@ void
testObserverFetchesBetweenLeaderAndBootstrapServers() throws Exception {
local.directoryId().get()
)
.withStaticVoters(voters)
-
.withBootstrapServers(Optional.of(List.of(RaftClientTestContext.mockAddress(otherVoter.id()))))
+ .withBootstrapServers(
+
Optional.of(List.of(RaftClientTestContext.mockAddress(bootstrapVoter.id())))
Review Comment:
Let's document that you are doing this to reliable check fetches to the
leader (known node) vs fetches to the bootstrap server (unknown nodes). Another
way to check this is that "bootstrap nodes" have an unknown id. We represent
this in the network client by giving those nodes an id less than -1. RPCs to
known kafka nodes have an id greater than or equal to 0.
##########
raft/src/main/java/org/apache/kafka/raft/FollowerState.java:
##########
@@ -106,6 +106,11 @@ public long remainingFetchTimeMs(long currentTimeMs) {
return fetchTimer.remainingMs();
}
+ public long remainingUpdateVoterSetTimeMs(long currentTimeMs) {
+ updateVoterSetPeriodTimer.update(currentTimeMs);
+ return updateVoterSetPeriodTimer.remainingMs();
+ }
Review Comment:
See my other comment but ideally we should not have this method if is not
used by the kraft implementation.
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -1725,6 +1725,13 @@ private boolean handleFetchResponse(
leaderEndpoints = Endpoints.empty();
}
+ maybeSwitchObserverFetchToLeader(
+ responseEpoch,
+ responseLeaderId,
+ leaderEndpoints,
+ currentTimeMs
+ );
Review Comment:
Why do you need this special handler for FETCH? How about the other RPCs
that follower send like FETCH_SNAPSHOT and UPDATE_VOTER?
##########
raft/src/testFixtures/java/org/apache/kafka/raft/RaftClientTestContext.java:
##########
@@ -998,8 +998,13 @@ void advanceTimeAndCompleteFetch(
int leaderId,
boolean expireUpdateVoterSetTimer
) throws Exception {
+ final var state = client.quorum().followerStateOrThrow();
Review Comment:
Can we implement this without using internal kraft state? These are protocol
tests and ideally should not know anything about the internal implementation.
In the future we should be able to change the kraft implementation and not have
to update any of the `KafkaRaftClient*Test` tests.
--
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]
