static DynamicState handleEmpty(DynamicState dynamicState, StaticState staticState) throws InterruptedException, IOException {
        if (dynamicState.container != null) {
            throw new IOException("dynamicState.container is not null");
        }
        if (dynamicState.currentAssignment != null) {
            throw new IOException("dynamicState.currentAssignment is not null");
        }
        if (!dynamicState.pendingChangingBlobs.isEmpty()) {
            throw new IOException("dynamicState.pendingChangingBlobs is not empty");
        }
        if (dynamicState.pendingChangingBlobsAssignment != null) {
            throw new IOException("dynamicState.pendingChangingBlobsAssignment is not null");
        }
        if (dynamicState.pendingLocalization != null) {
            throw new IOException("dynamicState.pendingLocalization is not null");
        }
        if (dynamicState.pendingDownload != null) {
            throw new IOException("dynamicState.pendingDownload is not null");
        }

        if (!EquivalenceUtils.areLocalAssignmentsEquivalent(dynamicState.newAssignment, dynamicState.currentAssignment)) {
            return prepareForNewAssignmentNoWorkersRunning(dynamicState, staticState);
        }
        dynamicState = updateAssignmentIfNeeded(dynamicState);

        //Both assignments are null, just wait
        if (dynamicState.profileActions != null && !dynamicState.profileActions.isEmpty()) {
            //Nothing is scheduled here so throw away all of the profileActions
            LOG.warn("Dropping {} no topology is running", dynamicState.profileActions);
            dynamicState = dynamicState.withProfileActions(Collections.emptySet(), Collections.emptySet());
        }
        //Drop the change notifications we are not running anything right now
        dynamicState = drainAllChangingBlobs(dynamicState);
        Time.sleep(1000);
        return dynamicState;
    }
@Test
    public void testEmptyToEmpty() throws Exception {
        try (SimulatedTime ignored = new SimulatedTime(1010)) {
            AsyncLocalizer localizer = mock(AsyncLocalizer.class);
            LocalState state = mock(LocalState.class);
            BlobChangingCallback cb = mock(BlobChangingCallback.class);
            ContainerLauncher containerLauncher = mock(ContainerLauncher.class);
            ISupervisor iSuper = mock(ISupervisor.class);
            SlotMetrics slotMetrics = new SlotMetrics(new StormMetricsRegistry());
            StaticState staticState = new StaticState(localizer, 1000, 1000, 1000, 1000,
                containerLauncher, "localhost", 8080, iSuper, state, cb, null, null,
                slotMetrics);
            DynamicState dynamicState = new DynamicState(null, null, null, slotMetrics);
            DynamicState nextState = Slot.handleEmpty(dynamicState, staticState);
            assertEquals(MachineState.EMPTY, nextState.state);
            assertTrue(Time.currentTimeMillis() > 1000);
        }
    }