When using the node shutdown API, we have observed that the API will indicate that shutdown is stalled due to a shard not being able to move. This can happen after several hours of progress on a busy cluster. 

Currently, the expectation is that the caller observes the stalled status and then waits some reasonable amount of time before giving up on shutdown making progress. One reason is to give ILM enough time to potentially address any issues that would prevent shards from migrating. This has the unfortunate side effect of pushing more logic to the callers of the API whereas if Elasticsearch handled this we would be much more likely to know if the shard migration will ever be un-stalled and/or make better decisions about how long to wait.
x-pack/plugin/shutdown/src/test/java/org/elasticsearch/xpack/shutdown/TransportGetShutdownStatusActionTests.java
import static org.mockito.Mockito.spy;
public class TransportGetShutdownStatusActionTests extends ESTestCase {
    public static final String SHUTTING_DOWN_NODE_ID = "node1";
    public static final String LIVE_NODE_ID = "node2";
    public static final String OTHER_LIVE_NODE_ID = "node3";
    private final AtomicReference<TestDecider> canAllocate = new AtomicReference<>();
    private final AtomicReference<TestDecider> canRemain = new AtomicReference<>();
    private ClusterInfoService clusterInfoService;
    private AllocationDeciders allocationDeciders;
    private AllocationService allocationService;
    private SnapshotsInfoService snapshotsInfoService;
    @FunctionalInterface
    private interface TestDecider {
        Decision test(ShardRouting shardRouting, RoutingNode node, RoutingAllocation allocation);
    }
    @Before
    private void setup() {
        canAllocate.set((r, n, a) -> { throw new UnsupportedOperationException("canAllocate not initiated in this test"); });
        canRemain.set((r, n, a) -> { throw new UnsupportedOperationException("canRemain not initiated in this test"); });
        clusterInfoService = EmptyClusterInfoService.INSTANCE;
        allocationDeciders = new AllocationDeciders(
            List.of(new NodeShutdownAllocationDecider(), new NodeReplacementAllocationDecider(), new AllocationDecider() {
                @Override
                public Decision canAllocate(ShardRouting shardRouting, RoutingNode node, RoutingAllocation allocation) {
                    return canAllocate.get().test(shardRouting, node, allocation);
                }
                @Override
                public Decision canRebalance(ShardRouting shardRouting, RoutingAllocation allocation) {
                    
                    return Decision.NO;