@Test public void testInactivityPeriodWithLateNotify() throws Exception { final File outDir = tempFolder.newFolder(); OneInputStreamOperatorTestHarness<String, Object> testHarness = createRescalingTestSink(outDir, 1, 0, 100); testHarness.setup(); testHarness.open(); testHarness.setProcessingTime(0L); testHarness.processElement(new StreamRecord<>("test1", 1L)); testHarness.processElement(new StreamRecord<>("test2", 1L)); checkLocalFs(outDir, 2, 0 , 0, 0); testHarness.setProcessingTime(101L); checkLocalFs(outDir, 0, 2, 0, 0); testHarness.snapshot(0, 0); checkLocalFs(outDir, 0, 2, 0, 0); testHarness.processElement(new StreamRecord<>("test3", 1L)); testHarness.processElement(new StreamRecord<>("test4", 1L)); testHarness.setProcessingTime(202L); testHarness.snapshot(1, 0); checkLocalFs(outDir, 0, 4, 0, 0); testHarness.notifyOfCompletedCheckpoint(0); checkLocalFs(outDir, 0, 2, 2, 0); testHarness.notifyOfCompletedCheckpoint(1); checkLocalFs(outDir, 0, 0, 4, 0); }
@Override public void open(Configuration parameters) throws Exception { super.open(parameters); state = new State<>(); processingTimeService = ((StreamingRuntimeContext) getRuntimeContext()).getProcessingTimeService(); long currentProcessingTime = processingTimeService.getCurrentProcessingTime(); processingTimeService.registerTimer(currentProcessingTime + inactiveBucketCheckInterval, this); this.clock = new Clock() { @Override public long currentTimeMillis() { return processingTimeService.getCurrentProcessingTime(); } }; }
BucketingSink extends RichSinkFunction<T> implements InputTypeConfigurable, CheckpointedFunction, CheckpointListener, ProcessingTimeCallback { @Override public void open(Configuration parameters) throws Exception { super.open(parameters); state = new State<>(); processingTimeService = ((StreamingRuntimeContext) getRuntimeContext()).getProcessingTimeService(); long currentProcessingTime = processingTimeService.getCurrentProcessingTime(); processingTimeService.registerTimer(currentProcessingTime + inactiveBucketCheckInterval, this); this.clock = new Clock() { @Override public long currentTimeMillis() { return processingTimeService.getCurrentProcessingTime(); } }; } }
BucketingSink extends RichSinkFunction<T> implements InputTypeConfigurable, CheckpointedFunction, CheckpointListener, ProcessingTimeCallback { @Override public void open(Configuration parameters) throws Exception { super.open(parameters); state = new State<>(); processingTimeService = ((StreamingRuntimeContext) getRuntimeContext()).getProcessingTimeService(); long currentProcessingTime = processingTimeService.getCurrentProcessingTime(); processingTimeService.registerTimer(currentProcessingTime + inactiveBucketCheckInterval, this); this.clock = new Clock() { @Override public long currentTimeMillis() { return processingTimeService.getCurrentProcessingTime(); } }; } BucketingSink(String basePath); }
BucketingSink extends RichSinkFunction<T> implements InputTypeConfigurable, CheckpointedFunction, CheckpointListener, ProcessingTimeCallback { @Override public void open(Configuration parameters) throws Exception { super.open(parameters); state = new State<>(); processingTimeService = ((StreamingRuntimeContext) getRuntimeContext()).getProcessingTimeService(); long currentProcessingTime = processingTimeService.getCurrentProcessingTime(); processingTimeService.registerTimer(currentProcessingTime + inactiveBucketCheckInterval, this); this.clock = new Clock() { @Override public long currentTimeMillis() { return processingTimeService.getCurrentProcessingTime(); } }; } BucketingSink(String basePath); BucketingSink<T> setFSConfig(Configuration config); BucketingSink<T> setFSConfig(org.apache.hadoop.conf.Configuration config); @Override @SuppressWarnings("unchecked") void setInputType(TypeInformation<?> type, ExecutionConfig executionConfig); @Override void initializeState(FunctionInitializationContext context); @Override void open(Configuration parameters); @Override void close(); @Override void invoke(T value); @Override void onProcessingTime(long timestamp); @Override void notifyCheckpointComplete(long checkpointId); @Override void snapshotState(FunctionSnapshotContext context); BucketingSink<T> setBatchSize(long batchSize); BucketingSink<T> setBatchRolloverInterval(long batchRolloverInterval); BucketingSink<T> setInactiveBucketCheckInterval(long interval); BucketingSink<T> setInactiveBucketThreshold(long threshold); BucketingSink<T> setBucketer(Bucketer<T> bucketer); BucketingSink<T> setWriter(Writer<T> writer); BucketingSink<T> setInProgressSuffix(String inProgressSuffix); BucketingSink<T> setInProgressPrefix(String inProgressPrefix); BucketingSink<T> setPendingSuffix(String pendingSuffix); BucketingSink<T> setPendingPrefix(String pendingPrefix); BucketingSink<T> setValidLengthSuffix(String validLengthSuffix); BucketingSink<T> setValidLengthPrefix(String validLengthPrefix); BucketingSink<T> setPartSuffix(String partSuffix); BucketingSink<T> setPartPrefix(String partPrefix); BucketingSink<T> setUseTruncate(boolean useTruncate); @Deprecated BucketingSink<T> disableCleanupOnOpen(); BucketingSink<T> setAsyncTimeout(long timeout); @VisibleForTesting State<T> getState(); static FileSystem createHadoopFileSystem(
			Path path,
			@Nullable Configuration extraUserConf); }
