[ https://issues.apache.org/jira/browse/FLINK-5056?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15668371#comment-15668371 ]
ASF GitHub Bot commented on FLINK-5056: --------------------------------------- Github user kl0u commented on the issue: https://github.com/apache/flink/pull/2797 The subtask matters for files that were opened but they did not make it into a checkpoint because a failure occurred after they were opened but before a checkpoint barrier arrived. These are the ones that are not cleaned up anymore. The rest, that are part of the checkpoint, they are "cleaned up” or rolled back if needed. Probably you may understand more how the cleaning up was happening if you checkout the code in the master for the cleanup. > On Nov 15, 2016, at 10:12 PM, zentol <notificati...@github.com> wrote: > > @zentol commented on this pull request. > > In flink-streaming-connectors/flink-connector-filesystem/src/main/java/org/apache/flink/streaming/connectors/fs/bucketing/BucketingSink.java <https://github.com/apache/flink/pull/2797>: > > > } > > - @Override > - public void restoreState(State<T> state) { > - this.state = state; > - > - try { > - initFileSystem(); > - } catch (IOException e) { > - LOG.error("Error while creating FileSystem in checkpoint restore.", e); > - throw new RuntimeException("Error while creating FileSystem in checkpoint restore.", e); > - } > + private void handleRestoredBucketState(BucketState<T> bucketState) { > + // we can clean all the pending files since they were renamed to > I don't understand why the subtask index even matters. In a simplified view, every subtask maintains a set of Files. When rescaling/restoring state, this set of files is given to another subtask. If this set contains a file marked as pending, it should be allowed to delete it, since no other task should have the same file since the state containing the information about the file (aka, the file path) is not given to multiple subtasks. > > — > You are receiving this because you authored the thread. > Reply to this email directly, view it on GitHub <https://github.com/apache/flink/pull/2797>, or mute the thread <https://github.com/notifications/unsubscribe-auth/ACS1qP4VEEL7BZsKH1rlECWjHcQutwANks5q-iBSgaJpZM4KxHhq>. > > BucketingSink deletes valid data when checkpoint notification is slow. > ---------------------------------------------------------------------- > > Key: FLINK-5056 > URL: https://issues.apache.org/jira/browse/FLINK-5056 > Project: Flink > Issue Type: Bug > Components: filesystem-connector > Affects Versions: 1.1.3 > Reporter: Kostas Kloudas > Assignee: Kostas Kloudas > Fix For: 1.2.0 > > > Currently if BucketingSink receives no data after a checkpoint and then a > notification about a previous checkpoint arrives, it clears its state. This > can > lead to not committing valid data about intermediate checkpoints for whom > a notification has not arrived yet. As a simple sequence that illustrates the > problem: > -> input data > -> snapshot(0) > -> input data > -> snapshot(1) > -> no data > -> notifyCheckpointComplete(0) > the last will clear the state of the Sink without committing as final the > data > that arrived for checkpoint 1. > -- This message was sent by Atlassian JIRA (v6.3.4#6332)