You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2020/11/22 14:36:33 UTC
[GitHub] [kafka] dengziming opened a new pull request #9639: KAFKA-10677; Complete fetches in purgatory immediately after resigning
dengziming opened a new pull request #9639:
URL: https://github.com/apache/kafka/pull/9639
*More detailed description of your change*
If the condition of fetch is satisfied, `BROKER_NOT_AVAILABLE` or `NOT_LEADER_OR_FOLLOWER` is returned when the leader is shutting down. so we just return `BROKER_NOT_AVAILABLE` with a message.
*Summary of testing strategy*
A simple unit test to verify fetches in purgatory is completed after resigning.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9639: KAFKA-10677; Complete fetches in purgatory immediately after resigning
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9639:
URL: https://github.com/apache/kafka/pull/9639#discussion_r538909730
##########
File path: raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientTest.java
##########
@@ -207,6 +208,41 @@ public void testEndQuorumEpochRetriesWhileResigned() throws Exception {
assertEquals(1, retries.size());
}
+ @Test
+ public void testResignWillCompleteFetchPurgatory() throws Exception {
+ int localId = 0;
+ int otherNodeId = 1;
+ Set<Integer> voters = Utils.mkSet(localId, otherNodeId);
+
+ RaftClientTestContext context = new RaftClientTestContext.Builder(localId, voters)
+ .build();
+
+ context.becomeLeader();
+ assertEquals(OptionalInt.of(localId), context.currentLeader());
+
+ // send fetch request when become leader
+ int epoch = context.currentEpoch();
+ context.deliverRequest(context.fetchRequest(epoch, otherNodeId, context.log.endOffset().offset, epoch, 1000));
+ context.client.poll();
+ assertEquals(context.client.numWaitingFetch(), 1);
+
+ // append some record, but the fetch in purgatory will still fail
+ context.log.appendAsLeader(Collections.singleton(new SimpleRecord("raft".getBytes())), epoch);
+
+ // when transition to resign, all request in fetchPurgatory will fail
+ context.client.shutdown(1000);
+ context.client.poll();
+ assertTrue(context.client.quorumState().isResigned());
+ assertEquals(context.client.numWaitingFetch(), 0);
Review comment:
I wonder if we need this. Is it not good enough to verify the response was received?
##########
File path: raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientTest.java
##########
@@ -207,6 +208,41 @@ public void testEndQuorumEpochRetriesWhileResigned() throws Exception {
assertEquals(1, retries.size());
}
+ @Test
+ public void testResignWillCompleteFetchPurgatory() throws Exception {
+ int localId = 0;
+ int otherNodeId = 1;
+ Set<Integer> voters = Utils.mkSet(localId, otherNodeId);
+
+ RaftClientTestContext context = new RaftClientTestContext.Builder(localId, voters)
+ .build();
+
+ context.becomeLeader();
+ assertEquals(OptionalInt.of(localId), context.currentLeader());
+
+ // send fetch request when become leader
+ int epoch = context.currentEpoch();
+ context.deliverRequest(context.fetchRequest(epoch, otherNodeId, context.log.endOffset().offset, epoch, 1000));
+ context.client.poll();
+ assertEquals(context.client.numWaitingFetch(), 1);
+
+ // append some record, but the fetch in purgatory will still fail
+ context.log.appendAsLeader(Collections.singleton(new SimpleRecord("raft".getBytes())), epoch);
+
+ // when transition to resign, all request in fetchPurgatory will fail
+ context.client.shutdown(1000);
+ context.client.poll();
+ assertTrue(context.client.quorumState().isResigned());
Review comment:
Maybe we can add a helper to `RaftClientTestContext` similar to `assertElectedLeader` and get rid of `quorumState()`?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dengziming commented on pull request #9639: KAFKA-10677; Complete fetches in purgatory immediately after resigning
Posted by GitBox <gi...@apache.org>.
dengziming commented on pull request #9639:
URL: https://github.com/apache/kafka/pull/9639#issuecomment-732863217
ping @hachikuji to have a look.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji merged pull request #9639: KAFKA-10677; Complete fetches in purgatory immediately after resigning
Posted by GitBox <gi...@apache.org>.
hachikuji merged pull request #9639:
URL: https://github.com/apache/kafka/pull/9639
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dengziming commented on a change in pull request #9639: KAFKA-10677; Complete fetches in purgatory immediately after resigning
Posted by GitBox <gi...@apache.org>.
dengziming commented on a change in pull request #9639:
URL: https://github.com/apache/kafka/pull/9639#discussion_r538989377
##########
File path: raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientTest.java
##########
@@ -207,6 +208,41 @@ public void testEndQuorumEpochRetriesWhileResigned() throws Exception {
assertEquals(1, retries.size());
}
+ @Test
+ public void testResignWillCompleteFetchPurgatory() throws Exception {
+ int localId = 0;
+ int otherNodeId = 1;
+ Set<Integer> voters = Utils.mkSet(localId, otherNodeId);
+
+ RaftClientTestContext context = new RaftClientTestContext.Builder(localId, voters)
+ .build();
+
+ context.becomeLeader();
+ assertEquals(OptionalInt.of(localId), context.currentLeader());
+
+ // send fetch request when become leader
+ int epoch = context.currentEpoch();
+ context.deliverRequest(context.fetchRequest(epoch, otherNodeId, context.log.endOffset().offset, epoch, 1000));
+ context.client.poll();
+ assertEquals(context.client.numWaitingFetch(), 1);
+
+ // append some record, but the fetch in purgatory will still fail
+ context.log.appendAsLeader(Collections.singleton(new SimpleRecord("raft".getBytes())), epoch);
+
+ // when transition to resign, all request in fetchPurgatory will fail
+ context.client.shutdown(1000);
+ context.client.poll();
+ assertTrue(context.client.quorumState().isResigned());
+ assertEquals(context.client.numWaitingFetch(), 0);
Review comment:
This is just a double-check, it's better to be removed.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dengziming commented on a change in pull request #9639: KAFKA-10677; Complete fetches in purgatory immediately after resigning
Posted by GitBox <gi...@apache.org>.
dengziming commented on a change in pull request #9639:
URL: https://github.com/apache/kafka/pull/9639#discussion_r539000267
##########
File path: raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientTest.java
##########
@@ -207,6 +208,41 @@ public void testEndQuorumEpochRetriesWhileResigned() throws Exception {
assertEquals(1, retries.size());
}
+ @Test
+ public void testResignWillCompleteFetchPurgatory() throws Exception {
+ int localId = 0;
+ int otherNodeId = 1;
+ Set<Integer> voters = Utils.mkSet(localId, otherNodeId);
+
+ RaftClientTestContext context = new RaftClientTestContext.Builder(localId, voters)
+ .build();
+
+ context.becomeLeader();
+ assertEquals(OptionalInt.of(localId), context.currentLeader());
+
+ // send fetch request when become leader
+ int epoch = context.currentEpoch();
+ context.deliverRequest(context.fetchRequest(epoch, otherNodeId, context.log.endOffset().offset, epoch, 1000));
+ context.client.poll();
+ assertEquals(context.client.numWaitingFetch(), 1);
+
+ // append some record, but the fetch in purgatory will still fail
+ context.log.appendAsLeader(Collections.singleton(new SimpleRecord("raft".getBytes())), epoch);
+
+ // when transition to resign, all request in fetchPurgatory will fail
+ context.client.shutdown(1000);
+ context.client.poll();
+ assertTrue(context.client.quorumState().isResigned());
Review comment:
Thank you for your suggestion, I added a method to `RaftClientTestContext`.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org