vamossagar12 commented on code in PR #14241:
URL: https://github.com/apache/kafka/pull/14241#discussion_r1349466119
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java:
##########
@@ -459,19 +459,14 @@ boolean joinGroupIfNeeded(final Timer timer) {
}
if (future.succeeded()) {
- Generation generationSnapshot;
- MemberState stateSnapshot;
-
// Generation data maybe concurrently cleared by Heartbeat
thread.
// Can't use synchronized for {@code onJoinComplete}, because
it can be long enough
// and shouldn't block heartbeat thread.
// See {@link
PlaintextConsumerTest#testMaxPollIntervalMsDelayInAssignment}
- synchronized (AbstractCoordinator.this) {
- generationSnapshot = this.generation;
- stateSnapshot = this.state;
- }
+ Generation generationSnapshot;
+ MemberState stateSnapshot = null;
- if (!hasGenerationReset(generationSnapshot) && stateSnapshot
== MemberState.STABLE) {
+ if (!hasGenerationReset(generationSnapshot = this.generation)
&& (stateSnapshot = this.state) == MemberState.STABLE) {
Review Comment:
I think, there is still a possibility of race condition here. `generation`
gets updated by the heartbeat thread in a synchronized block, but there's no
guarantee that what we see here might be the updated value because not we are
not even holding any locks.
--
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]
