IncrementalCheckpointManager extends FSWindowDataManager implements ManagedStateComponent { @Override public void committed(long committedWindowId) throws IOException { LOG.debug("data manager committed {}", committedWindowId); for (Long currentWindow : savedWindows.keySet()) { if (currentWindow <= largestWindowAddedToTransferQueue) { continue; } if (currentWindow <= committedWindowId) { LOG.debug("to transfer {}", currentWindow); largestWindowAddedToTransferQueue = currentWindow; windowsToTransfer.add(currentWindow); } else { break; } } } IncrementalCheckpointManager(); @Override void setup(Context.OperatorContext context); @Override void setup(@NotNull final ManagedStateContext managedStateContext); @Override void save(Object object, long windowId); void save(Map<Long, Map<Slice, Bucket.BucketedValue>> unsavedData, long windowId,
      boolean skipWriteToWindowFile); Map<Long, Object> retrieveAllWindows(); @Override void committed(long committedWindowId); @Override void teardown(); void setLatestExpiredTimeBucket(long timeBucket); long getLastTransferredWindow();  }
@Test public void testCommitted() throws IOException, InterruptedException { CountDownLatch latch = new CountDownLatch(5); MockBucketsFileSystem mockBucketsFileSystem = new MockBucketsFileSystem(latch); testMeta.managedStateContext.setBucketsFileSystem(mockBucketsFileSystem); mockBucketsFileSystem.setup(testMeta.managedStateContext); testMeta.checkpointManager.setup(testMeta.managedStateContext); Map<Long, Map<Slice, Bucket.BucketedValue>> data = ManagedStateTestUtils.getTestData(0, 5, 0); testMeta.checkpointManager.save(data, 10, false); testMeta.checkpointManager.committed(10); latch.await(); testMeta.checkpointManager.teardown(); Thread.sleep(500); for (int i = 0; i < 5; i++) { ManagedStateTestUtils.validateBucketOnFileSystem(testMeta.managedStateContext.getFileAccess(), i, data.get((long)i), 1); } }