@guozhangwang this is a draft implementation of one of the solutions you suggested. I will try to add a test, but in the meantime any comments you have are welcome.

@tombentley Thanks for the patch. I think the problem is in the callback inside KafkaApis.handleLeaderAndIsrRequest. With two concurrent LeaderAndIsr requests, there is no guarantee about the ordering of the callbacks even though ReplicaManager checks epoch. They can get reordered like the following:

1. Call ReplicaManager.becomeLeaderOrFollower with epoch=1 in thread 1
2. Call ReplicaManager.becomeLeaderOrFollower with epoch=2 in thread 2
3. Call onLeadershipChange callback in thread 2
4. Call onLeadershipChange callback in thread 1

At least that's my understanding of the issue. I think we need to find a way to push the epoch validation into GroupCoordinator.

@hachikuji, yes I see that although replica manager might be synchronized it doesn't prevent the race. I was already working on introducing the epoch into the group coordinator (as the transaction coordinator already does). I'll rework the patch, thanks for the feedback.

@tombentley Actually I missed the holding of the replicaStateChangeLock inside ReplicaManager.becomeLeaderOrFollower. So the reordering I suggested above seems not possible. Perhaps the problem really is submission to the scheduler. I think we had given the scheduler only a single thread assuming that was good enough to protect the submission order, but maybe it is not.

I assumed the scheduler would be using a heap to prioritize scheduled tasks, and that wouldn't preserve the order of submitted tasks with the same scheduled time. Since System.currentTimeMillis() doesn't tick regularly enough I can see you could have the two threads end up with the same scheduled time of execution (now), but execute in the wrong order. Obviously the FIFO queue approach would solve that, at the cost of an extra thread.

@hachikuji I'm now passing in the coordinator epoch and ignoring attempts to unload state if the epoch is old than what's loaded. I've also added a test. One thing I didn't understand was why I needed to change AbstractFetcherManagerTest. Please could you take another look?

One thing I didn't understand was why I needed to change AbstractFetcherManagerTest.

Stupid me, that was because I turned up the logging. I'll revert that.

@guozhangwang, @hachikuji I was thinking... solving it as I have now (keeping track of the epoch) doesn't address another potential problem case where the tenure as leader is very short and the background task to load the state runs after the background task to unload the state in the same epoch. Obviously in this case the load should not be done (I guess it could result in a broker not returning NOT_COORDINATOR when it should, based on incorrect state of ownedPartitions). We could track a high watermark coordinator epoch and guard the load with a check (as well as the unload).

Hey @tombentley Sorry for the late reply! I also checked the source code of ScheduledThreadPoolExecutor and I think what you've inferred is correct: although it did not keep a single thread alive all the time but dynamically creates and stops the thread in case there's no tasks scheduled for a long time (note we set num.thread == 1), the blocking queue should still guarantee a FIFO ordering.

So what's possibly happening is that, the thread handling an leaderISR request with the old controller epoch grabs the lock and proceeds first. In that case, maybe we can simplify your proposed solution as the following:

• note that, like @hachikuji mentioned, onLeadershipChange is actually protected by the replicaStateChangeLock inside the becomeLeaderOrFollower, and hence we only need to pass in the controller epoch in groupCoordinator.onElection/onResignation just like what we did in txnCoordinator.

• Inside the groupCoordinator.onElection/onResignation, which is lock-protected, we just remember the latest controller epoch at the GroupCoordinator, and then we just to one check right before scheduleLoadGroupAndOffsets/removeGroupsForPartition against that controller epoch, if not passed, we can log and skip the loading / unloading function call.

WDYT?

You're right @guozhangwang, ScheduledFutureTask contains a sequence number to break ties, so the executor is FIFO. So you're saying that the wrong network thread handling one of these two LeaderAndISR acquires replicaStateChangeLock first, right? If that's the case then might a similar problem affect the txnCoordinator too?

