StoreSnapshotTasklet implements Tasklet { @Nonnull @Override public ProgressState call() { progTracker.reset(); stateMachineStep(); return progTracker.toProgressState(); } StoreSnapshotTasklet(
            SnapshotContext snapshotContext,
            InboundEdgeStream inboundEdgeStream,
            AsyncSnapshotWriter ssWriter,
            ILogger logger,
            String vertexName,
            boolean isHigherPrioritySource
    ); @Nonnull @Override ProgressState call(); @Override void provideDynamicMetrics(MetricDescriptor descriptor, MetricsCollectionContext context); @Override String toString();  }
@Test public void when_doneItemOnInput_then_eventuallyDone() { init(singletonList(DONE_ITEM)); assertEquals(DONE, sst.call()); }
@Test public void when_item_then_offeredToSsWriter() { init(singletonList(entry("k", "v"))); assertEquals(MADE_PROGRESS, sst.call()); assertEquals(entry(serialize("k"), serialize("v")), mockSsWriter.poll()); assertNull(mockSsWriter.poll()); }
@Test public void when_notAbleToOffer_then_offeredLater() { init(singletonList(entry("k", "v"))); mockSsWriter.ableToOffer = false; assertEquals(MADE_PROGRESS, sst.call()); assertEquals(0, input.remainingItems().size()); assertEquals(NO_PROGRESS, sst.call()); assertNull(mockSsWriter.poll()); mockSsWriter.ableToOffer = true; assertEquals(MADE_PROGRESS, sst.call()); assertEquals(entry(serialize("k"), serialize("v")), mockSsWriter.poll()); assertNull(mockSsWriter.poll()); }
@Test public void when_barrier_then_snapshotDone() { init(singletonList(new SnapshotBarrier(2, false))); ssContext.startNewSnapshotPhase1(2, "map", 0); assertEquals(MADE_PROGRESS, sst.call()); assertEquals(MADE_PROGRESS, sst.call()); assertEquals(3, sst.pendingSnapshotId); }
@Test public void when_itemAndBarrier_then_snapshotDone() { 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()); mockSsWriter.hasPendingFlushes = false; assertEquals(MADE_PROGRESS, sst.call()); assertEquals(3, sst.pendingSnapshotId); assertEquals(entry(serialize("k"), serialize("v")), mockSsWriter.poll()); }
@Test public void when_notAbleToFlush_then_tryAgain() { init(singletonList(new SnapshotBarrier(2, false))); ssContext.startNewSnapshotPhase1(2, "map", 0); mockSsWriter.ableToFlushRemaining = false; assertEquals(MADE_PROGRESS, sst.call()); assertEquals(NO_PROGRESS, sst.call()); assertEquals(NO_PROGRESS, sst.call()); mockSsWriter.ableToFlushRemaining = true; assertEquals(MADE_PROGRESS, sst.call()); assertEquals(MADE_PROGRESS, sst.call()); assertEquals(NO_PROGRESS, sst.call()); assertEquals(3, sst.pendingSnapshotId); }
@Test public void test_waitingForFlushesToComplete() { Entry<String, String> entry = entry("k", "v"); init(asList(entry, new SnapshotBarrier(2, false))); ssContext.startNewSnapshotPhase1(2, "map", 0); assertEquals(MADE_PROGRESS, sst.call()); assertEquals(NO_PROGRESS, sst.call()); assertTrue(mockSsWriter.hasPendingFlushes); mockSsWriter.hasPendingFlushes = false; assertEquals(MADE_PROGRESS, sst.call()); assertEquals(NO_PROGRESS, sst.call()); assertEquals(3, sst.pendingSnapshotId); }