KAFKA-10011: Remove task id from lockedTaskDirectories during handleLostAll

[abbccdda]

As stated, we couldn't wait for handleRebalanceComplete in the case of handleLostAll, as we already closed the active task as dirty, and could potentially require its offset in the next thread.runOnce call.

Committer Checklist (excluded from commit message)

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

[abbccdda]

test this please

[vvcephei]

Test this please

[ableegoldman]

Instead of removing the tasks during handleLostAll, can we just clear the lockedTaskDirectories set at the end of releaseLockedUnassignedTaskDirectories? This set is only used to keep track of which task directories we only temporarily locked for the rebalance, so it makes sense that it should be empty outside of a rebalance.

[guozhangwang]

I agree with @ableegoldman here, after the while (taskIdIterator.hasNext() loop we can see if there are still remaining tasks, and then log an WARN similar to the end of handleRevocation before clearing them:

if (!remainingPartitions.isEmpty()) {
            log.warn("The following partitions {} are missing from the task partitions. It could potentially " +
                         "due to race condition of consumer detecting the heartbeat failure, or the tasks " +
                         "have been cleaned up by the handleAssignment callback.", remainingPartitions);
        }

[ableegoldman]

I also think we should reset/clear the set at the beginning of tryToLockAllNonEmptyTaskDirectories, so basically we're always dealing with the current set of actually-locked tasks and don't need to worry about removing them during handleLostAll or handleCorruption/Assignment, etc

[abbccdda]

Will attempt Sophie's suggestion here

[ableegoldman]

Cool this LGTM, thanks for the fix!

[guozhangwang]

I think it is still better to log a warn at the end of releaseLockedUnassignedTaskDirectories if the map is not empty at the end, will add it before merging.

[ableegoldman]

@guozhangwang Why would we warn? You mean if we have any tasks leftover? These are just tasks that are currently assigned

[guozhangwang]

@guozhangwang Why would we warn? You mean if we have any tasks leftover? These are just tasks that are currently assigned

I think we would only have leftover tasks if they are closed due to exception during the rebalance.

[ableegoldman]

Well, releaseLockedUnassignedTaskDirectories does exactly that -- it only releases the unassigned task directories (which we just lock in order to report the offset sums in the subscription info). So I would expect there to almost always be leftover tasks at the end of that method