squah-confluent commented on code in PR #23174:
URL: https://github.com/apache/kafka/pull/23174#discussion_r3859061286
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/classic/ClassicGroupMember.java:
##########
@@ -397,16 +398,35 @@ public void setAssignment(byte[] value) {
}
/**
+ * Set the member's join future. If a join future is already pending when
a new,
+ * non-null one is set, the new request supersedes it, so the earlier one
is
+ * completed with REBALANCE_IN_PROGRESS first -- otherwise it would never
resolve
+ * on its own.
+ *
* @param value the updated join future.
*/
public void setAwaitingJoinFuture(CompletableFuture<JoinGroupResponseData>
value) {
+ if (value != null && awaitingJoinFuture != null) {
Review Comment:
> There are two paths where we call setAwaitingJoin/SyncFuture(null):
I missed the rollback path in
`updateStaticMemberThenRebalanceOrCompleteJoin`, my mistake.
```
group.updateMember(
newMember,
request.protocols(),
request.rebalanceTimeoutMs(),
request.sessionTimeoutMs(),
responseFuture
);
...
CompletableFuture<Void> appendFuture = new CompletableFuture<>();
appendFuture.whenComplete((__, t) -> {
if (t != null) {
...
// Failed to persist the member id of the given static member,
revert the update of the static member in the group.
group.updateMember(newMember, oldProtocols, oldRebalanceTimeoutMs,
oldSessionTimeoutMs, null);
...
responseFuture.complete(
new JoinGroupResponseData()
.setMemberId(UNKNOWN_MEMBER_ID)
```
Perhaps we could reorder things and call `newMember.completeJoinFuture()`
before `group.updateMember`? Then we can remove the null check from
`setAwaitingJoinFuture`.
--
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]
