[FLINK-10840] BucketingSink incorrectly clears the pendingFiles List

• Type: Bug
• Status: Closed
• Priority: Major
• Resolution: Not A Bug
• Component/s: Connectors / Common
• Assignee: vinoyang
• Reporter: vinoyang
• Created: 09/Nov/18 12:32
• Updated: 02/Oct/19 17:49
• Resolved: 12/Nov/18 07:22

Description

BucketingSink#snapshotState : (see the comment in this method)

public void snapshotState(FunctionSnapshotContext context) throws Exception {
   Preconditions.checkNotNull(restoredBucketStates, "The operator has not been properly initialized.");

   restoredBucketStates.clear();

   synchronized (state.bucketStates) {
      int subtaskIdx = getRuntimeContext().getIndexOfThisSubtask();

      for (Map.Entry<String, BucketState<T>> bucketStateEntry : state.bucketStates.entrySet()) {
         BucketState<T> bucketState = bucketStateEntry.getValue();

         if (bucketState.isWriterOpen) {
            bucketState.currentFileValidLength = bucketState.writer.flush();
         }

         synchronized (bucketState.pendingFilesPerCheckpoint) {
            bucketState.pendingFilesPerCheckpoint.put(context.getCheckpointId(), bucketState.pendingFiles);
         }
         //This operation will make this collection prematurely emptied
         bucketState.pendingFiles = new ArrayList<>();
      }
      restoredBucketStates.add(state);

      if (LOG.isDebugEnabled()) {
         LOG.debug("{} idx {} checkpointed {}.", getClass().getSimpleName(), subtaskIdx, state);
      }
   }
}