@Nonnull @Override
    public ProgressState call() {
        progTracker.reset();
        stateMachineStep();
        return progTracker.toProgressState();
    }
@Test
    public void when_itemAndBarrier_then_snapshotDone() {
        // When
        Entry<String, String> entry = entry("k", "v");
        init(asList(entry, new SnapshotBarrier(2, false)));
        ssContext.startNewSnapshotPhase1(2, "map", 0);
        assertEquals(2, sst.pendingSnapshotId);
        assertEquals(MADE_PROGRESS, sst.call());
        assertEquals(MADE_PROGRESS, sst.call());
        mockSsWriter.hasPendingFlushes = false;
        assertEquals(MADE_PROGRESS, sst.call());

        // Then
        assertEquals(3, sst.pendingSnapshotId);
        assertEquals(entry(serialize("k"), serialize("v")), mockSsWriter.poll());
    }