KAFKA-16312, KAFKA-16185: Local epochs in reconciliation

[lucasbru]

The goal of this PR is to change the following internals of the reconciliation:

• Introduce a "local epoch" to the local target assignment. When a new target is received by the server, we compare it with the current value. If it is the same, no change. Otherwise, we bump the local epoch and store the new target assignment. Then, on the reconciliation, we also store the epoch in the reconciled assignment and keep using target != current to trigger the reconciliation.
• When we are not in a group (we have not received an assignment), we use null to represent the local target assignment instead of an empty list, to avoid confusions with an empty assignment received by the server. Similarly, we use null to represent the current assignment, when we haven't reconciled the assignment yet.
• We also carry the new epoch into the request builder to ensure that we report the owned partitions for the last local epoch.
• To address KAFKA-16312 (call onPartitionsAssigned on empty assignments after joining), we apply the initial assignment returned by the group coordinator (whether empty or not) as a normal reconciliation. This avoids introducing another code path to trigger rebalance listeners - reconciliation is the only way to transition to STABLE. The unneeded parts of reconciliation (autocommit, revocation) will be skipped in the existing. Since a lot of unit tests assumed that not reconciliation behavior is invoked when joining the group with an empty assignment, this required a lot of the changes in the unit tests.

Testing

These changes allow the new consumer to pass the first 10 system tests. We piggy-back a minor change to the HeartbeatManager that are required for those system tests as well: always send rebalanceTimoutMs, subscriptions when (re-)joining.

Committer Checklist (excluded from commit message)

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

[lucasbru]

@lianetm @dajac Could you please have a look?

[dajac]

@lucasbru Thanks for the patch. I left a few initial comments.

[lianetm]

This is short-circuiting the reconciliation if the same assignment is received (same epoch, same partitions). But we also need to consider the case that we get the same partitions assigned but in a different epoch. In that case, we should not carry on with the full reconciliation process (there is truly nothing to reconcile), but we should send an ack to the broker, so I would expect we need a similar short-circuit for if sameAssignmentInDiffEpoch => transitionToAck & return;.

It's mainly thinking about the case:

• member owns t1-1 epoch 3
• receives new assignment [t1-1, t2-1] epoch 4, stuck reconciling, ex. missing t2 metadata
• receives new assignment [t1-1] epoch 5 (ex. broker realized t2 has been deleted)
• member does not need to reconcile t1-1, but should send an ack to the broker with t1-1 that it received on a newer epoch
  Makes sense? Not sure if I may be missing something regarding the expectations

[lucasbru]

I guess we could do it, but do we have to, and is it the best thing we should do? It seems like a corner case to me, and maybe the easiest and cleanest behavior is to just run a full reconciliation, because the assignment changed in the meantime, even if our client never managed to reconcile the intermediate assignment. It seems users are using the ConsumerRebalanceListener to detect rebalances (as do our system tests). So the case you are describing, sounds like a rebalance event to me, so I think we should call the listener.

[lianetm]

well I was expecting we only need to trigger the callbacks if the assignment changed (could be to empty, but something needs to change), and that's not the case if the member ends up with t1-1 again, that it already owns.

By running a full reconciliation when the the resolved assignment is the same as the current but received later, we end up with a client reconciling the exact same assignment it already owns :S It would turn out noisy I expect, accounting for 2 rebalances in cases probably much more common, where a new topic assigned is temporarily not in metadata and then discovered: member owns [t1-1], receives assignment [t1-1, t2-1] with missing metadata for t2 (t2 discovered shortly after). Wouldn't we generate 2 rebalances??? (a 1st one with no changes in assignment, a 2nd one with the added topic once discovered) when truly things only changed once)

[dajac]

So, my understanding is the following:

1. Start with [A] at local epoch 10;
2. Go to [A, B] at local epoch 11;
3. B is not resolvable yet so we end up with [A] at local epoch 11 to reconcile;
4. [A] is effectively the current assignment so we would trigger the callback with [];
5. When B is finally available, we get [A, B] at local epoch 11 and reconcile.

@lianetm I think that you're saying that step 4. is not needed, right? My concern with your suggestion is that it defeats a bit the purpose of the patch. However, I do agree with you that in this particular case, we don't have to reconcile at all. Perhaps, a better way to phrase this would be to no trigger the reconciliation if all the newly partitions are waiting on the metadata and there is no revocation. Would something like this work?

