KAFKA-12352: Make sure all rejoin group and reset state has a reason

[guozhangwang]

1. Create a reason string to be used for INFO log entry whenever we request re-join or reset generation state.
2. Some minor cleanups.

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[guozhangwang]

@ableegoldman @vcrfxia for reviews.

[guozhangwang]

We can remove this since it is a bit redundant now as we call for each case if necessary.

[ableegoldman]

Just to clarify, you mean we don't need to rejoin here since we will always raise an error, and always rejoin (if necessary) when checking that error?

Or are you referring to the requestRejoinOnResponseError calls you added to the two last cases in the below if/else?

[guozhangwang]

I meant the latter: we call that inside the conditions already -- for those fatal errors, we do not need to call this anyways since the consumer will throw and crash.

[ableegoldman]

@guozhangwang I think something may have been messed up during a merge/rebase: I no longer see requestRejoinOnResponseError being invoked anywhere

[guozhangwang]

I added that function for sync group handler that handles retriable COORDINATOR_NOT_AVAILABLE / NOT_COORDINATOR and any unexpected error. After the refactoring PR they are not all fall into the joinGroupIfNeeded in

final RuntimeException exception = future.exception();

                resetJoinGroupFuture();

                if (exception instanceof UnknownMemberIdException ||
                    exception instanceof IllegalGenerationException ||
                    exception instanceof RebalanceInProgressException ||
                    exception instanceof MemberIdRequiredException)
                    continue;
                else if (!future.isRetriable())
                    throw exception;

                resetStateAndRejoin(String.format("rebalance failed with retriable error %s", exception));
                timer.sleep(rebalanceConfig.retryBackoffMs);

This is part of the principle I mentioned:

We may reset generationa and request rejoin in two different places: 1) in join/sync-group handler, and 2) in joinGroupIfNeeded, when the future is received. The principle is that these two should not overlap, and 2) is used as a fallback for those common errors from join/sync that we do not handle specifically.

But I forgot to remove this function as part of the second pass; will remove.

[ableegoldman]

Ok cool, thanks. One last question then: after this refactoring, since we no longer call requestRejoinOnResponseError below, should we re-add the requestRejoin() call here? Or add a requestRejoin to the specific cases in the SyncGroup handler, eg

} else if (error == Errors.REBALANCE_IN_PROGRESS) {
    log.info("SyncGroup failed: The group began another rebalance. Need to re-join the group. " +
                      "Sent generation was {}", sentGeneration);
    future.raise(error);
}

[guozhangwang]

I think we do not need to, since it would be called on resetStateAndRejoin(String.format("rebalance failed with retriable error %s", exception)); --- previously we are calling rejoin double times.

[ableegoldman]

Hmm...but resetStateAndRejoin(String.format("rebalance failed with retriable error %s", exception)); is only called in joinGroupIfNeeded which is only called in ensureActiveGroup, which is in turn only invoked in ConsumerCoordinator#poll.
That said, inside SyncGroupResponseHandler#handle we would already have rejoinNeeded = true and only set it to false if the SyncGroup succeeds. So for that reason I guess we don't need the requestRejoin anywhere inside the SyncGroup handler

[guozhangwang]

Minor cleanup, we only need to print the error message but not the stack trace.

[guozhangwang]

Note that I intentionally bumped up the log level from debug to info here since I think this is necessarily a message that users should pay attention to in production, where they mostly use INFO. Open for counter suggestions though.

[ableegoldman]

SGTM. If we find it flooding the logs and not helpful we can reconsider

[ableegoldman]

Build failed with Execution failed for task ':connect:runtime:compileJava', I guess trunk is broken atm?

[ableegoldman]

Just to clarify, you mean we don't need to rejoin here since we will always raise an error, and always rejoin (if necessary) when checking that error?

Or are you referring to the requestRejoinOnResponseError calls you added to the two last cases in the below if/else?

[ableegoldman]

SGTM. If we find it flooding the logs and not helpful we can reconsider

[ableegoldman]

nit: rename to resetStateAndGeneration?

[ableegoldman]

Why do we explicitly rejoin in this case, but not eg REBALANCE_IN_PROGRESS? or UNKNOWN_MEMBER_ID/ILLEGAL_GENERATION ?

[guozhangwang]

You're right, we do not, I've updated this section.

[guozhangwang]

We may reset generationa and request rejoin in two different places: 1) in join/sync-group handler, and 2) in joinGroupIfNeeded, when the future is received. The principle is that these two should not overlap, and 2) is used as a fallback for those common errors from join/sync that we do not handle specifically.

[ableegoldman]

LGTM, thanks for the improvement! Feel free to merge if the build passes

[ableegoldman]

Failed with unrelated connect.integration.RebalanceSourceConnectorsIntegrationTest.testMultipleWorkersRejoining() and kafka.server.ScramServerStartupTest.testAuthentications()