lianetm commented on code in PR #18050:
URL: https://github.com/apache/kafka/pull/18050#discussion_r1905888150
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/CoordinatorRequestManagerTest.java:
##########
@@ -194,25 +190,11 @@ public void testBackoffAfterRetriableFailure() {
public void testPropagateAndBackoffAfterFatalError() {
CoordinatorRequestManager coordinatorManager =
setupCoordinatorManager(GROUP_ID);
expectFindCoordinatorRequest(coordinatorManager,
Errors.GROUP_AUTHORIZATION_FAILED);
-
- verify(backgroundEventHandler).add(argThat(backgroundEvent -> {
- if (!(backgroundEvent instanceof ErrorEvent))
- return false;
-
- RuntimeException exception = ((ErrorEvent)
backgroundEvent).error();
-
- if (!(exception instanceof GroupAuthorizationException))
- return false;
-
- GroupAuthorizationException groupAuthException =
(GroupAuthorizationException) exception;
- return groupAuthException.groupId().equals(GROUP_ID);
- }));
-
- time.sleep(RETRY_BACKOFF_MS - 1);
Review Comment:
I think we should keep this line, and the we don't need to change the
assertion below (I would expect the only change in this test is to remove the
verify). Should we rename it to `testBackoffAfterFatalError`?
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/CommitRequestManagerTest.java:
##########
@@ -1481,6 +1482,23 @@ public void testSignalClose() {
OffsetCommitRequestData data = (OffsetCommitRequestData)
res.unsentRequests.get(0).requestBuilder().build().data();
assertEquals("topic", data.topics().get(0).name());
}
+
+ @Test
+ public void testPollWithFatalErrorShouldFailingAllUnsetRequest() {
Review Comment:
typo Unsent
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractHeartbeatRequestManager.java:
##########
@@ -263,6 +264,11 @@ public void resetPollTimer(final long pollMs) {
pollTimer.reset(maxPollIntervalMs);
}
+ private void maybePropagateCoordinatorFatalErrorEvent() {
+ coordinatorRequestManager.getAndClearFatalError()
+ .ifPresent(fatalError -> backgroundEventHandler.add(new
ErrorEvent(fatalError)));
+ }
+
Review Comment:
good question. Actually, if my other
[comment](https://github.com/apache/kafka/pull/18050/files#r1904292495) makes
sense to you all, maybe we don't need to clear the error in the
`AbstractHeartbeatRequestManager`?
We would end up with a simpler alternative where it's the
`coordinatorReqMgr` the sole responsible for maintaining the fatal coordinator
error (sets it when it receives a fatal error, clears it when it receives a
successful response). HBMgr and CommitMgr only read the error to take actions
specific the them, never clear it. Would that work?
--
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
