In the current code, the blocking snapshot examines the running and the snapshot's outcome to determine whether it should resume the streaming thread, shown here:

                    if (running && snapshotResult.isCompletedOrSkipped()) {
                        resumeStreaming(partition);
                    }

This can lead to an unintended state in which the streaming thread is not resumed because the coordinator is currently being stopped, the stop request is waiting for the streaming thread run by the executor to end, or the snapshot status is not completed or skipped.

I believe we should always resume the streaming and set the metrics back on the dispatcher, regardless of a finally block, and remove the predication on this behavior to avoid the appearance of a deadlock.