[fix][broker] Prevent stale service unit callbacks from dropping active lookup and cleanup jobs

[void-ptr974]

Motivation

ServiceUnitStateChannelImpl stores pending owner lookups and inactive-broker cleanup jobs in maps keyed by service unit or broker. These futures can be replaced when a lookup is retried, when an Owned event completes a waiting lookup, or when cleanup is rescheduled.

Some completion and cancellation paths removed entries by key only. A stale callback could remove or cancel a newer future for the same key, causing owner lookups to wait for timeout or delaying inactive-broker ownership cleanup.

Modifications

• Guard getOwnerRequests removals with the captured request future.
• Guard cleanupJobs removals with the captured cleanup future.
• Guard broker-creation cleanup cancellation with the cleanup future observed before the async health check.
• Add regression tests for stale owner-lookup, skipped Owned event, cleanup-job completion, and broker-creation cancellation callbacks.

Verifying this change

• ./gradlew :pulsar-broker:test --tests org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateChannelTest.testCompletedGetOwnerRequestDoesNotRemoveNewRequest --tests org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateChannelTest.testSkippedEventDoesNotRemoveNewGetOwnerRequest --tests org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateChannelTest.testCompletedCleanupJobDoesNotRemoveNewCleanupJob --tests org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateChannelTest.handleBrokerCreationEventDoesNotCancelNewCleanupJobTest
• ./gradlew :pulsar-broker:checkstyleMain :pulsar-broker:checkstyleTest
• git diff --check

[heesung-sohn]

The intention was to complete whatever earliest finished lookup asap and unblock all pending lookup requests.

The problem statement seems like the dedup logic didnt happen correctly in some cases.

These futures can be replaced when a lookup is retried, when an Owned event completes a waiting lookup, or when cleanup is rescheduled.

Can you explain the order here? How can the first lookup future be replaced? Shouldn't the later lookup be deduped?

Do we have any case that the lookup dedupe map does not get cleaned up because the completed future is not the one in the map? In this case, this PR will keep the original lookup request forever, which will be a bigger issue.

[void-ptr974]

Thanks for the question. While F1 remains in getOwnerRequests, later lookups are still deduplicated to F1. A replacement becomes possible only after a state-event handler has intentionally removed F1 before completing it. A concurrent lookup or completion continuation may then install F2 before F1's cleanup callback runs:

T1: map[serviceUnit] = F1
T2: event handler removes and completes F1
T3: a later lookup installs F2
T4: F1 cleanup runs
T5: F2 completes and removes itself

With the previous unconditional remove(serviceUnit), the cleanup at T4 could incorrectly remove F2. Using remove(serviceUnit, F1) only removes the future owned by that callback.

This cannot retain F1 indefinitely: if the conditional removal fails, the entry is either already absent or contains F2. A production F2 created through dedupeGetOwnerRequest registers its own completion cleanup.

I agree that the current test's direct put does not make this lifecycle clear. I will update it to create F2 through the dedupe path, verify that completing F1 preserves F2, and then verify that completing F2 removes the map entry.

[heesung-sohn]

T1: map[serviceUnit] = F1
T2: event handler removes and completes F1
T3: a later lookup installs F2
T4: F1 cleanup runs
T5: F2 completes and removes itself

At T4, what's the issue when it can return the ownership early? Isnt it better it doesn't need to wait til T5?

Like I mentioned, the intention was to "early return any ownership". If F1 contains valid ownership, without exception, it can save time til T5?

Also,
at T2, why does the event handler remove F1 from the map? Shouldn't F1 cleanup remove it from the map?

[void-ptr974]

Thanks for the questions. I think the key distinction is between the lifecycle you described and the lifecycle in the original implementation.

My understanding of the intended lifecycle is:

1. F1 remains in getOwnerRequests while the event handler completes it.
2. A lookup arriving before F1 cleanup reuses F1 and can return its valid owner early.
3. F1 cleanup removes the entry, and a later request can then install F2.

Under that lifecycle, F1 still owns the map entry when its cleanup runs, so remove(serviceUnit) would be safe.

However, the original event handlers use a different lifecycle. handleOwnEvent, handleFreeEvent, handleInitEvent, and handleDeleteEvent first remove F1 from the map and then complete it. Once the handler has removed F1, another request or the next assignment can install F2 before the cleanup registered by F1 runs:

Time	Original implementation	getOwnerRequests
T1	A request is waiting for the current state transition	F1
T2	The event handler removes F1 and completes it	empty
T3	Another request or the next assignment installs F2	F2
T4	The cleanup registered by F1 runs	should remain F2
T5	The corresponding state event completes F2	empty

With the previous key-only removal, T4 changes the map from F2 to empty. It does not complete F2 or transfer the result held by F1 to F2. The handler at T5 therefore cannot find F2, and a deferred F2 reaches only its timeout fallback.

If F1 contains a valid owner, callers already holding F1 return that owner when F1 completes at T2; the conditional removal does not make them wait until T5. If F2 can independently resolve an owner from the current Owned state, it can also complete earlier on its own. The change only prevents cleanup belonging to F1 from removing an entry that now belongs to F2.

The remove-before-complete behavior is also important for no-owner results. For example, handleFreeEvent completes F1 with null. If F1 remained in the map, a completion continuation entering the next selection/assignment path could call publishAssignEventAsync and deduplicate against the already-completed null F1 instead of installing F2 to wait for the new Owned event.

Therefore, the gap is that the original cleanup assumes the map entry still belongs to F1, while the original event-handler behavior does not guarantee that assumption. remove(serviceUnit, F1) keeps the existing handler lifecycle and makes the cleanup remove only the future that registered it.

[heesung-sohn]

Thank you for sharing the racing case.

The shared scenario removed F1 twice at T2 and T4. I think the removal should only happen once.

I agree with this safe-guarded removal direction.