[FLINK-4534] Lack of synchronization in BucketingSink#restoreState()

• Type: Bug
• Status: Closed
• Priority: Major
• Resolution: Not A Problem
• Component/s: Connectors / Common
• Labels: pull-request-available
• Assignee: zhangminglei
• Reporter: Ted Yu
• Created: 30/Aug/16 17:32
• Updated: 29/Aug/18 15:24
• Resolved: 29/Aug/18 15:24

Description

Iteration over state.bucketStates is protected by synchronization in other methods, except for the following in restoreState():

    for (BucketState<T> bucketState : state.bucketStates.values()) {

and following in close():

    for (Map.Entry<String, BucketState<T>> entry : state.bucketStates.entrySet()) {
      closeCurrentPartFile(entry.getValue());

w.r.t. bucketState.pendingFilesPerCheckpoint , there is similar issue starting line 752:

      Set<Long> pastCheckpointIds = bucketState.pendingFilesPerCheckpoint.keySet();
      LOG.debug("Moving pending files to final location on restore.");
      for (Long pastCheckpointId : pastCheckpointIds) {

Issue Links

• links to: GitHub Pull Request #4482
• links to: GitHub Pull Request #4482