[
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)
