IncrementalCheckpointManager extends FSWindowDataManager implements ManagedStateComponent { protected void transferWindowFiles() { try { Long windowId = windowsToTransfer.poll(); if (windowId != null) { try { LOG.debug("transfer window {}", windowId); Map<Long, Map<Slice, Bucket.BucketedValue>> buckets = savedWindows.remove(windowId); for (Map.Entry<Long, Map<Slice, Bucket.BucketedValue>> singleBucket : buckets.entrySet()) { long bucketId = singleBucket.getKey(); managedStateContext.getBucketsFileSystem().writeBucketData(windowId, bucketId, singleBucket.getValue(), latestPurgedTimeBucket); } } catch (Throwable t) { throwable.set(t); LOG.debug("transfer window {}", windowId, t); Throwables.propagate(t); } this.lastTransferredWindow = windowId; } else { Thread.sleep(waitMillis); } } catch (InterruptedException ex) { LOG.debug("interrupted", ex); } } 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 testTransferWindowFiles() throws IOException, InterruptedException { testMeta.checkpointManager.setup(testMeta.managedStateContext); Map<Long, Map<Slice, Bucket.BucketedValue>> buckets5 = ManagedStateTestUtils.getTestData(0, 5, 0); testMeta.checkpointManager.save(buckets5, 10, false); testMeta.checkpointManager.teardown(); Thread.sleep(500); testMeta.checkpointManager.committed(10); testMeta.checkpointManager.transferWindowFiles(); for (int i = 0; i < 5; i++) { ManagedStateTestUtils.validateBucketOnFileSystem(testMeta.managedStateContext.getFileAccess(), i, buckets5.get((long)i), 1); } }