static DynamicState stateMachineStep(DynamicState dynamicState, StaticState staticState) throws Exception {
        LOG.debug("STATE {}", dynamicState.state);
        switch (dynamicState.state) {
            case EMPTY:
                return handleEmpty(dynamicState, staticState);
            case RUNNING:
                return handleRunning(dynamicState, staticState);
            case WAITING_FOR_WORKER_START:
                return handleWaitingForWorkerStart(dynamicState, staticState);
            case KILL_BLOB_UPDATE:
                return handleKillBlobUpdate(dynamicState, staticState);
            case KILL_AND_RELAUNCH:
                return handleKillAndRelaunch(dynamicState, staticState);
            case KILL:
                return handleKill(dynamicState, staticState);
            case WAITING_FOR_BLOB_LOCALIZATION:
                return handleWaitingForBlobLocalization(dynamicState, staticState);
            case WAITING_FOR_BLOB_UPDATE:
                return handleWaitingForBlobUpdate(dynamicState, staticState);
            default:
                throw new IllegalStateException("Code not ready to handle a state of " + dynamicState.state);
        }
    }
@Test
    public void testResourcesChangedFiltered() throws Exception {
        try (SimulatedTime ignored = new SimulatedTime(1010)) {
            int port = 8080;
            String cTopoId = "CURRENT";
            List<ExecutorInfo> cExecList = mkExecutorInfoList(1, 2, 3, 4, 5);
            LocalAssignment cAssignment =
                mkLocalAssignment(cTopoId, cExecList, mkWorkerResources(100.0, 100.0, 100.0));

            String otherTopoId = "OTHER";
            LocalAssignment otherAssignment = mkLocalAssignment(otherTopoId, cExecList, mkWorkerResources(100.0, 100.0, 100.0));

            BlobChangingCallback cb = mock(BlobChangingCallback.class);

            Container cContainer = mock(Container.class);
            LSWorkerHeartbeat chb = mkWorkerHB(cTopoId, port, cExecList, Time.currentTimeSecs());
            when(cContainer.readHeartbeat()).thenReturn(chb);
            when(cContainer.areAllProcessesDead()).thenReturn(false, false, true);
            AsyncLocalizer localizer = mock(AsyncLocalizer.class);
            Container nContainer = mock(Container.class);
            LocalState state = mock(LocalState.class);
            ContainerLauncher containerLauncher = mock(ContainerLauncher.class);
            when(containerLauncher.launchContainer(port, cAssignment, state)).thenReturn(nContainer);
            when(nContainer.readHeartbeat()).thenReturn(chb, chb);

            ISupervisor iSuper = mock(ISupervisor.class);
            long heartbeatTimeoutMs = 5000;
            StaticState staticState = new StaticState(localizer, heartbeatTimeoutMs, 120_000, 1000, 1000,
                                                      containerLauncher, "localhost", port, iSuper, state, cb, null, null, new SlotMetrics(new StormMetricsRegistry()));

            Set<Slot.BlobChanging> changing = new HashSet<>();

            LocallyCachedBlob stormJar = mock(LocallyCachedBlob.class);
            GoodToGo.GoodToGoLatch stormJarLatch = mock(GoodToGo.GoodToGoLatch.class);
            CompletableFuture<Void> stormJarLatchFuture = mock(CompletableFuture.class);
            when(stormJarLatch.countDown()).thenReturn(stormJarLatchFuture);
            changing.add(new Slot.BlobChanging(cAssignment, stormJar, stormJarLatch));
            Set<Slot.BlobChanging> desired = new HashSet<>(changing);

            LocallyCachedBlob otherJar = mock(LocallyCachedBlob.class);
            GoodToGo.GoodToGoLatch otherJarLatch = mock(GoodToGo.GoodToGoLatch.class);
            changing.add(new Slot.BlobChanging(otherAssignment, otherJar, otherJarLatch));

            SlotMetrics slotMetrics = new SlotMetrics(new StormMetricsRegistry());
            DynamicState dynamicState = new DynamicState(cAssignment, cContainer, cAssignment, slotMetrics).withChangingBlobs(changing);

            DynamicState nextState = Slot.stateMachineStep(dynamicState, staticState);
            assertEquals(MachineState.KILL_BLOB_UPDATE, nextState.state);
            verify(iSuper).killedWorker(port);
            verify(cContainer).kill();
            verify(localizer, never()).requestDownloadTopologyBlobs(any(), anyInt(), any());
            verify(stormJarLatch, never()).countDown();
            verify(otherJarLatch, times(1)).countDown();
            assertNull(nextState.pendingDownload);
            assertNull(nextState.pendingLocalization);
            assertEquals(desired, nextState.changingBlobs);
            assertTrue(nextState.pendingChangingBlobs.isEmpty());
            assertNull(nextState.pendingChangingBlobsAssignment);
            assertThat(Time.currentTimeMillis(), greaterThan(1000L));

            nextState = Slot.stateMachineStep(nextState, staticState);
            assertEquals(MachineState.KILL_BLOB_UPDATE, nextState.state);
            verify(cContainer).forceKill();
            assertNull(nextState.pendingDownload);
            assertNull(nextState.pendingLocalization);
            assertEquals(desired, nextState.changingBlobs);
            assertTrue(nextState.pendingChangingBlobs.isEmpty());
            assertNull(nextState.pendingChangingBlobsAssignment);
            assertThat(Time.currentTimeMillis(), greaterThan(2000L));

            nextState = Slot.stateMachineStep(nextState, staticState);
            assertEquals(MachineState.WAITING_FOR_BLOB_UPDATE, nextState.state);
            verify(cContainer).cleanUp();
            assertThat(Time.currentTimeMillis(), greaterThan(2000L));

            nextState = Slot.stateMachineStep(nextState, staticState);
            verify(stormJarLatchFuture).get(anyLong(), any());
            verify(containerLauncher).launchContainer(port, cAssignment, state);
            assertEquals(MachineState.WAITING_FOR_WORKER_START, nextState.state);
            assertNull(nextState.pendingChangingBlobsAssignment);
            assertTrue(nextState.pendingChangingBlobs.isEmpty());
            assertSame(cAssignment, nextState.currentAssignment);
            assertSame(nContainer, nextState.container);
            assertThat(Time.currentTimeMillis(), greaterThan(2000L));
            assertThat(Time.currentTimeMillis(), lessThan(heartbeatTimeoutMs));

            nextState = Slot.stateMachineStep(nextState, staticState);
            assertEquals(MachineState.RUNNING, nextState.state);
            assertNull(nextState.pendingChangingBlobsAssignment);
            assertTrue(nextState.pendingChangingBlobs.isEmpty());
            assertSame(cAssignment, nextState.currentAssignment);
            assertSame(nContainer, nextState.container);
            assertTrue(Time.currentTimeMillis() > 2000);

            nextState = Slot.stateMachineStep(nextState, staticState);
            assertEquals(MachineState.RUNNING, nextState.state);
            assertNull(nextState.pendingChangingBlobsAssignment);
            assertTrue(nextState.pendingChangingBlobs.isEmpty());
            assertSame(cAssignment, nextState.currentAssignment);
            assertSame(nContainer, nextState.container);
            assertTrue(Time.currentTimeMillis() > 3000);

            nextState = Slot.stateMachineStep(nextState, staticState);
            assertEquals(MachineState.RUNNING, nextState.state);
            assertNull(nextState.pendingChangingBlobsAssignment);
            assertTrue(nextState.pendingChangingBlobs.isEmpty());
            assertSame(cAssignment, nextState.currentAssignment);
            assertSame(nContainer, nextState.container);
            assertTrue(Time.currentTimeMillis() > 4000);
        }
    }