[server][dvc] Drop partitions asynchronously

[kvargha]

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

When a storage node is transitioning from LEADER -> STANDBY -> OFFLINE -> DROPPED, a race condition can occur. Specifically, the DROPPED state transition may be executed synchronously before the other state transitions are processed (LEADER -> STANDBY -> OFFLINE are executed asynchronously). This results in the store partition being deleted prematurely. Consequently, when the LEADER -> STANDBY message is eventually processed, it triggers a PersistenceFailureException since the storage partition no longer exists.

The solution for this is to drop the store partition asynchronously by adding a DROP_PARTITION message to the consumerActionsQueue if the ingestion task is still running. In the case that it's not running (this can happen if it was killed), the store partition will be dropped synchronously.

How was this PR tested?

Added unit and integration tests.

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.

[lluwm]

This exception could cause the ST to be in the ERROR state, is that right? I read function stopConsumptionAndWait and, today, we simply log a warning message if consumption couldn't be stopped in time. This sounds like a behavior change in the new PR and we probably want to be careful about it.

[kvargha]

STANDBY->OFFLINE issues an UNSUBSCRIBE message, and so will stopConsumptionAndWait.

By the time SIT processes the DROP_PARTITION message, it should have been already unsubscribed.

I think it's safe to remove this check. What do you think?

[lluwm]

I am thinking of a race condition that after we add DROP_PARTITION actions to the queue, then SIT terminates due to some exceptions (as we see several cases today) before executing all the remaining actions from the queue and it could probably cause some partition leaks. If this race is possible, we probably need to add some logic in the SIT to make sure that all DROP_PARTITION actions have to be executed before it can terminate itself, or maybe some other measures to avoid it.