[lucasbru]

Okay. I will short-cut the reconciliation if the resolvable assignment did not change and if the local target epoch either did not change or was bumped once (the latter being your example). In the latter case, I'll still bump the local current epoch.

It's a bit more special casing than I was hoping for, but I see that if these "delayed topic names" are common, I'd be annoying to get two reconciliations every time.

[lianetm]

nit: here we are checking if the new assignment is "the same as" the current (not diff)....even maybe just remove this comment, as the one above seems enough?

[lucasbru]

Done

[lianetm]

This new reconciliation triggered is an example of what I was referring to in the comment above, that seems to complicate the flow with 2 rebalances instead of 1. I wouldn't expect a member rebalancing/reconciling at this point (no assignment change for him yet, t2 not in metadata), and then reconciling/rebalance again once it discovers it.

[lucasbru]

Done

[lianetm]

inherited, but now that we're here: "or if it has changed..."

[lucasbru]

Done

[lianetm]

I cannot find any test for one of the 2 issues we're after with this PR: ensure that we are triggering the callbacks when joining and getting empty assignment. Could we add it? I expect it should be very similar to this one, but just providing a CounterConsumerRebalanceListener and asserting that it called the onPartitionsAssigned

[dajac]

I wonder if we could also add a test for the case described in KAFKA-16185. Is it possible?

[lucasbru]

I added the two tests.

[lianetm]

hey @lucasbru, I took another full pass, left a few comments. Thanks for the changes!

[dajac]

@lucasbru Thanks for the update. I left some comments for consideration.

[dajac]

While we are here, we may be able to do the same for serverAssignor field as it never changes during the runtime.

[lucasbru]

Done

[dajac]

I had a second look at all this logic and I actually wonder if it is better to keep the previous logic for rebalanceTimeoutMs and serverAssignor. My concern is that we should actually send out a full request when reset has been called and with the change it is not the case anymore. The main was the send a full request when joining or on any errors (e.g. request timeout).

[lucasbru]

Summary of slack discussion: We need to resend a full request after a failure, to let the GC know that it needs to send a full response (including repeating the assignment). The reason is that if you had a request timeout for a heartbeat request, that heartbeat could have been the one with a new assignment for you. If you don't send a full request next, the server will assume that you've receive that last one so it won't deliver the new assignment again. So sending a full request is an implicit "force full response".

I reverted that part of the changes (so I brought back the rebalanceMs and serverAssignor in sentFields). We could consider only using reset on the heartbeat state (resetting when transitioning to JOINING may be enough). However, that seems more brittle, it’s quite easy to introduce a code path where we forget to reset the heartbeat state. Having a contract like “always send when epoch==0” for the joining case is easier in my eyes.

[lucasbru]

Note that this was indepndently also implemented in #15401 which I had seen but forgotten about. So I will rebase and epoch == 0 conditions will come from the other PR (in the form of an equivalent state == JOINING condition).

[dajac]

nit: We usually don't prefix getters with get.

[lucasbru]

Done

[dajac]

So, my understanding is the following:

1. Start with [A] at local epoch 10;
2. Go to [A, B] at local epoch 11;
3. B is not resolvable yet so we end up with [A] at local epoch 11 to reconcile;
4. [A] is effectively the current assignment so we would trigger the callback with [];
5. When B is finally available, we get [A, B] at local epoch 11 and reconcile.

@lianetm I think that you're saying that step 4. is not needed, right? My concern with your suggestion is that it defeats a bit the purpose of the patch. However, I do agree with you that in this particular case, we don't have to reconcile at all. Perhaps, a better way to phrase this would be to no trigger the reconciliation if all the newly partitions are waiting on the metadata and there is no revocation. Would something like this work?

[dajac]

nit: Should we add LocalAssignmentImpl too?

[lucasbru]

Done

[dajac]

nit: I wonder if we could use equals here. Would it work?

[lucasbru]

No, that's comparing a list and a set and wouldn't work

[dajac]

I wonder if we could also add a test for the case described in KAFKA-16185. Is it possible?

[dajac]

nit: We could use isNone.

[lucasbru]

Done

[lucasbru]

Thanks for the reviews! I updated the PR @lianetm @dajac

[lucasbru]

Done

[lucasbru]

Done

[lucasbru]

Done

[lucasbru]

Done

[lucasbru]

