You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by kl0u <gi...@git.apache.org> on 2016/10/27 16:56:46 UTC
[GitHub] flink pull request #2707: [FLINK-4939] GenericWriteAheadSink: Decouple the c...
GitHub user kl0u opened a pull request:
https://github.com/apache/flink/pull/2707
[FLINK-4939] GenericWriteAheadSink: Decouple the creating from the committing subtask for a pending checkpoint
So far the GenericWriteAheadSink expected that the subtask that wrote a temporary buffer to the
state backend upon checkpointing, will be also the one to commit it to the third-party storage system.
This commit removes this assumption. To do this it changes the CheckpointCommitter to dynamically take the subtaskIdx as a parameter when committing something to the third-party storage system ( [void commitCheckpoint(int subtaskIdx, long checkpointID)] ) and when asking
if a checkpoint was committed ( [boolean isCheckpointCommitted(int subtaskIdx, long checkpointID)] ) and also changes the state kept by the [GenericWriteAheadSink] to also
include that subtask index of the subtask that wrote the pending buffer.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/kl0u/flink write_ahead_sink
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/2707.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #2707
----
commit 207c5239c9fab6ef09b7bdd410ee83d3d8ca105f
Author: kl0u <kk...@gmail.com>
Date: 2016-10-26T15:19:12Z
[FLINK-4939] GenericWriteAheadSink: Decouple the creating from the committing subtask for a pending checkpoint
So far the GenericWriteAheadSink expected that
the subtask that wrote a temporary buffer to the
state backend, will be also the one to commit it to
the third-party storage system.
This commit removes this assumption. To do this
it changes the CheckpointCommitter to dynamically
take the subtaskIdx as a parameter when asking
if a checkpoint was committed and also changes the
state kept by the GenericWriteAheadSink to also
include that subtask index of the subtask that wrote
the pending buffer.
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2707: [FLINK-4939] GenericWriteAheadSink: Decouple the creating...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on the issue:
https://github.com/apache/flink/pull/2707
merging
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2707: [FLINK-4939] GenericWriteAheadSink: Decouple the c...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/2707#discussion_r86109500
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericWriteAheadSink.java ---
@@ -142,46 +184,62 @@ private void cleanState() throws Exception {
public void notifyOfCompletedCheckpoint(long checkpointId) throws Exception {
super.notifyOfCompletedCheckpoint(checkpointId);
- synchronized (state.pendingHandles) {
- Set<Long> pastCheckpointIds = state.pendingHandles.keySet();
- Set<Long> checkpointsToRemove = new HashSet<>();
- for (Long pastCheckpointId : pastCheckpointIds) {
+ synchronized (pendingHandles) {
+ Set<PendingCheckpointId> checkpointsToRemove = new HashSet<>();
+
+ for (Map.Entry<PendingCheckpointId, PendingHandle> pendingHandle : pendingHandles.entrySet()) {
+
+ PendingCheckpointId pendingId = pendingHandle.getKey();
+ long pastCheckpointId = pendingId.checkpointId;
+ int subtaskId = pendingId.subtaskId;
+
if (pastCheckpointId <= checkpointId) {
+
+ PendingHandle handle = pendingHandle.getValue();
+ long timestamp = handle.timestamp;
+ StreamStateHandle streamHandle = handle.stateHandle;
+
try {
- if (!committer.isCheckpointCommitted(pastCheckpointId)) {
- Tuple2<Long, StreamStateHandle> handle = state.pendingHandles.get(pastCheckpointId);
- try (FSDataInputStream in = handle.f1.openInputStream()) {
+ if (!committer.isCheckpointCommitted(subtaskId, pastCheckpointId)) {
+ try (FSDataInputStream in = streamHandle.openInputStream()) {
boolean success = sendValues(
new ReusingMutableToRegularIteratorWrapper<>(
new InputViewIterator<>(
new DataInputViewStreamWrapper(
in),
serializer),
serializer),
- handle.f0);
- if (success) { //if the sending has failed we will retry on the next notify
- committer.commitCheckpoint(pastCheckpointId);
- checkpointsToRemove.add(pastCheckpointId);
+ timestamp);
+ if (success) {
+ // in case the checkpoint was successfully committed, discard its state from the
+ // backend and mark it for removal
+ // in case it failed, we retry on the next checkpoint
+
--- End diff --
remove empty line
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2707: [FLINK-4939] GenericWriteAheadSink: Decouple the c...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/2707#discussion_r86108131
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericWriteAheadSink.java ---
@@ -77,63 +86,96 @@ public GenericWriteAheadSink(CheckpointCommitter committer, TypeSerializer<IN> s
public void open() throws Exception {
super.open();
committer.setOperatorId(id);
- committer.setOperatorSubtaskId(getRuntimeContext().getIndexOfThisSubtask());
committer.open();
- cleanState();
- checkpointStreamFactory =
- getContainingTask().createCheckpointStreamFactory(this);
+
+ checkpointStreamFactory = getContainingTask()
+ .createCheckpointStreamFactory(this);
+
+ cleanRestoredHandles();
}
public void close() throws Exception {
committer.close();
}
/**
- * Saves a handle in the state.
+ * Called when a checkpoint barrier arrives.
+ * Closes any open streams to the backend and marks them as pending for
+ * committing to the final output system, e.g. Cassandra.
*
- * @param checkpointId
- * @throws IOException
+ * @param checkpointId the id of the latest received checkpoint.
+ * @throws IOException in case something went wrong when handling the stream to the backend.
*/
private void saveHandleInState(final long checkpointId, final long timestamp) throws Exception {
+ Preconditions.checkNotNull(this.pendingHandles, "The operator has not been properly initialized.");
--- End diff --
this seems unnecessary.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2707: [FLINK-4939] GenericWriteAheadSink: Decouple the c...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/2707#discussion_r86109229
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericWriteAheadSink.java ---
@@ -142,46 +184,62 @@ private void cleanState() throws Exception {
public void notifyOfCompletedCheckpoint(long checkpointId) throws Exception {
super.notifyOfCompletedCheckpoint(checkpointId);
- synchronized (state.pendingHandles) {
- Set<Long> pastCheckpointIds = state.pendingHandles.keySet();
- Set<Long> checkpointsToRemove = new HashSet<>();
- for (Long pastCheckpointId : pastCheckpointIds) {
+ synchronized (pendingHandles) {
+ Set<PendingCheckpointId> checkpointsToRemove = new HashSet<>();
+
+ for (Map.Entry<PendingCheckpointId, PendingHandle> pendingHandle : pendingHandles.entrySet()) {
+
+ PendingCheckpointId pendingId = pendingHandle.getKey();
+ long pastCheckpointId = pendingId.checkpointId;
+ int subtaskId = pendingId.subtaskId;
+
if (pastCheckpointId <= checkpointId) {
+
--- End diff --
remove empty line
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2707: [FLINK-4939] GenericWriteAheadSink: Decouple the c...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/2707#discussion_r86561350
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericWriteAheadSink.java ---
@@ -77,111 +84,144 @@ public GenericWriteAheadSink(CheckpointCommitter committer, TypeSerializer<IN> s
public void open() throws Exception {
super.open();
committer.setOperatorId(id);
- committer.setOperatorSubtaskId(getRuntimeContext().getIndexOfThisSubtask());
committer.open();
- cleanState();
- checkpointStreamFactory =
- getContainingTask().createCheckpointStreamFactory(this);
+
+ checkpointStreamFactory = getContainingTask()
+ .createCheckpointStreamFactory(this);
+
+ cleanRestoredHandles();
}
public void close() throws Exception {
committer.close();
}
/**
- * Saves a handle in the state.
+ * Called when a checkpoint barrier arrives. It closes any open streams to the backend
+ * and marks them as pending for committing to the external, third-party storage system.
*
- * @param checkpointId
- * @throws IOException
+ * @param checkpointId the id of the latest received checkpoint.
+ * @throws IOException in case something went wrong when handling the stream to the backend.
*/
private void saveHandleInState(final long checkpointId, final long timestamp) throws Exception {
+ Preconditions.checkNotNull(this.pendingCheckpoints, "The operator has not been properly initialized.");
+
//only add handle if a new OperatorState was created since the last snapshot
if (out != null) {
+ int subtaskIdx = getRuntimeContext().getIndexOfThisSubtask();
StreamStateHandle handle = out.closeAndGetHandle();
- if (state.pendingHandles.containsKey(checkpointId)) {
+
+ PendingCheckpoint pendingCheckpoint = new PendingCheckpoint(
+ checkpointId, subtaskIdx, timestamp, handle);
+
+ if (pendingCheckpoints.contains(pendingCheckpoint)) {
//we already have a checkpoint stored for that ID that may have been partially written,
//so we discard this "alternate version" and use the stored checkpoint
handle.discardState();
} else {
- state.pendingHandles.put(checkpointId, new Tuple2<>(timestamp, handle));
+ pendingCheckpoints.add(pendingCheckpoint);
}
out = null;
}
}
@Override
- public void snapshotState(FSDataOutputStream out,
- long checkpointId,
- long timestamp) throws Exception {
+ public void snapshotState(FSDataOutputStream out, long checkpointId, long timestamp) throws Exception {
saveHandleInState(checkpointId, timestamp);
- InstantiationUtil.serializeObject(out, state);
+ DataOutputViewStreamWrapper outStream = new DataOutputViewStreamWrapper(out);
+ outStream.writeInt(pendingCheckpoints.size());
+ for (PendingCheckpoint pendingCheckpoint : pendingCheckpoints) {
+ pendingCheckpoint.serialize(outStream);
+ }
}
@Override
public void restoreState(FSDataInputStream in) throws Exception {
- this.state = InstantiationUtil.deserializeObject(in, getUserCodeClassloader());
+ final DataInputViewStreamWrapper inStream = new DataInputViewStreamWrapper(in);
+ int numPendingHandles = inStream.readInt();
+ for (int i = 0; i < numPendingHandles; i++) {
+ pendingCheckpoints.add(PendingCheckpoint.restore(inStream, getUserCodeClassloader()));
+ }
}
- private void cleanState() throws Exception {
- synchronized (this.state.pendingHandles) { //remove all handles that were already committed
- Set<Long> pastCheckpointIds = this.state.pendingHandles.keySet();
- Set<Long> checkpointsToRemove = new HashSet<>();
- for (Long pastCheckpointId : pastCheckpointIds) {
- if (committer.isCheckpointCommitted(pastCheckpointId)) {
- checkpointsToRemove.add(pastCheckpointId);
+ /**
+ * Called at {@link #open()} to clean-up the pending handle list.
+ * It iterates over all restored pending handles, checks which ones are already
+ * committed to the outside storage system and removes them from the list.
+ */
+ private void cleanRestoredHandles() throws Exception {
+ synchronized (pendingCheckpoints) {
+
+ // for each of the pending handles...
+ Iterator<PendingCheckpoint> pendingCheckpointIt = pendingCheckpoints.iterator();
+ while (pendingCheckpointIt.hasNext()) {
+
+ PendingCheckpoint pendingCheckpoint = pendingCheckpointIt.next();
+ long checkpointId = pendingCheckpoint.checkpointId;
+ int subtaskId = pendingCheckpoint.subtaskId;
+
+ //...check if the temporary buffer is already committed and if yes,
+ // remove it from the list of pending checkpoints.
+ if (committer.isCheckpointCommitted(subtaskId, checkpointId)) {
+ pendingCheckpoint.stateHandle.discardState();
+ pendingCheckpointIt.remove();
--- End diff --
No "The iterators returned by this class's iterator method are fail-fast: if the set is modified at any time after the iterator is created, in any way **except through the iterator's own remove method**" from the same javadocs.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2707: [FLINK-4939] GenericWriteAheadSink: Decouple the c...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/2707#discussion_r86109164
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericWriteAheadSink.java ---
@@ -142,46 +184,62 @@ private void cleanState() throws Exception {
public void notifyOfCompletedCheckpoint(long checkpointId) throws Exception {
super.notifyOfCompletedCheckpoint(checkpointId);
- synchronized (state.pendingHandles) {
- Set<Long> pastCheckpointIds = state.pendingHandles.keySet();
- Set<Long> checkpointsToRemove = new HashSet<>();
- for (Long pastCheckpointId : pastCheckpointIds) {
+ synchronized (pendingHandles) {
+ Set<PendingCheckpointId> checkpointsToRemove = new HashSet<>();
--- End diff --
same as above, use a `List` instead
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2707: [FLINK-4939] GenericWriteAheadSink: Decouple the c...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/2707#discussion_r86110695
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericWriteAheadSink.java ---
@@ -201,27 +259,83 @@ public void processElement(StreamRecord<IN> element) throws Exception {
serializer.serialize(value, new DataOutputViewStreamWrapper(out));
}
- /**
- * This state is used to keep a list of all StateHandles (essentially references to past OperatorStates) that were
- * used since the last completed checkpoint.
- **/
- public static class ExactlyOnceState implements Serializable {
+ private static final class PendingCheckpointId implements Comparable<PendingCheckpointId>, Serializable {
+
+ private static final long serialVersionUID = -3571036395734603443L;
+
+ private final long checkpointId;
+ private final int subtaskId;
+
+ PendingCheckpointId(long checkpointId, int subtaskId) {
+ this.checkpointId = checkpointId;
+ this.subtaskId = subtaskId;
+ }
+
+ void serialize(DataOutputViewStreamWrapper outputStream) throws IOException {
+ outputStream.writeLong(this.checkpointId);
+ outputStream.writeInt(this.subtaskId);
+ }
- private static final long serialVersionUID = -3571063495273460743L;
+ static PendingCheckpointId restore(DataInputViewStreamWrapper inputStream) throws IOException, ClassNotFoundException {
+ long checkpointId = inputStream.readLong();
+ int subtaskId = inputStream.readInt();
+ return new PendingCheckpointId(checkpointId, subtaskId);
+ }
+
+ @Override
+ public int compareTo(PendingCheckpointId o) {
+ if (o == null) {
+ return -1;
+ }
- protected TreeMap<Long, Tuple2<Long, StreamStateHandle>> pendingHandles;
+ long res = this.checkpointId - o.checkpointId;
--- End diff --
you could use `Long#compare(long, long)` instead of all this.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2707: [FLINK-4939] GenericWriteAheadSink: Decouple the c...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/2707#discussion_r86557000
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericWriteAheadSink.java ---
@@ -77,111 +84,144 @@ public GenericWriteAheadSink(CheckpointCommitter committer, TypeSerializer<IN> s
public void open() throws Exception {
super.open();
committer.setOperatorId(id);
- committer.setOperatorSubtaskId(getRuntimeContext().getIndexOfThisSubtask());
committer.open();
- cleanState();
- checkpointStreamFactory =
- getContainingTask().createCheckpointStreamFactory(this);
+
+ checkpointStreamFactory = getContainingTask()
+ .createCheckpointStreamFactory(this);
+
+ cleanRestoredHandles();
}
public void close() throws Exception {
committer.close();
}
/**
- * Saves a handle in the state.
+ * Called when a checkpoint barrier arrives. It closes any open streams to the backend
+ * and marks them as pending for committing to the external, third-party storage system.
*
- * @param checkpointId
- * @throws IOException
+ * @param checkpointId the id of the latest received checkpoint.
+ * @throws IOException in case something went wrong when handling the stream to the backend.
*/
private void saveHandleInState(final long checkpointId, final long timestamp) throws Exception {
+ Preconditions.checkNotNull(this.pendingCheckpoints, "The operator has not been properly initialized.");
--- End diff --
This can't happen and is thus unnecessary.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2707: [FLINK-4939] GenericWriteAheadSink: Decouple the c...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/2707#discussion_r86104979
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/CheckpointCommitter.java ---
@@ -98,17 +89,19 @@ public void setOperatorSubtaskId(int id) throws Exception {
/**
* Mark the given checkpoint as completed in the resource.
*
- * @param checkpointID
+ * @param subtaskIdx the index of the subtask responsible for committing the checkpoint.
+ * @param checkpointID this id of the checkpoint to be committed.
--- End diff --
this id -> the id
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2707: [FLINK-4939] GenericWriteAheadSink: Decouple the creating...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on the issue:
https://github.com/apache/flink/pull/2707
@zentol let me know if you have any more comments.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2707: [FLINK-4939] GenericWriteAheadSink: Decouple the c...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/2707#discussion_r86106408
--- Diff: flink-streaming-connectors/flink-connector-cassandra/src/main/java/org/apache/flink/streaming/connectors/cassandra/CassandraCommitter.java ---
@@ -118,16 +117,32 @@ public void close() throws Exception {
}
@Override
- public void commitCheckpoint(long checkpointID) {
- session.execute(updateStatement.bind(checkpointID));
- this.lastCommittedCheckpointID = checkpointID;
+ public void commitCheckpoint(int subtaskIdx, long checkpointId) {
+ String statement = String.format(
+ "UPDATE %s.%s set checkpoint_id=%d where sink_id='%s' and sub_id=%d;",
+ keySpace, table, checkpointId, operatorId, subtaskIdx);
+
+ session.execute(statement);
--- End diff --
You forgot to update the `lastCommittedCheckpointID` here. It will now only be updated once in `isCheckpointCommitted`.
As a follow up the test for this behavior should be extended to cover more than 1 commit.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2707: [FLINK-4939] GenericWriteAheadSink: Decouple the c...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/2707#discussion_r86563313
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericWriteAheadSink.java ---
@@ -77,111 +84,144 @@ public GenericWriteAheadSink(CheckpointCommitter committer, TypeSerializer<IN> s
public void open() throws Exception {
super.open();
committer.setOperatorId(id);
- committer.setOperatorSubtaskId(getRuntimeContext().getIndexOfThisSubtask());
committer.open();
- cleanState();
- checkpointStreamFactory =
- getContainingTask().createCheckpointStreamFactory(this);
+
+ checkpointStreamFactory = getContainingTask()
+ .createCheckpointStreamFactory(this);
+
+ cleanRestoredHandles();
}
public void close() throws Exception {
committer.close();
}
/**
- * Saves a handle in the state.
+ * Called when a checkpoint barrier arrives. It closes any open streams to the backend
+ * and marks them as pending for committing to the external, third-party storage system.
*
- * @param checkpointId
- * @throws IOException
+ * @param checkpointId the id of the latest received checkpoint.
+ * @throws IOException in case something went wrong when handling the stream to the backend.
*/
private void saveHandleInState(final long checkpointId, final long timestamp) throws Exception {
+ Preconditions.checkNotNull(this.pendingCheckpoints, "The operator has not been properly initialized.");
+
//only add handle if a new OperatorState was created since the last snapshot
if (out != null) {
+ int subtaskIdx = getRuntimeContext().getIndexOfThisSubtask();
StreamStateHandle handle = out.closeAndGetHandle();
- if (state.pendingHandles.containsKey(checkpointId)) {
+
+ PendingCheckpoint pendingCheckpoint = new PendingCheckpoint(
+ checkpointId, subtaskIdx, timestamp, handle);
+
+ if (pendingCheckpoints.contains(pendingCheckpoint)) {
//we already have a checkpoint stored for that ID that may have been partially written,
//so we discard this "alternate version" and use the stored checkpoint
handle.discardState();
} else {
- state.pendingHandles.put(checkpointId, new Tuple2<>(timestamp, handle));
+ pendingCheckpoints.add(pendingCheckpoint);
}
out = null;
}
}
@Override
- public void snapshotState(FSDataOutputStream out,
- long checkpointId,
- long timestamp) throws Exception {
+ public void snapshotState(FSDataOutputStream out, long checkpointId, long timestamp) throws Exception {
saveHandleInState(checkpointId, timestamp);
- InstantiationUtil.serializeObject(out, state);
+ DataOutputViewStreamWrapper outStream = new DataOutputViewStreamWrapper(out);
+ outStream.writeInt(pendingCheckpoints.size());
+ for (PendingCheckpoint pendingCheckpoint : pendingCheckpoints) {
+ pendingCheckpoint.serialize(outStream);
+ }
}
@Override
public void restoreState(FSDataInputStream in) throws Exception {
- this.state = InstantiationUtil.deserializeObject(in, getUserCodeClassloader());
+ final DataInputViewStreamWrapper inStream = new DataInputViewStreamWrapper(in);
+ int numPendingHandles = inStream.readInt();
+ for (int i = 0; i < numPendingHandles; i++) {
+ pendingCheckpoints.add(PendingCheckpoint.restore(inStream, getUserCodeClassloader()));
+ }
}
- private void cleanState() throws Exception {
- synchronized (this.state.pendingHandles) { //remove all handles that were already committed
- Set<Long> pastCheckpointIds = this.state.pendingHandles.keySet();
- Set<Long> checkpointsToRemove = new HashSet<>();
- for (Long pastCheckpointId : pastCheckpointIds) {
- if (committer.isCheckpointCommitted(pastCheckpointId)) {
- checkpointsToRemove.add(pastCheckpointId);
+ /**
+ * Called at {@link #open()} to clean-up the pending handle list.
+ * It iterates over all restored pending handles, checks which ones are already
+ * committed to the outside storage system and removes them from the list.
+ */
+ private void cleanRestoredHandles() throws Exception {
+ synchronized (pendingCheckpoints) {
+
+ // for each of the pending handles...
+ Iterator<PendingCheckpoint> pendingCheckpointIt = pendingCheckpoints.iterator();
+ while (pendingCheckpointIt.hasNext()) {
+
+ PendingCheckpoint pendingCheckpoint = pendingCheckpointIt.next();
+ long checkpointId = pendingCheckpoint.checkpointId;
+ int subtaskId = pendingCheckpoint.subtaskId;
+
+ //...check if the temporary buffer is already committed and if yes,
+ // remove it from the list of pending checkpoints.
+ if (committer.isCheckpointCommitted(subtaskId, checkpointId)) {
+ pendingCheckpoint.stateHandle.discardState();
+ pendingCheckpointIt.remove();
--- End diff --
ah, my bad :)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2707: [FLINK-4939] GenericWriteAheadSink: Decouple the creating...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on the issue:
https://github.com/apache/flink/pull/2707
Hi @zentol thanks for the review. I suppose that the comment that it does not work, implies that it does not work when rescaling right? Not in general. If I am right, then this is expected. As you can see I have not even changed the implementation of the snapshot/restore cycle, as we first have to figure out how to solve more important design issues of the sink. This is the reason why I also mention in the jira that this is just a first step.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2707: [FLINK-4939] GenericWriteAheadSink: Decouple the c...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/2707#discussion_r86109012
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericWriteAheadSink.java ---
@@ -77,63 +86,96 @@ public GenericWriteAheadSink(CheckpointCommitter committer, TypeSerializer<IN> s
public void open() throws Exception {
super.open();
committer.setOperatorId(id);
- committer.setOperatorSubtaskId(getRuntimeContext().getIndexOfThisSubtask());
committer.open();
- cleanState();
- checkpointStreamFactory =
- getContainingTask().createCheckpointStreamFactory(this);
+
+ checkpointStreamFactory = getContainingTask()
+ .createCheckpointStreamFactory(this);
+
+ cleanRestoredHandles();
}
public void close() throws Exception {
committer.close();
}
/**
- * Saves a handle in the state.
+ * Called when a checkpoint barrier arrives.
+ * Closes any open streams to the backend and marks them as pending for
+ * committing to the final output system, e.g. Cassandra.
*
- * @param checkpointId
- * @throws IOException
+ * @param checkpointId the id of the latest received checkpoint.
+ * @throws IOException in case something went wrong when handling the stream to the backend.
*/
private void saveHandleInState(final long checkpointId, final long timestamp) throws Exception {
+ Preconditions.checkNotNull(this.pendingHandles, "The operator has not been properly initialized.");
+
//only add handle if a new OperatorState was created since the last snapshot
if (out != null) {
StreamStateHandle handle = out.closeAndGetHandle();
- if (state.pendingHandles.containsKey(checkpointId)) {
+
+ PendingCheckpointId pendingCheckpoint = new PendingCheckpointId(
+ checkpointId, getRuntimeContext().getIndexOfThisSubtask());
+
+ if (pendingHandles.containsKey(pendingCheckpoint)) {
//we already have a checkpoint stored for that ID that may have been partially written,
//so we discard this "alternate version" and use the stored checkpoint
handle.discardState();
} else {
- state.pendingHandles.put(checkpointId, new Tuple2<>(timestamp, handle));
+ this.pendingHandles.put(pendingCheckpoint, new PendingHandle(timestamp, handle));
}
out = null;
}
}
@Override
- public void snapshotState(FSDataOutputStream out,
- long checkpointId,
- long timestamp) throws Exception {
+ public void snapshotState(FSDataOutputStream out, long checkpointId, long timestamp) throws Exception {
saveHandleInState(checkpointId, timestamp);
- InstantiationUtil.serializeObject(out, state);
+ DataOutputViewStreamWrapper outStream = new DataOutputViewStreamWrapper(out);
+ outStream.writeInt(pendingHandles.size());
+ for (Map.Entry<PendingCheckpointId, PendingHandle> pendingCheckpoint : pendingHandles.entrySet()) {
+ pendingCheckpoint.getKey().serialize(outStream);
+ pendingCheckpoint.getValue().serialize(outStream);
+ }
}
@Override
public void restoreState(FSDataInputStream in) throws Exception {
- this.state = InstantiationUtil.deserializeObject(in, getUserCodeClassloader());
+ final DataInputViewStreamWrapper inStream = new DataInputViewStreamWrapper(in);
+ int noOfPendingHandles = inStream.readInt();
+ for (int i = 0; i < noOfPendingHandles; i++) {
+ PendingCheckpointId id = PendingCheckpointId.restore(inStream);
+ PendingHandle handle = PendingHandle.restore(inStream, getUserCodeClassloader());
+ this.pendingHandles.put(id, handle);
+ }
}
- private void cleanState() throws Exception {
- synchronized (this.state.pendingHandles) { //remove all handles that were already committed
- Set<Long> pastCheckpointIds = this.state.pendingHandles.keySet();
- Set<Long> checkpointsToRemove = new HashSet<>();
- for (Long pastCheckpointId : pastCheckpointIds) {
- if (committer.isCheckpointCommitted(pastCheckpointId)) {
- checkpointsToRemove.add(pastCheckpointId);
+ /**
+ * Called at {@link #open()} to clean-up the pending handle list.
+ * It iterates over all restored pending handles, checks which ones are already
+ * committed to the outside storage system and removes them from the list.
+ */
+ private void cleanRestoredHandles() throws Exception {
+ synchronized (pendingHandles) {
+
+ Set<PendingCheckpointId> checkpointIdsToRemove = new HashSet<>();
+
+ // for each of the pending handles...
--- End diff --
this comments here don't really add value; the javadoc explains everything already.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2707: [FLINK-4939] GenericWriteAheadSink: Decouple the c...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/2707#discussion_r86109187
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericWriteAheadSink.java ---
@@ -142,46 +184,62 @@ private void cleanState() throws Exception {
public void notifyOfCompletedCheckpoint(long checkpointId) throws Exception {
super.notifyOfCompletedCheckpoint(checkpointId);
- synchronized (state.pendingHandles) {
- Set<Long> pastCheckpointIds = state.pendingHandles.keySet();
- Set<Long> checkpointsToRemove = new HashSet<>();
- for (Long pastCheckpointId : pastCheckpointIds) {
+ synchronized (pendingHandles) {
+ Set<PendingCheckpointId> checkpointsToRemove = new HashSet<>();
+
+ for (Map.Entry<PendingCheckpointId, PendingHandle> pendingHandle : pendingHandles.entrySet()) {
+
--- End diff --
remove empty line
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2707: [FLINK-4939] GenericWriteAheadSink: Decouple the c...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/2707#discussion_r86107589
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericWriteAheadSink.java ---
@@ -27,27 +26,30 @@
import org.apache.flink.runtime.state.CheckpointStreamFactory;
import org.apache.flink.runtime.state.StreamStateHandle;
import org.apache.flink.runtime.util.ReusingMutableToRegularIteratorWrapper;
-import org.apache.flink.streaming.api.operators.StreamCheckpointedOperator;
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.operators.StreamCheckpointedOperator;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.util.InstantiationUtil;
+import org.apache.flink.util.Preconditions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.io.Serializable;
import java.util.HashSet;
+import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
import java.util.UUID;
/**
- * Generic Sink that emits its input elements into an arbitrary backend. This sink is integrated with the checkpointing
- * mechanism and can provide exactly-once guarantees; depending on the storage backend and sink/committer implementation.
+ * Generic Sink that emits its input elements into an arbitrary backend. This sink is integrated with
+ * Flink's checkpointing mechanism and can provide exactly-once guarantees; depending on the storage
--- End diff --
the javadocs don't exceed the line limit, please don't reformat them.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2707: [FLINK-4939] GenericWriteAheadSink: Decouple the c...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/2707#discussion_r86108910
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericWriteAheadSink.java ---
@@ -77,63 +86,96 @@ public GenericWriteAheadSink(CheckpointCommitter committer, TypeSerializer<IN> s
public void open() throws Exception {
super.open();
committer.setOperatorId(id);
- committer.setOperatorSubtaskId(getRuntimeContext().getIndexOfThisSubtask());
committer.open();
- cleanState();
- checkpointStreamFactory =
- getContainingTask().createCheckpointStreamFactory(this);
+
+ checkpointStreamFactory = getContainingTask()
+ .createCheckpointStreamFactory(this);
+
+ cleanRestoredHandles();
}
public void close() throws Exception {
committer.close();
}
/**
- * Saves a handle in the state.
+ * Called when a checkpoint barrier arrives.
+ * Closes any open streams to the backend and marks them as pending for
+ * committing to the final output system, e.g. Cassandra.
*
- * @param checkpointId
- * @throws IOException
+ * @param checkpointId the id of the latest received checkpoint.
+ * @throws IOException in case something went wrong when handling the stream to the backend.
*/
private void saveHandleInState(final long checkpointId, final long timestamp) throws Exception {
+ Preconditions.checkNotNull(this.pendingHandles, "The operator has not been properly initialized.");
+
//only add handle if a new OperatorState was created since the last snapshot
if (out != null) {
StreamStateHandle handle = out.closeAndGetHandle();
- if (state.pendingHandles.containsKey(checkpointId)) {
+
+ PendingCheckpointId pendingCheckpoint = new PendingCheckpointId(
+ checkpointId, getRuntimeContext().getIndexOfThisSubtask());
+
+ if (pendingHandles.containsKey(pendingCheckpoint)) {
//we already have a checkpoint stored for that ID that may have been partially written,
//so we discard this "alternate version" and use the stored checkpoint
handle.discardState();
} else {
- state.pendingHandles.put(checkpointId, new Tuple2<>(timestamp, handle));
+ this.pendingHandles.put(pendingCheckpoint, new PendingHandle(timestamp, handle));
}
out = null;
}
}
@Override
- public void snapshotState(FSDataOutputStream out,
- long checkpointId,
- long timestamp) throws Exception {
+ public void snapshotState(FSDataOutputStream out, long checkpointId, long timestamp) throws Exception {
saveHandleInState(checkpointId, timestamp);
- InstantiationUtil.serializeObject(out, state);
+ DataOutputViewStreamWrapper outStream = new DataOutputViewStreamWrapper(out);
+ outStream.writeInt(pendingHandles.size());
+ for (Map.Entry<PendingCheckpointId, PendingHandle> pendingCheckpoint : pendingHandles.entrySet()) {
+ pendingCheckpoint.getKey().serialize(outStream);
+ pendingCheckpoint.getValue().serialize(outStream);
+ }
}
@Override
public void restoreState(FSDataInputStream in) throws Exception {
- this.state = InstantiationUtil.deserializeObject(in, getUserCodeClassloader());
+ final DataInputViewStreamWrapper inStream = new DataInputViewStreamWrapper(in);
+ int noOfPendingHandles = inStream.readInt();
+ for (int i = 0; i < noOfPendingHandles; i++) {
+ PendingCheckpointId id = PendingCheckpointId.restore(inStream);
+ PendingHandle handle = PendingHandle.restore(inStream, getUserCodeClassloader());
+ this.pendingHandles.put(id, handle);
+ }
}
- private void cleanState() throws Exception {
- synchronized (this.state.pendingHandles) { //remove all handles that were already committed
- Set<Long> pastCheckpointIds = this.state.pendingHandles.keySet();
- Set<Long> checkpointsToRemove = new HashSet<>();
- for (Long pastCheckpointId : pastCheckpointIds) {
- if (committer.isCheckpointCommitted(pastCheckpointId)) {
- checkpointsToRemove.add(pastCheckpointId);
+ /**
+ * Called at {@link #open()} to clean-up the pending handle list.
+ * It iterates over all restored pending handles, checks which ones are already
+ * committed to the outside storage system and removes them from the list.
+ */
+ private void cleanRestoredHandles() throws Exception {
+ synchronized (pendingHandles) {
+
+ Set<PendingCheckpointId> checkpointIdsToRemove = new HashSet<>();
--- End diff --
a `List` would be more efficient; it is already guaranteed that the `PendingCheckpointId` is unique anyway.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2707: [FLINK-4939] GenericWriteAheadSink: Decouple the c...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/2707#discussion_r86561357
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericWriteAheadSink.java ---
@@ -77,111 +84,144 @@ public GenericWriteAheadSink(CheckpointCommitter committer, TypeSerializer<IN> s
public void open() throws Exception {
super.open();
committer.setOperatorId(id);
- committer.setOperatorSubtaskId(getRuntimeContext().getIndexOfThisSubtask());
committer.open();
- cleanState();
- checkpointStreamFactory =
- getContainingTask().createCheckpointStreamFactory(this);
+
+ checkpointStreamFactory = getContainingTask()
+ .createCheckpointStreamFactory(this);
+
+ cleanRestoredHandles();
}
public void close() throws Exception {
committer.close();
}
/**
- * Saves a handle in the state.
+ * Called when a checkpoint barrier arrives. It closes any open streams to the backend
+ * and marks them as pending for committing to the external, third-party storage system.
*
- * @param checkpointId
- * @throws IOException
+ * @param checkpointId the id of the latest received checkpoint.
+ * @throws IOException in case something went wrong when handling the stream to the backend.
*/
private void saveHandleInState(final long checkpointId, final long timestamp) throws Exception {
+ Preconditions.checkNotNull(this.pendingCheckpoints, "The operator has not been properly initialized.");
+
//only add handle if a new OperatorState was created since the last snapshot
if (out != null) {
+ int subtaskIdx = getRuntimeContext().getIndexOfThisSubtask();
StreamStateHandle handle = out.closeAndGetHandle();
- if (state.pendingHandles.containsKey(checkpointId)) {
+
+ PendingCheckpoint pendingCheckpoint = new PendingCheckpoint(
+ checkpointId, subtaskIdx, timestamp, handle);
+
+ if (pendingCheckpoints.contains(pendingCheckpoint)) {
//we already have a checkpoint stored for that ID that may have been partially written,
//so we discard this "alternate version" and use the stored checkpoint
handle.discardState();
} else {
- state.pendingHandles.put(checkpointId, new Tuple2<>(timestamp, handle));
+ pendingCheckpoints.add(pendingCheckpoint);
}
out = null;
}
}
@Override
- public void snapshotState(FSDataOutputStream out,
- long checkpointId,
- long timestamp) throws Exception {
+ public void snapshotState(FSDataOutputStream out, long checkpointId, long timestamp) throws Exception {
saveHandleInState(checkpointId, timestamp);
- InstantiationUtil.serializeObject(out, state);
+ DataOutputViewStreamWrapper outStream = new DataOutputViewStreamWrapper(out);
+ outStream.writeInt(pendingCheckpoints.size());
+ for (PendingCheckpoint pendingCheckpoint : pendingCheckpoints) {
+ pendingCheckpoint.serialize(outStream);
+ }
}
@Override
public void restoreState(FSDataInputStream in) throws Exception {
- this.state = InstantiationUtil.deserializeObject(in, getUserCodeClassloader());
+ final DataInputViewStreamWrapper inStream = new DataInputViewStreamWrapper(in);
+ int numPendingHandles = inStream.readInt();
+ for (int i = 0; i < numPendingHandles; i++) {
+ pendingCheckpoints.add(PendingCheckpoint.restore(inStream, getUserCodeClassloader()));
+ }
}
- private void cleanState() throws Exception {
- synchronized (this.state.pendingHandles) { //remove all handles that were already committed
- Set<Long> pastCheckpointIds = this.state.pendingHandles.keySet();
- Set<Long> checkpointsToRemove = new HashSet<>();
- for (Long pastCheckpointId : pastCheckpointIds) {
- if (committer.isCheckpointCommitted(pastCheckpointId)) {
- checkpointsToRemove.add(pastCheckpointId);
+ /**
+ * Called at {@link #open()} to clean-up the pending handle list.
+ * It iterates over all restored pending handles, checks which ones are already
+ * committed to the outside storage system and removes them from the list.
+ */
+ private void cleanRestoredHandles() throws Exception {
+ synchronized (pendingCheckpoints) {
+
+ // for each of the pending handles...
+ Iterator<PendingCheckpoint> pendingCheckpointIt = pendingCheckpoints.iterator();
+ while (pendingCheckpointIt.hasNext()) {
+
+ PendingCheckpoint pendingCheckpoint = pendingCheckpointIt.next();
+ long checkpointId = pendingCheckpoint.checkpointId;
+ int subtaskId = pendingCheckpoint.subtaskId;
+
+ //...check if the temporary buffer is already committed and if yes,
+ // remove it from the list of pending checkpoints.
+ if (committer.isCheckpointCommitted(subtaskId, checkpointId)) {
+ pendingCheckpoint.stateHandle.discardState();
+ pendingCheckpointIt.remove();
}
}
- for (Long toRemove : checkpointsToRemove) {
- this.state.pendingHandles.remove(toRemove);
- }
}
}
@Override
public void notifyOfCompletedCheckpoint(long checkpointId) throws Exception {
super.notifyOfCompletedCheckpoint(checkpointId);
- synchronized (state.pendingHandles) {
- Set<Long> pastCheckpointIds = state.pendingHandles.keySet();
- Set<Long> checkpointsToRemove = new HashSet<>();
- for (Long pastCheckpointId : pastCheckpointIds) {
+ synchronized (pendingCheckpoints) {
+ Iterator<PendingCheckpoint> pendingCheckpointIt = pendingCheckpoints.iterator();
+ while (pendingCheckpointIt.hasNext()) {
+
+ PendingCheckpoint pendingCheckpoint = pendingCheckpointIt.next();
+ long pastCheckpointId = pendingCheckpoint.checkpointId;
+ int subtaskId = pendingCheckpoint.subtaskId;
+ long timestamp = pendingCheckpoint.timestamp;
+ StreamStateHandle streamHandle = pendingCheckpoint.stateHandle;
+
if (pastCheckpointId <= checkpointId) {
try {
- if (!committer.isCheckpointCommitted(pastCheckpointId)) {
- Tuple2<Long, StreamStateHandle> handle = state.pendingHandles.get(pastCheckpointId);
- try (FSDataInputStream in = handle.f1.openInputStream()) {
+ if (!committer.isCheckpointCommitted(subtaskId, pastCheckpointId)) {
+ try (FSDataInputStream in = streamHandle.openInputStream()) {
boolean success = sendValues(
new ReusingMutableToRegularIteratorWrapper<>(
new InputViewIterator<>(
new DataInputViewStreamWrapper(
in),
serializer),
serializer),
- handle.f0);
- if (success) { //if the sending has failed we will retry on the next notify
- committer.commitCheckpoint(pastCheckpointId);
- checkpointsToRemove.add(pastCheckpointId);
+ timestamp);
+ if (success) {
+
+ // in case the checkpoint was successfully committed,
+ // discard its state from the backend and mark it for removal
+ // in case it failed, we retry on the next checkpoint
+ committer.commitCheckpoint(subtaskId, pastCheckpointId);
+ streamHandle.discardState();
+ pendingCheckpointIt.remove();
--- End diff --
No "The iterators returned by this class's iterator method are fail-fast: if the set is modified at any time after the iterator is created, in any way **except through the iterator's own remove method**" from the same javadocs.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2707: [FLINK-4939] GenericWriteAheadSink: Decouple the c...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/2707#discussion_r86110136
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericWriteAheadSink.java ---
@@ -57,18 +59,25 @@
private static final long serialVersionUID = 1L;
protected static final Logger LOG = LoggerFactory.getLogger(GenericWriteAheadSink.class);
+
+ private final String id;
private final CheckpointCommitter committer;
- private transient CheckpointStreamFactory.CheckpointStateOutputStream out;
protected final TypeSerializer<IN> serializer;
- private final String id;
+
+ private transient CheckpointStreamFactory.CheckpointStateOutputStream out;
private transient CheckpointStreamFactory checkpointStreamFactory;
- private ExactlyOnceState state = new ExactlyOnceState();
+ private final TreeMap<PendingCheckpointId, PendingHandle> pendingHandles = new TreeMap<>();
--- End diff --
the `PendingCheckpointId` and `PendingHandle` can be consolidated into a single object, since they are always used in conjunction, and stored in a TreeSet.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2707: [FLINK-4939] GenericWriteAheadSink: Decouple the c...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/2707#discussion_r86111394
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericWriteAheadSink.java ---
@@ -77,63 +86,96 @@ public GenericWriteAheadSink(CheckpointCommitter committer, TypeSerializer<IN> s
public void open() throws Exception {
super.open();
committer.setOperatorId(id);
- committer.setOperatorSubtaskId(getRuntimeContext().getIndexOfThisSubtask());
committer.open();
- cleanState();
- checkpointStreamFactory =
- getContainingTask().createCheckpointStreamFactory(this);
+
+ checkpointStreamFactory = getContainingTask()
+ .createCheckpointStreamFactory(this);
+
+ cleanRestoredHandles();
}
public void close() throws Exception {
committer.close();
}
/**
- * Saves a handle in the state.
+ * Called when a checkpoint barrier arrives.
+ * Closes any open streams to the backend and marks them as pending for
+ * committing to the final output system, e.g. Cassandra.
--- End diff --
to the final output system -> to the external backend; remove mentioning of cassandra.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2707: [FLINK-4939] GenericWriteAheadSink: Decouple the c...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/2707#discussion_r86107282
--- Diff: flink-streaming-connectors/flink-connector-cassandra/src/main/java/org/apache/flink/streaming/connectors/cassandra/CassandraCommitter.java ---
@@ -95,16 +100,10 @@ public void open() throws Exception {
}
cluster = builder.getCluster();
session = cluster.connect();
-
- updateStatement = session.prepare(String.format("UPDATE %s.%s set checkpoint_id=? where sink_id='%s' and sub_id=%d;", keySpace, table, operatorId, subtaskId));
- selectStatement = session.prepare(String.format("SELECT checkpoint_id FROM %s.%s where sink_id='%s' and sub_id=%d;", keySpace, table, operatorId, subtaskId));
-
- session.execute(String.format("INSERT INTO %s.%s (sink_id, sub_id, checkpoint_id) values ('%s', %d, " + -1 + ") IF NOT EXISTS;", keySpace, table, operatorId, subtaskId));
}
@Override
public void close() throws Exception {
- this.lastCommittedCheckpointID = -1;
--- End diff --
where is the replacement for this line? (This is useful for testing)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2707: [FLINK-4939] GenericWriteAheadSink: Decouple the creating...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on the issue:
https://github.com/apache/flink/pull/2707
no, I'm not talking about rescaling,
```
cc.commitCheckpoint(0, 1);
assertTrue(cc.isCheckpointCommitted(0, 1); // works
cc.commitCheckpoint(0, 2);
assertTrue(cc.isCheckpointCommitted(0, 2); // will fail
```
This may not actually break anything, but i find it dangerous to change this behavior.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2707: [FLINK-4939] GenericWriteAheadSink: Decouple the c...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/2707#discussion_r86557736
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericWriteAheadSink.java ---
@@ -77,111 +84,144 @@ public GenericWriteAheadSink(CheckpointCommitter committer, TypeSerializer<IN> s
public void open() throws Exception {
super.open();
committer.setOperatorId(id);
- committer.setOperatorSubtaskId(getRuntimeContext().getIndexOfThisSubtask());
committer.open();
- cleanState();
- checkpointStreamFactory =
- getContainingTask().createCheckpointStreamFactory(this);
+
+ checkpointStreamFactory = getContainingTask()
+ .createCheckpointStreamFactory(this);
+
+ cleanRestoredHandles();
}
public void close() throws Exception {
committer.close();
}
/**
- * Saves a handle in the state.
+ * Called when a checkpoint barrier arrives. It closes any open streams to the backend
+ * and marks them as pending for committing to the external, third-party storage system.
*
- * @param checkpointId
- * @throws IOException
+ * @param checkpointId the id of the latest received checkpoint.
+ * @throws IOException in case something went wrong when handling the stream to the backend.
*/
private void saveHandleInState(final long checkpointId, final long timestamp) throws Exception {
+ Preconditions.checkNotNull(this.pendingCheckpoints, "The operator has not been properly initialized.");
+
//only add handle if a new OperatorState was created since the last snapshot
if (out != null) {
+ int subtaskIdx = getRuntimeContext().getIndexOfThisSubtask();
StreamStateHandle handle = out.closeAndGetHandle();
- if (state.pendingHandles.containsKey(checkpointId)) {
+
+ PendingCheckpoint pendingCheckpoint = new PendingCheckpoint(
+ checkpointId, subtaskIdx, timestamp, handle);
+
+ if (pendingCheckpoints.contains(pendingCheckpoint)) {
//we already have a checkpoint stored for that ID that may have been partially written,
//so we discard this "alternate version" and use the stored checkpoint
handle.discardState();
} else {
- state.pendingHandles.put(checkpointId, new Tuple2<>(timestamp, handle));
+ pendingCheckpoints.add(pendingCheckpoint);
}
out = null;
}
}
@Override
- public void snapshotState(FSDataOutputStream out,
- long checkpointId,
- long timestamp) throws Exception {
+ public void snapshotState(FSDataOutputStream out, long checkpointId, long timestamp) throws Exception {
saveHandleInState(checkpointId, timestamp);
- InstantiationUtil.serializeObject(out, state);
+ DataOutputViewStreamWrapper outStream = new DataOutputViewStreamWrapper(out);
+ outStream.writeInt(pendingCheckpoints.size());
+ for (PendingCheckpoint pendingCheckpoint : pendingCheckpoints) {
+ pendingCheckpoint.serialize(outStream);
+ }
}
@Override
public void restoreState(FSDataInputStream in) throws Exception {
- this.state = InstantiationUtil.deserializeObject(in, getUserCodeClassloader());
+ final DataInputViewStreamWrapper inStream = new DataInputViewStreamWrapper(in);
+ int numPendingHandles = inStream.readInt();
+ for (int i = 0; i < numPendingHandles; i++) {
+ pendingCheckpoints.add(PendingCheckpoint.restore(inStream, getUserCodeClassloader()));
+ }
}
- private void cleanState() throws Exception {
- synchronized (this.state.pendingHandles) { //remove all handles that were already committed
- Set<Long> pastCheckpointIds = this.state.pendingHandles.keySet();
- Set<Long> checkpointsToRemove = new HashSet<>();
- for (Long pastCheckpointId : pastCheckpointIds) {
- if (committer.isCheckpointCommitted(pastCheckpointId)) {
- checkpointsToRemove.add(pastCheckpointId);
+ /**
+ * Called at {@link #open()} to clean-up the pending handle list.
+ * It iterates over all restored pending handles, checks which ones are already
+ * committed to the outside storage system and removes them from the list.
+ */
+ private void cleanRestoredHandles() throws Exception {
+ synchronized (pendingCheckpoints) {
+
+ // for each of the pending handles...
+ Iterator<PendingCheckpoint> pendingCheckpointIt = pendingCheckpoints.iterator();
+ while (pendingCheckpointIt.hasNext()) {
+
+ PendingCheckpoint pendingCheckpoint = pendingCheckpointIt.next();
+ long checkpointId = pendingCheckpoint.checkpointId;
+ int subtaskId = pendingCheckpoint.subtaskId;
+
+ //...check if the temporary buffer is already committed and if yes,
+ // remove it from the list of pending checkpoints.
+ if (committer.isCheckpointCommitted(subtaskId, checkpointId)) {
+ pendingCheckpoint.stateHandle.discardState();
+ pendingCheckpointIt.remove();
}
}
- for (Long toRemove : checkpointsToRemove) {
- this.state.pendingHandles.remove(toRemove);
- }
}
}
@Override
public void notifyOfCompletedCheckpoint(long checkpointId) throws Exception {
super.notifyOfCompletedCheckpoint(checkpointId);
- synchronized (state.pendingHandles) {
- Set<Long> pastCheckpointIds = state.pendingHandles.keySet();
- Set<Long> checkpointsToRemove = new HashSet<>();
- for (Long pastCheckpointId : pastCheckpointIds) {
+ synchronized (pendingCheckpoints) {
+ Iterator<PendingCheckpoint> pendingCheckpointIt = pendingCheckpoints.iterator();
+ while (pendingCheckpointIt.hasNext()) {
+
+ PendingCheckpoint pendingCheckpoint = pendingCheckpointIt.next();
+ long pastCheckpointId = pendingCheckpoint.checkpointId;
+ int subtaskId = pendingCheckpoint.subtaskId;
+ long timestamp = pendingCheckpoint.timestamp;
+ StreamStateHandle streamHandle = pendingCheckpoint.stateHandle;
+
if (pastCheckpointId <= checkpointId) {
try {
- if (!committer.isCheckpointCommitted(pastCheckpointId)) {
- Tuple2<Long, StreamStateHandle> handle = state.pendingHandles.get(pastCheckpointId);
- try (FSDataInputStream in = handle.f1.openInputStream()) {
+ if (!committer.isCheckpointCommitted(subtaskId, pastCheckpointId)) {
+ try (FSDataInputStream in = streamHandle.openInputStream()) {
boolean success = sendValues(
new ReusingMutableToRegularIteratorWrapper<>(
new InputViewIterator<>(
new DataInputViewStreamWrapper(
in),
serializer),
serializer),
- handle.f0);
- if (success) { //if the sending has failed we will retry on the next notify
- committer.commitCheckpoint(pastCheckpointId);
- checkpointsToRemove.add(pastCheckpointId);
+ timestamp);
+ if (success) {
+
+ // in case the checkpoint was successfully committed,
+ // discard its state from the backend and mark it for removal
+ // in case it failed, we retry on the next checkpoint
+ committer.commitCheckpoint(subtaskId, pastCheckpointId);
+ streamHandle.discardState();
+ pendingCheckpointIt.remove();
--- End diff --
same as above
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2707: [FLINK-4939] GenericWriteAheadSink: Decouple the c...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/2707#discussion_r86109608
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericWriteAheadSink.java ---
@@ -142,46 +184,62 @@ private void cleanState() throws Exception {
public void notifyOfCompletedCheckpoint(long checkpointId) throws Exception {
super.notifyOfCompletedCheckpoint(checkpointId);
- synchronized (state.pendingHandles) {
- Set<Long> pastCheckpointIds = state.pendingHandles.keySet();
- Set<Long> checkpointsToRemove = new HashSet<>();
- for (Long pastCheckpointId : pastCheckpointIds) {
+ synchronized (pendingHandles) {
+ Set<PendingCheckpointId> checkpointsToRemove = new HashSet<>();
+
+ for (Map.Entry<PendingCheckpointId, PendingHandle> pendingHandle : pendingHandles.entrySet()) {
+
+ PendingCheckpointId pendingId = pendingHandle.getKey();
+ long pastCheckpointId = pendingId.checkpointId;
+ int subtaskId = pendingId.subtaskId;
+
if (pastCheckpointId <= checkpointId) {
+
+ PendingHandle handle = pendingHandle.getValue();
+ long timestamp = handle.timestamp;
+ StreamStateHandle streamHandle = handle.stateHandle;
+
try {
- if (!committer.isCheckpointCommitted(pastCheckpointId)) {
- Tuple2<Long, StreamStateHandle> handle = state.pendingHandles.get(pastCheckpointId);
- try (FSDataInputStream in = handle.f1.openInputStream()) {
+ if (!committer.isCheckpointCommitted(subtaskId, pastCheckpointId)) {
+ try (FSDataInputStream in = streamHandle.openInputStream()) {
boolean success = sendValues(
new ReusingMutableToRegularIteratorWrapper<>(
new InputViewIterator<>(
new DataInputViewStreamWrapper(
in),
serializer),
serializer),
- handle.f0);
- if (success) { //if the sending has failed we will retry on the next notify
- committer.commitCheckpoint(pastCheckpointId);
- checkpointsToRemove.add(pastCheckpointId);
+ timestamp);
+ if (success) {
+ // in case the checkpoint was successfully committed, discard its state from the
+ // backend and mark it for removal
+ // in case it failed, we retry on the next checkpoint
+
+ committer.commitCheckpoint(subtaskId, pastCheckpointId);
+ streamHandle.discardState();
+ checkpointsToRemove.add(pendingId);
}
}
} else {
- checkpointsToRemove.add(pastCheckpointId);
+ streamHandle.discardState();
+ checkpointsToRemove.add(pendingId);
}
} catch (Exception e) {
+ // we have to break here to prevent a new (later) checkpoint
+ // from being committed before this one
LOG.error("Could not commit checkpoint.", e);
- break; // we have to break here to prevent a new checkpoint from being committed before this one
+ break;
}
}
}
- for (Long toRemove : checkpointsToRemove) {
- Tuple2<Long, StreamStateHandle> handle = state.pendingHandles.get(toRemove);
- state.pendingHandles.remove(toRemove);
- handle.f1.discardState();
+
+ // ...finally remove the checkpoints that are marked as committed
--- End diff --
unnecessary comment
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2707: [FLINK-4939] GenericWriteAheadSink: Decouple the c...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/2707#discussion_r86106025
--- Diff: flink-streaming-connectors/flink-connector-cassandra/src/main/java/org/apache/flink/streaming/connectors/cassandra/CassandraCommitter.java ---
@@ -118,16 +117,32 @@ public void close() throws Exception {
}
@Override
- public void commitCheckpoint(long checkpointID) {
- session.execute(updateStatement.bind(checkpointID));
- this.lastCommittedCheckpointID = checkpointID;
+ public void commitCheckpoint(int subtaskIdx, long checkpointId) {
+ String statement = String.format(
+ "UPDATE %s.%s set checkpoint_id=%d where sink_id='%s' and sub_id=%d;",
+ keySpace, table, checkpointId, operatorId, subtaskIdx);
+
+ session.execute(statement);
}
@Override
- public boolean isCheckpointCommitted(long checkpointID) {
- if (this.lastCommittedCheckpointID == -1) {
- this.lastCommittedCheckpointID = session.execute(selectStatement.bind()).one().getLong("checkpoint_id");
+ public boolean isCheckpointCommitted(int subtaskIdx, long checkpointId) {
+
+ // Pending checkpointed buffers are committed in ascending order of their
+ // checkpoint id. This way we can tell if a checkpointed buffer was committed
+ // just by asking the third party storage system for the last checkpointed
+ // checkpoint id committed by the specific subtask.
+
+ Long lastCommittedCheckpoint = lastCommittedCheckpoints.get(subtaskIdx);
+ if (lastCommittedCheckpoint == null || lastCommittedCheckpoint == -1) {
--- End diff --
we don't need to put -1 into the Map in case no checkpoint was committed; the map returning null serves the same purpose.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2707: [FLINK-4939] GenericWriteAheadSink: Decouple the creating...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on the issue:
https://github.com/apache/flink/pull/2707
Looks good, +1 from my side.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2707: [FLINK-4939] GenericWriteAheadSink: Decouple the c...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/2707#discussion_r86557631
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericWriteAheadSink.java ---
@@ -77,111 +84,144 @@ public GenericWriteAheadSink(CheckpointCommitter committer, TypeSerializer<IN> s
public void open() throws Exception {
super.open();
committer.setOperatorId(id);
- committer.setOperatorSubtaskId(getRuntimeContext().getIndexOfThisSubtask());
committer.open();
- cleanState();
- checkpointStreamFactory =
- getContainingTask().createCheckpointStreamFactory(this);
+
+ checkpointStreamFactory = getContainingTask()
+ .createCheckpointStreamFactory(this);
+
+ cleanRestoredHandles();
}
public void close() throws Exception {
committer.close();
}
/**
- * Saves a handle in the state.
+ * Called when a checkpoint barrier arrives. It closes any open streams to the backend
+ * and marks them as pending for committing to the external, third-party storage system.
*
- * @param checkpointId
- * @throws IOException
+ * @param checkpointId the id of the latest received checkpoint.
+ * @throws IOException in case something went wrong when handling the stream to the backend.
*/
private void saveHandleInState(final long checkpointId, final long timestamp) throws Exception {
+ Preconditions.checkNotNull(this.pendingCheckpoints, "The operator has not been properly initialized.");
+
//only add handle if a new OperatorState was created since the last snapshot
if (out != null) {
+ int subtaskIdx = getRuntimeContext().getIndexOfThisSubtask();
StreamStateHandle handle = out.closeAndGetHandle();
- if (state.pendingHandles.containsKey(checkpointId)) {
+
+ PendingCheckpoint pendingCheckpoint = new PendingCheckpoint(
+ checkpointId, subtaskIdx, timestamp, handle);
+
+ if (pendingCheckpoints.contains(pendingCheckpoint)) {
//we already have a checkpoint stored for that ID that may have been partially written,
//so we discard this "alternate version" and use the stored checkpoint
handle.discardState();
} else {
- state.pendingHandles.put(checkpointId, new Tuple2<>(timestamp, handle));
+ pendingCheckpoints.add(pendingCheckpoint);
}
out = null;
}
}
@Override
- public void snapshotState(FSDataOutputStream out,
- long checkpointId,
- long timestamp) throws Exception {
+ public void snapshotState(FSDataOutputStream out, long checkpointId, long timestamp) throws Exception {
saveHandleInState(checkpointId, timestamp);
- InstantiationUtil.serializeObject(out, state);
+ DataOutputViewStreamWrapper outStream = new DataOutputViewStreamWrapper(out);
+ outStream.writeInt(pendingCheckpoints.size());
+ for (PendingCheckpoint pendingCheckpoint : pendingCheckpoints) {
+ pendingCheckpoint.serialize(outStream);
+ }
}
@Override
public void restoreState(FSDataInputStream in) throws Exception {
- this.state = InstantiationUtil.deserializeObject(in, getUserCodeClassloader());
+ final DataInputViewStreamWrapper inStream = new DataInputViewStreamWrapper(in);
+ int numPendingHandles = inStream.readInt();
+ for (int i = 0; i < numPendingHandles; i++) {
+ pendingCheckpoints.add(PendingCheckpoint.restore(inStream, getUserCodeClassloader()));
+ }
}
- private void cleanState() throws Exception {
- synchronized (this.state.pendingHandles) { //remove all handles that were already committed
- Set<Long> pastCheckpointIds = this.state.pendingHandles.keySet();
- Set<Long> checkpointsToRemove = new HashSet<>();
- for (Long pastCheckpointId : pastCheckpointIds) {
- if (committer.isCheckpointCommitted(pastCheckpointId)) {
- checkpointsToRemove.add(pastCheckpointId);
+ /**
+ * Called at {@link #open()} to clean-up the pending handle list.
+ * It iterates over all restored pending handles, checks which ones are already
+ * committed to the outside storage system and removes them from the list.
+ */
+ private void cleanRestoredHandles() throws Exception {
+ synchronized (pendingCheckpoints) {
+
+ // for each of the pending handles...
+ Iterator<PendingCheckpoint> pendingCheckpointIt = pendingCheckpoints.iterator();
+ while (pendingCheckpointIt.hasNext()) {
+
+ PendingCheckpoint pendingCheckpoint = pendingCheckpointIt.next();
+ long checkpointId = pendingCheckpoint.checkpointId;
+ int subtaskId = pendingCheckpoint.subtaskId;
+
+ //...check if the temporary buffer is already committed and if yes,
+ // remove it from the list of pending checkpoints.
+ if (committer.isCheckpointCommitted(subtaskId, checkpointId)) {
+ pendingCheckpoint.stateHandle.discardState();
+ pendingCheckpointIt.remove();
--- End diff --
This will throw a `ConcurrentModificationException` according to the [javadocs](https://docs.oracle.com/javase/7/docs/api/java/util/TreeSet.html).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2707: [FLINK-4939] GenericWriteAheadSink: Decouple the c...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/2707#discussion_r86108509
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericWriteAheadSink.java ---
@@ -77,63 +86,96 @@ public GenericWriteAheadSink(CheckpointCommitter committer, TypeSerializer<IN> s
public void open() throws Exception {
super.open();
committer.setOperatorId(id);
- committer.setOperatorSubtaskId(getRuntimeContext().getIndexOfThisSubtask());
committer.open();
- cleanState();
- checkpointStreamFactory =
- getContainingTask().createCheckpointStreamFactory(this);
+
+ checkpointStreamFactory = getContainingTask()
+ .createCheckpointStreamFactory(this);
+
+ cleanRestoredHandles();
}
public void close() throws Exception {
committer.close();
}
/**
- * Saves a handle in the state.
+ * Called when a checkpoint barrier arrives.
+ * Closes any open streams to the backend and marks them as pending for
+ * committing to the final output system, e.g. Cassandra.
*
- * @param checkpointId
- * @throws IOException
+ * @param checkpointId the id of the latest received checkpoint.
+ * @throws IOException in case something went wrong when handling the stream to the backend.
*/
private void saveHandleInState(final long checkpointId, final long timestamp) throws Exception {
+ Preconditions.checkNotNull(this.pendingHandles, "The operator has not been properly initialized.");
+
//only add handle if a new OperatorState was created since the last snapshot
if (out != null) {
StreamStateHandle handle = out.closeAndGetHandle();
- if (state.pendingHandles.containsKey(checkpointId)) {
+
+ PendingCheckpointId pendingCheckpoint = new PendingCheckpointId(
+ checkpointId, getRuntimeContext().getIndexOfThisSubtask());
+
+ if (pendingHandles.containsKey(pendingCheckpoint)) {
//we already have a checkpoint stored for that ID that may have been partially written,
//so we discard this "alternate version" and use the stored checkpoint
handle.discardState();
} else {
- state.pendingHandles.put(checkpointId, new Tuple2<>(timestamp, handle));
+ this.pendingHandles.put(pendingCheckpoint, new PendingHandle(timestamp, handle));
}
out = null;
}
}
@Override
- public void snapshotState(FSDataOutputStream out,
- long checkpointId,
- long timestamp) throws Exception {
+ public void snapshotState(FSDataOutputStream out, long checkpointId, long timestamp) throws Exception {
saveHandleInState(checkpointId, timestamp);
- InstantiationUtil.serializeObject(out, state);
+ DataOutputViewStreamWrapper outStream = new DataOutputViewStreamWrapper(out);
+ outStream.writeInt(pendingHandles.size());
+ for (Map.Entry<PendingCheckpointId, PendingHandle> pendingCheckpoint : pendingHandles.entrySet()) {
+ pendingCheckpoint.getKey().serialize(outStream);
+ pendingCheckpoint.getValue().serialize(outStream);
+ }
}
@Override
public void restoreState(FSDataInputStream in) throws Exception {
- this.state = InstantiationUtil.deserializeObject(in, getUserCodeClassloader());
+ final DataInputViewStreamWrapper inStream = new DataInputViewStreamWrapper(in);
+ int noOfPendingHandles = inStream.readInt();
--- End diff --
noOfPendingHandlers -> numPendingHandles
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2707: [FLINK-4939] GenericWriteAheadSink: Decouple the c...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/2707#discussion_r86104959
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/CheckpointCommitter.java ---
@@ -98,17 +89,19 @@ public void setOperatorSubtaskId(int id) throws Exception {
/**
* Mark the given checkpoint as completed in the resource.
*
- * @param checkpointID
+ * @param subtaskIdx the index of the subtask responsible for committing the checkpoint.
+ * @param checkpointID this id of the checkpoint to be committed.
* @throws Exception
*/
- public abstract void commitCheckpoint(long checkpointID) throws Exception;
+ public abstract void commitCheckpoint(int subtaskIdx, long checkpointID) throws Exception;
/**
* Checked the resource whether the given checkpoint was committed completely.
*
- * @param checkpointID
+ * @param subtaskIdx the index of the subtask responsible for committing the checkpoint.
+ * @param checkpointID this id of the checkpoint we are interested in.
--- End diff --
this id -> the id
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2707: [FLINK-4939] GenericWriteAheadSink: Decouple the c...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/flink/pull/2707
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---