Parallelize supervisor stop logic to make it run faster

[georgew5656]

Sometimes the LifecycleStop method of SupervisorManager (SupervisorManager.stop()) can take a long time to run. This is because the method iterates through all running supervisors and calls stop on them serially. Each streaming supervisor.stop() call tries to push a ShutdownNotice to its notice queue and then wait for the ShutdownNotice to run and set stopped = true up to tuningConfig.shutdownTimeout. This means the total run time can be the sum of tuningConfig.shutdownTimeout (default 80 seconds) across all supervisors.

This long stop time can cause lots of issues, most notably overlord leadership issues if the ZK leader is terminated (but the ensemble maintains quorum). This is because a overlord pod can get becomeLeader queued up behind stopLeader if it disconnects and then reconnects to ZK (the giant lock shared between the two methods).

This PR attempts to ensure SupervisorManager completes faster to prevent this issue.

Description

• In SupervisorManager use a static pool of shutdownThreads to stop supervisors in parallel in the stop method to prevent a single or few slow supervisors from slowing down overall shutdown.
• In SeekableStreamSupervisor, when stopGracefully is false (as it is when we are shutting down SupervisorManager), don't wait for the ShutdownNotice to run. This means that the recordSupplier (e.g. kafka consumer) may not be cleaned up immediately, but since all the supervisor objects are dereferenced and can be GC'd later i don't think this is a huge deal.

Fixed the bug ...

Renamed the class ...

Added a forbidden-apis entry ...

• I used a static thread pool in SupervisorManager for now, it's possible it should be configurable but IMO the main point of the thread pool is to not let a few slow supervisor shutdowns run in series block the entire supervisor manager from shutting down. (changed in review)
• I'm not sure if changing the SeekableStreamSupervisor.stop method when stopGracefully = false is necessary, but it didn't make sense to me to specify a non-graceful shutdown and then try to wait for things to clean up. (changed in review)

Release note

Improve recovery time for overlord leadership after zk nodes are bounced.

Key changed/added classes in this PR

• SupervisorManager
• SeekableStreamSupervisor

This PR has:

• been self-reviewed.
  • using the concurrency checklist (Remove this item if the PR doesn't have any relation to concurrency.)
• added documentation for new or modified features or behaviors.
• a release note entry in the PR description.
• added Javadocs for most classes and all non-trivial methods. Linked related entities via Javadoc links.
• added or updated version, license, or notice information in licenses.yaml
• added comments explaining the "why" and the intent of the code wherever would not be obvious for an unfamiliar reader.
• added unit tests or modified existing tests to cover new code paths, ensuring the threshold for code coverage is met.
• added integration tests.
• been tested in a test Druid cluster.

[kfaraz]

left some suggestions

[kfaraz]

I don't think we should use a timeout of 80s here since each supervisor could have a different value of shutdown timeout. We could either just do get() with no args (which would be no worse than what the code is currently doing) or use a longer timeout.

[kfaraz]

25 may be excessive in some cases and inadequate in others. Maybe initialize the executor lazily inside the stop() method, then the number of required threads can be computed at run time. The shutdownExec need not be a class-level field either.

---

Alternatively, instead of using a completely new executor, you could consider using the scheduledExec inside each supervisor. That executor basically just sits idle most of the time and is responsible only for submitting RunNotice to the notice queue.

You could add a stopAsync method to SeekableStreamSupervisor that does the following:

• returns a future that we coalesce and wait upon
• internally submits a runnable to the scheduledExec to perform the actual stop

I guess the only thing we will miss out on is parallelizing the autoscaler.stop() which should not be a concern, I guess?

[georgew5656]

the issue i was running into with this strategy is that part of the stop logic is shutting down the scheduledExec executor, and I couldn't really think of a great way to avoid this chicken-and-egg problem.

[kfaraz]

You could perhaps work around that problem by doing something like this:

• stopAsync sets the supervisor state to STOPPING
• stopAsync then submits a stop() runnable to the scheduledExec
• buildRunTask method should check and submit the RunNotice only if the state of the supervisor is not STOPPING
• stop() can call scheduledExec.shutdown() instead of scheduledExec.shutdownNow()

Another alternative is to simply create a shutdownExec inside stopAsync.
Difference from the current approach would be that the SupervisorManager doesn't need to handle the lifecycle of the shutdown executor.

Let me know what you think.

[georgew5656]

i think this makes sense, will update

[kfaraz]

If we have already parallelized the stop of supervisors, is this still needed?

[georgew5656]

we could probably pull it out

[kfaraz]

Maybe add a javadoc.
Also, does this need the stopGracefully parameter?

[kfaraz]

If this method throws an exception, the future should be completed with the exception.

[kfaraz]

Thanks for the changes, @georgew5656 !