BucketingSink extends RichSinkFunction<T> implements InputTypeConfigurable, CheckpointedFunction, CheckpointListener, ProcessingTimeCallback { @Override public void open(Configuration parameters) throws Exception { super.open(parameters); state = new State<>(); processingTimeService = ((StreamingRuntimeContext) getRuntimeContext()).getProcessingTimeService(); long currentProcessingTime = processingTimeService.getCurrentProcessingTime(); processingTimeService.registerTimer(currentProcessingTime + inactiveBucketCheckInterval, this); this.clock = new Clock() { @Override public long currentTimeMillis() { return processingTimeService.getCurrentProcessingTime(); } }; } BucketingSink(String basePath); BucketingSink<T> setFSConfig(Configuration config); BucketingSink<T> setFSConfig(org.apache.hadoop.conf.Configuration config); @Override @SuppressWarnings("unchecked") void setInputType(TypeInformation<?> type, ExecutionConfig executionConfig); @Override void initializeState(FunctionInitializationContext context); @Override void open(Configuration parameters); @Override void close(); @Override void invoke(T value); @Override void onProcessingTime(long timestamp); @Override void notifyCheckpointComplete(long checkpointId); @Override void snapshotState(FunctionSnapshotContext context); BucketingSink<T> setBatchSize(long batchSize); BucketingSink<T> setBatchRolloverInterval(long batchRolloverInterval); BucketingSink<T> setInactiveBucketCheckInterval(long interval); BucketingSink<T> setInactiveBucketThreshold(long threshold); BucketingSink<T> setBucketer(Bucketer<T> bucketer); BucketingSink<T> setWriter(Writer<T> writer); BucketingSink<T> setInProgressSuffix(String inProgressSuffix); BucketingSink<T> setInProgressPrefix(String inProgressPrefix); BucketingSink<T> setPendingSuffix(String pendingSuffix); BucketingSink<T> setPendingPrefix(String pendingPrefix); BucketingSink<T> setValidLengthSuffix(String validLengthSuffix); BucketingSink<T> setValidLengthPrefix(String validLengthPrefix); BucketingSink<T> setPartSuffix(String partSuffix); BucketingSink<T> setPartPrefix(String partPrefix); BucketingSink<T> setUseTruncate(boolean useTruncate); @Deprecated BucketingSink<T> disableCleanupOnOpen(); BucketingSink<T> setAsyncTimeout(long timeout); @VisibleForTesting State<T> getState(); static FileSystem createHadoopFileSystem(
			Path path,
			@Nullable Configuration extraUserConf);  }