Done

[lucasbru]

Done

[lucasbru]

Done

[lucasbru]

No, that's comparing a list and a set and wouldn't work

[lucasbru]

I added the two tests.

[lianetm]

just to consider maybe simplifying the log and clarify the situation: isn't the message here simply that we're ignoring the reconciliation because resolved target is equals to the current assignment? I get the point about intermediate assignments, but an intermediate one would have updated the current assignment so it wouldn't be equals to the resolved target (or leave a reconciliation in progress so it wouldn't even make it to this check)

[lucasbru]

I didn't want to talk about the local epoch here, since it's more of implementation detail how to detect intermediate assignments. But then I should log the local epoch I supposed. Updated it accordingly.

[lucasbru]

@dajac and I discussed various options to implement the short-cutting. In the end, the outcome was to only skip calling the callbacks but not skipping the entire reconciliation if the resolved assignment is equal to the current assignment but with a different epoch. We still need to update the current assignment with the resolved assignment and the new epoch in order to trigger the "ack", and transition to sending ack as well.

[lianetm]

Agree in the need to update the assignment. I was exactly pushing for doing only only what's needed (vs doing all that the reconciliation does), so this sounds good to me. Thanks!

[lianetm]

Agree with the renamed param, but could we update it also in the same overloaded method above, just for consistency

[lucasbru]

Done

[lianetm]

I would suggest we add the check that a reconciliation was triggered here, just adding
verifyReconciliationTriggeredAndCompleted(membershipManager, Collections.emptyList()); right after poll. It's part of what this PR is introducing and it completes the pic of what's happening when getting the first (empty) assignment that can be reconciled.

[lucasbru]

Done

[lianetm]

Thanks for the updates @lucasbru, left some minor comments.

[lucasbru]

Thanks @lianetm . Addressed the second round of comments.

[lucasbru]

Done

[lucasbru]

Done

[lucasbru]

I didn't want to talk about the local epoch here, since it's more of implementation detail how to detect intermediate assignments. But then I should log the local epoch I supposed. Updated it accordingly.

[dajac]

@lucasbru Thanks for the update. Overall, it looks pretty good to me. I left a few minor comments.

[dajac]

Do we need membershipManager.memberEpoch() == 0 here too? I suppose that it works because the current assignment is reset in the membership manager but it may be better to add it here for consistency.

[lucasbru]

Done

[dajac]

nit: Should we bring this one on the previous line?

[lucasbru]

Done

[dajac]

nit: There is an extra space before target.

[lucasbru]

Done

[dajac]

nit: Could we use exceptionally?

[lucasbru]

Done - but it requires a return null :/

[dajac]

nit: If there are final, could we remove them and directly use the constants where we need to?

[lucasbru]

Done

[dajac]

nit: We could remove this empty line.

[lucasbru]

Done

[dajac]

I wonder if we could add the example that we discussed offline as a test:

0: [T1, T2] -- T2 unresolved (only T1 is reconciled)
1: [T1, T2, T3] -- T2 unresolved (skipped, since reconciliation in progress)
2: [T1, T2] -- T2 unresolved

What do you think?

[lianetm]

Agree it's not covered. (Happy to take that separately myself, right after this if it helps @lucasbru...knowing you're getting busy very soon)

[lucasbru]

Done (slighly simplified, but same idea)

[dajac]

nit: Should we remove this empty line?

[lucasbru]

Done

[lianetm]

Thanks for the updates @lucasbru, took another look, LGTM considering the latests nits.

Just for the record, I will follow-up with KAFKA-16375 for how ongoing reconciliations are discarded after rejoining.

[lucasbru]

Thanks, all comments addressed

[lucasbru]

Done

[lucasbru]

Done

[lucasbru]

Done

[lucasbru]

Done - but it requires a return null :/

[lucasbru]

Done

[lucasbru]

Done

[lucasbru]

Done (slighly simplified, but same idea)

[lucasbru]

Done

[lianetm]

Just to double check, it is intentional here that you're not calling poll right? (that's why a reconciliation is not triggered here, otherwise I would expect we do trigger a reconciliation for t1-1)...I guess this is what you refer to as "intermediate" assignment (an assignment received and removed before a call to poll)

[lucasbru]

Exactly

[lianetm]

Went over the latest updates, LGTM. Thanks!

[dajac]

LGTM, thanks @lucasbru!