I updated the PR if you want to take another look, but I still need to address synchronization, since although onLeadershipChange is executed with ReplicaManager.replicaStateChangeLock, onResignation is also called via handleStopReplicaRequest. Perhaps it's enough to make onResignation synchronized.

So you're saying that the wrong network thread handling one of these two LeaderAndISR acquires replicaStateChangeLock first, right?

Sort of :) I think it is possible that, two ISR request reaches the same broker from old and new controller, and they are handled by two different handler threads of the broker. The handler thread handling the old controller, may grab the lock first.

since although onLeadershipChange is executed with ReplicaManager.replicaStateChangeLock, onResignation is also called via handleStopReplicaRequest

How about this: since handleStopReplicaRequest would call replicaManager.stopReplicas which is also protected by the lock, we can just make the following block

if (error == Errors.NONE) {
          if (topicPartition.topic == GROUP_METADATA_TOPIC_NAME
              && partitionStates(topicPartition).deletePartition) {
            groupCoordinator.onResignation(topicPartition.partition)
          } else if (topicPartition.topic == TRANSACTION_STATE_TOPIC_NAME
                     && partitionStates(topicPartition).deletePartition) {
            val partitionState = partitionStates(topicPartition)
            val leaderEpoch = if (partitionState.leaderEpoch >= 0)
                Some(partitionState.leaderEpoch)
            else
              None
            txnCoordinator.onResignation(topicPartition.partition, coordinatorEpoch = leaderEpoch)
          }

which is currently triggered after the call, as passed in as a delta function parameter into that call, just like what becomeLeaderOrFollower did. Then as long as we can pass in the coordinatorEpoch = leaderEpoch to the groupCoordinator#onResignation that should be sufficient.

I looked at your current PR, and I think that looks promising. If you agree my above suggestion and made the changes accordingly, I think we can turn this draft as an official PR and merge.

cc @cadonna @hachikuji

@guozhangwang that's a much better solution, thanks! I've implemented that and rebased for a conflict.

Rebased for conflict.

@guozhangwang @cadonna @hachikuji please could one of you take a look?

Yes! I will review it again. Thanks for hanging on there and my apologies... Review has always been a bit overwhelming for me :)

Thanks, and no worries about the wait.

This change looks good to me. @hachikuji could you take another look before we merge?

Rebased for conflict. Grateful if you could take a look @hachikuji.

Hey @guozhangwang and @hachikuji , I just ran across this PR as part of the 2.8 release. It's not going to make 2.8, but we should go ahead and close the loop to get it fixed for 3.0.

Thanks!

@tombentley could you see if @hachikuji 's comments can be addressed? This is a pretty tricky bug that I would like to get fixed in 3.0. Thanks!

@hachikuji I've addressed your comments, if you want to take another look?

Thanks @hachikuji, it's much simpler now, if you could take another look?

@hachikuji Thanks, I was being slow on the uptake, but what you describe makes perfect sense. I've moved the logic to the GroupMetadataManager, am using the max observed epoch and have added some more tests. Very grateful if you could take another look.

One thing I found awkward was using ConcurrentHashMap#compute with the possibility of returning null (in the case of stop replicas it makes sense to remove the mapping, right?). I had to switch the java.lang.Integer type args, but maybe you know a better way of handling that?

@tombentley Thanks for the updates. I made a few small changes in this patch: hachikuji@27ba937. The main things are taking the loading itself out of compute (which is intended to be a cheap operation) and moving the loadingPartitions check. Feel free to pull in the changes if they make sense to you.

@hachikuji that's much better, thank you. I added some logging where we ignore on the remove path, hope that's OK.

@hachikuji I fixed the tests and made you suggested changes, but what do you think about @guozhangwang's point above?

@tombentley Apologies for the delay. I was out on leave for the past month. I responded to Guozhang's comment. Let me know if it makes sense or not.

Made another pass on the patch. LGTM! I think we can merge after resolved the conflicts.

Thanks for the reviews @hachikuji @guozhangwang.

Thank YOU for the great patience @tombentley (it lasts for more than 6 months..) !

Cherry-picked to 2.8