Prevent coordinator from getting stuck if leadership changes during coordinator run

[kfaraz]

Description

If the current leader coordinator is asked to stop being leader, the following happens:

• The DruidCoordinator.balancerExec (used for strategy cost computations) is shutdown
• The currently running duty finishes execution normally and no more duties are executed
• An exception to this is the BalanceSegments duty, which can exit abnormally or even get stuck in the race conditions explained below.

✅ Case 1: balancerExec.submit() after balancerExec.shutdown(), BalanceSegments exits abnormally

Typical sequence of events:

• Current coordinator stops being leader and balancerExec is shutdown
• CostBalancerStrategy.findNewSegmentHomeBalancer() or any other strategy method is invoked
• balancerExec.submit() is invoked with computeCost() tasks
• Since the executor has already been shutdown, submission of new tasks throws a RejectedExecutionException and ends the coordinator run as desired

❌ Case 2: balancerExec.submit() before balancerExec.shutdown(), BalanceSegments gets stuck

Typical sequence of events:

• BalanceSegments duty is in progress
• CostBalancerStrategy.findNewSegmentHomeBalancer() is invoked for some segment
• computeCost() tasks for, say 5 servers, are submitted to the executor
• Executor picks up 3 of these tasks and starts executing them
• Coordinator stops being leader and balancerExec is shutdown
• Since the computeCost() tasks do not handle interrupts, the 3 picked up tasks finish execution normally
• But the 2 remaining tasks are never picked up by the executor as it is already shutdown
• The method findNewSegmentHomeBalancer waits indefinitely for the futures to finish

✅ Case 3: Change in balancerComputeThreads dynamic config

A change in this config also results in a shutdown of the balancerExec. But this shutdown is never done concurrently with the coordinator duties and thus doesn't cause the coordinator to get stuck.

Changes

• Add a timeout of 1 minute to the resultFuture.get(). 1 minute is the typical time for a full coordinator run and is more than enough time for cost computations of a single segment.
• Raise an alert if an exception is encountered while computing costs and if the executor has not been shutdown. This is because a shutdown is intentional and does not require an alert.