[controller] Fix multiple AdminExecutionTasks working on the same store at the same time.

[haoxu07]

Summary, imperative, start upper case, don't end with a period

Current state: when we create AdminExecutionTask , we did not check if there is AdminExecutionTask for one store on the fly, so if one AdminExecutionTask is waiting for a lock (e.g. updating store operation waiting for the store level write lock and that lock is currently unavailable), it is possible that AdminConsumptionTask will create multiple AdminExecutionTask s for a single store until they occupy all threads from ExecutorService 's thread pool, they are all waiting for the same store-level lock.

Besides, executorService.invokeAll(tasks, processingCycleTimeoutInMs, TimeUnit.MILLISECONDS)
will cancel every invoked tasks, even the task is emitted by not getting thread from pool to execute. ExecutorService will try to cancel the AdminExecutionTask waiting for a lock after timeout, but that will not terminate that thread (as the acquiring locking operation by AutoCloseableLock is not interruptible). Then that AdminExecutionTask will keep occupying one thread from ExecutorService 's thread pool until the lock is released.

This PR has the following changes to address the issue:

1. Adding check to see if there is AdminExecutionTask is running for a store.
2. Integration test to simulate the lock blocking one thread from pool and recover after acquiring the lock.

How was this PR tested?

CI

Does this PR introduce any user-facing changes?

• No. You can skip the rest of this section.
• Yes. Make sure to explain your proposed changes and call out the behavior change.

[gaojieliu]

I have a few questions:

1. is it expected to run some admin logic, which is not interruptible? I think we should make all the logic interruptible, otherwise, it can block the graceful shutdown.
2. Is there any particular type of admin message, which would stuck or take longer time than others?

We face a similar situation when implementing async gauge execution, and we adopted the following strategy, which seems to work well:
tehuti-io/tehuti#31

cc @xunyin8

[haoxu07]

I have a few questions:

1. is it expected to run some admin logic, which is not interruptible? I think we should make all the logic interruptible, otherwise, it can block the graceful shutdown.
2. Is there any particular type of admin message, which would stuck or take longer time than others?

We face a similar situation when implementing async gauge execution, and we adopted the following strategy, which seems to work well: tehuti-io/tehuti#31

cc @xunyin8

Regarding your questions:
One case for logic non-interruptible is: our current store level lock acquiring inside VeniceHelixAdmin through AutoCloseableLock is not interruptible by using lock.lock():

venice/internal/venice-common/src/main/java/com/linkedin/venice/utils/locks/AutoCloseableMultiLock.java

Line 18 in f07d0c1

locks[i].lock();

. When a thread is waiting for the lock, it is not interruptible. We could improve this by using lock.lockInterruptibly(), but I found it will some test failure, which I am trying to figure out.

But thanks for suggested solutions, let me figure out if it works.

[lluwm]

Looks good overall, just some comments on the integration part.

[lluwm]

@haoxu07, LGTM and thanks for the fix.