You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by sr...@apache.org on 2017/04/26 17:54:47 UTC
flink git commit: [FLINK-6048] [checkpoint] Implement asynchronous
snapshots for DefaultOperatorStateBackend
Repository: flink
Updated Branches:
refs/heads/master 336b95d4e -> 6bdaf1e4a
[FLINK-6048] [checkpoint] Implement asynchronous snapshots for DefaultOperatorStateBackend
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/6bdaf1e4
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/6bdaf1e4
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/6bdaf1e4
Branch: refs/heads/master
Commit: 6bdaf1e4a5db1c1609cbd7f96b20aab285877535
Parents: 336b95d
Author: Stefan Richter <s....@data-artisans.com>
Authored: Tue Mar 14 14:25:57 2017 +0100
Committer: Stefan Richter <s....@data-artisans.com>
Committed: Wed Apr 26 19:54:17 2017 +0200
----------------------------------------------------------------------
.../state/RocksDBKeyedStateBackend.java | 2 +-
.../streaming/state/RocksDBStateBackend.java | 17 +-
.../state/RocksDBStateBackendTest.java | 2 +-
.../AbstractAsyncSnapshotIOCallable.java | 109 +++++++
.../runtime/state/AbstractStateBackend.java | 7 +-
.../state/DefaultOperatorStateBackend.java | 220 +++++++++++---
.../runtime/state/OperatorStateHandle.java | 6 +-
.../state/filesystem/FsStateBackend.java | 13 +
.../state/heap/HeapKeyedStateBackend.java | 107 +++----
.../state/memory/MemoryStateBackend.java | 13 +
.../BlockerCheckpointStreamFactory.java | 112 +++++++
.../runtime/state/OperatorStateBackendTest.java | 301 +++++++++++++++++--
.../runtime/tasks/BlockingCheckpointsTest.java | 10 +
.../streaming/runtime/StateBackendITCase.java | 9 +
14 files changed, 787 insertions(+), 141 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/6bdaf1e4/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java b/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
index 91d947e..199a5a4 100644
--- a/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
+++ b/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
@@ -421,7 +421,6 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
*/
public void closeCheckpointStream() throws IOException {
if (outStream != null) {
- stateBackend.cancelStreamRegistry.unregisterClosable(outStream);
snapshotResultStateHandle = closeSnapshotStreamAndGetHandle();
} else {
snapshotResultStateHandle = null;
@@ -592,6 +591,7 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
}
private KeyGroupsStateHandle closeSnapshotStreamAndGetHandle() throws IOException {
+ stateBackend.cancelStreamRegistry.unregisterClosable(outStream);
StreamStateHandle stateHandle = outStream.closeAndGetHandle();
outStream = null;
return stateHandle != null ? new KeyGroupsStateHandle(keyGroupRangeOffsets, stateHandle) : null;
http://git-wip-us.apache.org/repos/asf/flink/blob/6bdaf1e4/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackend.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackend.java b/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackend.java
index 16b0869..80c9a29 100644
--- a/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackend.java
+++ b/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackend.java
@@ -26,15 +26,15 @@ import org.apache.flink.runtime.query.TaskKvStateRegistry;
import org.apache.flink.runtime.state.AbstractKeyedStateBackend;
import org.apache.flink.runtime.state.AbstractStateBackend;
import org.apache.flink.runtime.state.CheckpointStreamFactory;
+import org.apache.flink.runtime.state.DefaultOperatorStateBackend;
import org.apache.flink.runtime.state.KeyGroupRange;
+import org.apache.flink.runtime.state.OperatorStateBackend;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.util.AbstractID;
-
import org.rocksdb.ColumnFamilyOptions;
import org.rocksdb.DBOptions;
import org.rocksdb.NativeLibraryLoader;
import org.rocksdb.RocksDB;
-
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -427,6 +427,19 @@ public class RocksDBStateBackend extends AbstractStateBackend {
}
@Override
+ public OperatorStateBackend createOperatorStateBackend(
+ Environment env,
+ String operatorIdentifier) throws Exception {
+
+ //the default for RocksDB; eventually there can be a operator state backend based on RocksDB, too.
+ final boolean asyncSnapshots = true;
+ return new DefaultOperatorStateBackend(
+ env.getUserClassLoader(),
+ env.getExecutionConfig(),
+ asyncSnapshots);
+ }
+
+ @Override
public String toString() {
return "RocksDB State Backend {" +
"isInitialized=" + isInitialized +
http://git-wip-us.apache.org/repos/asf/flink/blob/6bdaf1e4/flink-contrib/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackendTest.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackendTest.java b/flink-contrib/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackendTest.java
index d95a9b4..b5f18a4 100644
--- a/flink-contrib/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackendTest.java
+++ b/flink-contrib/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackendTest.java
@@ -26,6 +26,7 @@ import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeutils.base.IntSerializer;
import org.apache.flink.core.testutils.OneShotLatch;
+import org.apache.flink.runtime.checkpoint.BlockerCheckpointStreamFactory;
import org.apache.flink.runtime.checkpoint.CheckpointOptions;
import org.apache.flink.runtime.execution.Environment;
import org.apache.flink.runtime.operators.testutils.DummyEnvironment;
@@ -38,7 +39,6 @@ import org.apache.flink.runtime.state.StateBackendTestBase;
import org.apache.flink.runtime.state.VoidNamespace;
import org.apache.flink.runtime.state.VoidNamespaceSerializer;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
-import org.apache.flink.runtime.util.BlockerCheckpointStreamFactory;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
http://git-wip-us.apache.org/repos/asf/flink/blob/6bdaf1e4/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/AbstractAsyncSnapshotIOCallable.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/AbstractAsyncSnapshotIOCallable.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/AbstractAsyncSnapshotIOCallable.java
new file mode 100644
index 0000000..1aaa473
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/AbstractAsyncSnapshotIOCallable.java
@@ -0,0 +1,109 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.checkpoint;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.flink.core.fs.CloseableRegistry;
+import org.apache.flink.runtime.io.async.AbstractAsyncIOCallable;
+import org.apache.flink.runtime.state.CheckpointStreamFactory;
+import org.apache.flink.runtime.state.StateObject;
+import org.apache.flink.runtime.state.StreamStateHandle;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * Abstract base class for async IO operations of snapshots against a
+ * {@link java.util.zip.CheckedOutputStream}. This includes participating in lifecycle management
+ * through a {@link CloseableRegistry}.
+ */
+public abstract class AbstractAsyncSnapshotIOCallable<H extends StateObject>
+ extends AbstractAsyncIOCallable<H, CheckpointStreamFactory.CheckpointStateOutputStream> {
+
+ protected final long checkpointId;
+ protected final long timestamp;
+
+ protected final CheckpointStreamFactory streamFactory;
+ protected final CloseableRegistry closeStreamOnCancelRegistry;
+ protected final AtomicBoolean open;
+
+ public AbstractAsyncSnapshotIOCallable(
+ long checkpointId,
+ long timestamp,
+ CheckpointStreamFactory streamFactory,
+ CloseableRegistry closeStreamOnCancelRegistry) {
+
+ this.streamFactory = Preconditions.checkNotNull(streamFactory);
+ this.closeStreamOnCancelRegistry = Preconditions.checkNotNull(closeStreamOnCancelRegistry);
+ this.checkpointId = checkpointId;
+ this.timestamp = timestamp;
+ this.open = new AtomicBoolean(false);
+ }
+
+ @Override
+ public CheckpointStreamFactory.CheckpointStateOutputStream openIOHandle() throws Exception {
+ if (checkStreamClosedAndDoTransitionToOpen()) {
+ CheckpointStreamFactory.CheckpointStateOutputStream stream =
+ streamFactory.createCheckpointStateOutputStream(checkpointId, timestamp);
+ try {
+ closeStreamOnCancelRegistry.registerClosable(stream);
+ return stream;
+ } catch (Exception ex) {
+ open.set(false);
+ throw ex;
+ }
+ } else {
+ throw new IOException("Async snapshot: a checkpoint stream was already opened.");
+ }
+ }
+
+ @Override
+ public void done(boolean canceled) {
+ if (checkStreamOpenAndDoTransitionToClose()) {
+ CheckpointStreamFactory.CheckpointStateOutputStream stream = getIoHandle();
+ if (stream != null) {
+ closeStreamOnCancelRegistry.unregisterClosable(stream);
+ IOUtils.closeQuietly(stream);
+ }
+ }
+ }
+
+ protected boolean checkStreamClosedAndDoTransitionToOpen() {
+ return open.compareAndSet(false, true);
+ }
+
+ protected boolean checkStreamOpenAndDoTransitionToClose() {
+ return open.compareAndSet(true, false);
+ }
+
+ protected StreamStateHandle closeStreamAndGetStateHandle() throws IOException {
+ if (checkStreamOpenAndDoTransitionToClose()) {
+ final CheckpointStreamFactory.CheckpointStateOutputStream stream = getIoHandle();
+ try {
+ return stream.closeAndGetHandle();
+ } finally {
+ closeStreamOnCancelRegistry.unregisterClosable(stream);
+ }
+ } else {
+ throw new IOException("Checkpoint stream already closed.");
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/6bdaf1e4/flink-runtime/src/main/java/org/apache/flink/runtime/state/AbstractStateBackend.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/AbstractStateBackend.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/AbstractStateBackend.java
index 74025bf..1594e2e 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/AbstractStateBackend.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/AbstractStateBackend.java
@@ -91,12 +91,9 @@ public abstract class AbstractStateBackend implements StateBackend, java.io.Seri
TaskKvStateRegistry kvStateRegistry) throws IOException;
@Override
- public OperatorStateBackend createOperatorStateBackend(
+ public abstract OperatorStateBackend createOperatorStateBackend(
Environment env,
- String operatorIdentifier) throws Exception {
-
- return new DefaultOperatorStateBackend(env.getUserClassLoader(), env.getExecutionConfig());
- }
+ String operatorIdentifier) throws Exception;
// ------------------------------------------------------------------------
// Loading the state backend from a configuration
http://git-wip-us.apache.org/repos/asf/flink/blob/6bdaf1e4/flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackend.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackend.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackend.java
index eb3ba01..e7ed26f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackend.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackend.java
@@ -20,10 +20,10 @@ package org.apache.flink.runtime.state;
import org.apache.commons.io.IOUtils;
import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeutils.TypeSerializer;
-import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.core.fs.CloseableRegistry;
import org.apache.flink.core.fs.FSDataInputStream;
import org.apache.flink.core.fs.FSDataOutputStream;
@@ -31,8 +31,12 @@ import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.core.memory.DataInputViewStreamWrapper;
import org.apache.flink.core.memory.DataOutputView;
import org.apache.flink.core.memory.DataOutputViewStreamWrapper;
+import org.apache.flink.runtime.checkpoint.AbstractAsyncSnapshotIOCallable;
import org.apache.flink.runtime.checkpoint.CheckpointOptions;
+import org.apache.flink.runtime.io.async.AsyncStoppableTaskWithCallback;
import org.apache.flink.util.Preconditions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.io.Serializable;
@@ -50,21 +54,54 @@ import java.util.concurrent.RunnableFuture;
@Internal
public class DefaultOperatorStateBackend implements OperatorStateBackend {
- /** The default namespace for state in cases where no state name is provided */
+ private static final Logger LOG = LoggerFactory.getLogger(DefaultOperatorStateBackend.class);
+
+ /**
+ * The default namespace for state in cases where no state name is provided
+ */
public static final String DEFAULT_OPERATOR_STATE_NAME = "_default_";
-
+
+ /**
+ * Map for all registered operator states. Maps state name -> state
+ */
private final Map<String, PartitionableListState<?>> registeredStates;
+
+ /**
+ * CloseableRegistry to participate in the tasks lifecycle.
+ */
private final CloseableRegistry closeStreamOnCancelRegistry;
+
+ /**
+ * Default serializer. Only used for the default operator state.
+ */
private final JavaSerializer<Serializable> javaSerializer;
+
+ /**
+ * The user code classloader.
+ */
private final ClassLoader userClassloader;
+
+ /**
+ * The execution configuration.
+ */
private final ExecutionConfig executionConfig;
- public DefaultOperatorStateBackend(ClassLoader userClassLoader, ExecutionConfig executionConfig) throws IOException {
+ /**
+ * Flag to de/activate asynchronous snapshots.
+ */
+ private final boolean asynchronousSnapshots;
+
+ public DefaultOperatorStateBackend(
+ ClassLoader userClassLoader,
+ ExecutionConfig executionConfig,
+ boolean asynchronousSnapshots) throws IOException {
+
this.closeStreamOnCancelRegistry = new CloseableRegistry();
this.userClassloader = Preconditions.checkNotNull(userClassLoader);
this.executionConfig = executionConfig;
this.javaSerializer = new JavaSerializer<>();
this.registeredStates = new HashMap<>();
+ this.asynchronousSnapshots = asynchronousSnapshots;
}
public ExecutionConfig getExecutionConfig() {
@@ -131,59 +168,109 @@ public class DefaultOperatorStateBackend implements OperatorStateBackend {
@Override
public RunnableFuture<OperatorStateHandle> snapshot(
- long checkpointId,
- long timestamp,
- CheckpointStreamFactory streamFactory,
- CheckpointOptions checkpointOptions) throws Exception {
+ final long checkpointId,
+ final long timestamp,
+ final CheckpointStreamFactory streamFactory,
+ final CheckpointOptions checkpointOptions) throws Exception {
+
+ final long syncStartTime = System.currentTimeMillis();
if (registeredStates.isEmpty()) {
return DoneFuture.nullValue();
}
- List<OperatorBackendSerializationProxy.StateMetaInfo<?>> metaInfoList =
- new ArrayList<>(registeredStates.size());
-
- for (Map.Entry<String, PartitionableListState<?>> entry : registeredStates.entrySet()) {
- PartitionableListState<?> state = entry.getValue();
- OperatorBackendSerializationProxy.StateMetaInfo<?> metaInfo =
- new OperatorBackendSerializationProxy.StateMetaInfo<>(
- state.getName(),
- state.getPartitionStateSerializer(),
- state.getAssignmentMode());
- metaInfoList.add(metaInfo);
+ final Map<String, PartitionableListState<?>> registeredStatesDeepCopies =
+ new HashMap<>(registeredStates.size());
+
+ // eagerly create deep copies of the list states in the sync phase, so that we can use them in the async writing
+ for (Map.Entry<String, PartitionableListState<?>> entry : this.registeredStates.entrySet()) {
+
+ PartitionableListState<?> listState = entry.getValue();
+ if (null != listState) {
+ listState = listState.deepCopy();
+ }
+ registeredStatesDeepCopies.put(entry.getKey(), listState);
}
- Map<String, OperatorStateHandle.StateMetaInfo> writtenStatesMetaData = new HashMap<>(registeredStates.size());
+ // implementation of the async IO operation, based on FutureTask
+ final AbstractAsyncSnapshotIOCallable<OperatorStateHandle> ioCallable =
+ new AbstractAsyncSnapshotIOCallable<OperatorStateHandle>(
+ checkpointId,
+ timestamp,
+ streamFactory,
+ closeStreamOnCancelRegistry) {
+
+ @Override
+ public OperatorStateHandle performOperation() throws Exception {
+ long asyncStartTime = System.currentTimeMillis();
+
+ final Map<String, OperatorStateHandle.StateMetaInfo> writtenStatesMetaData =
+ new HashMap<>(registeredStatesDeepCopies.size());
+
+ List<OperatorBackendSerializationProxy.StateMetaInfo<?>> metaInfoList =
+ new ArrayList<>(registeredStatesDeepCopies.size());
+
+ for (Map.Entry<String, PartitionableListState<?>> entry :
+ registeredStatesDeepCopies.entrySet()) {
+
+ PartitionableListState<?> state = entry.getValue();
+ OperatorBackendSerializationProxy.StateMetaInfo<?> metaInfo =
+ new OperatorBackendSerializationProxy.StateMetaInfo<>(
+ state.getName(),
+ state.getPartitionStateSerializer(),
+ state.getAssignmentMode());
+ metaInfoList.add(metaInfo);
+ }
- CheckpointStreamFactory.CheckpointStateOutputStream out = streamFactory.
- createCheckpointStateOutputStream(checkpointId, timestamp);
+ CheckpointStreamFactory.CheckpointStateOutputStream out = getIoHandle();
+ DataOutputView dov = new DataOutputViewStreamWrapper(out);
- try {
- closeStreamOnCancelRegistry.registerClosable(out);
+ OperatorBackendSerializationProxy backendSerializationProxy =
+ new OperatorBackendSerializationProxy(metaInfoList);
- DataOutputView dov = new DataOutputViewStreamWrapper(out);
+ backendSerializationProxy.write(dov);
+
+ dov.writeInt(registeredStatesDeepCopies.size());
- OperatorBackendSerializationProxy backendSerializationProxy =
- new OperatorBackendSerializationProxy(metaInfoList);
+ for (Map.Entry<String, PartitionableListState<?>> entry :
+ registeredStatesDeepCopies.entrySet()) {
- backendSerializationProxy.write(dov);
+ PartitionableListState<?> value = entry.getValue();
+ long[] partitionOffsets = value.write(out);
+ OperatorStateHandle.Mode mode = value.getAssignmentMode();
+ writtenStatesMetaData.put(
+ entry.getKey(),
+ new OperatorStateHandle.StateMetaInfo(partitionOffsets, mode));
+ }
- dov.writeInt(registeredStates.size());
- for (Map.Entry<String, PartitionableListState<?>> entry : registeredStates.entrySet()) {
+ StreamStateHandle stateHandle = closeStreamAndGetStateHandle();
- PartitionableListState<?> value = entry.getValue();
- long[] partitionOffsets = value.write(out);
- OperatorStateHandle.Mode mode = value.getAssignmentMode();
- writtenStatesMetaData.put(entry.getKey(), new OperatorStateHandle.StateMetaInfo(partitionOffsets, mode));
- }
+ if (asynchronousSnapshots) {
+ LOG.info("DefaultOperatorStateBackend snapshot ({}, asynchronous part) in thread {} took {} ms.",
+ streamFactory, Thread.currentThread(), (System.currentTimeMillis() - asyncStartTime));
+ }
+
+ if (stateHandle == null) {
+ return null;
+ }
- OperatorStateHandle handle = new OperatorStateHandle(writtenStatesMetaData, out.closeAndGetHandle());
+ OperatorStateHandle operatorStateHandle =
+ new OperatorStateHandle(writtenStatesMetaData, stateHandle);
- return new DoneFuture<>(handle);
- } finally {
- closeStreamOnCancelRegistry.unregisterClosable(out);
- out.close();
+ return operatorStateHandle;
+ }
+ };
+
+ AsyncStoppableTaskWithCallback<OperatorStateHandle> task = AsyncStoppableTaskWithCallback.from(ioCallable);
+
+ if (!asynchronousSnapshots) {
+ task.run();
}
+
+ LOG.info("DefaultOperatorStateBackend snapshot (" + streamFactory + ", synchronous part) in thread " +
+ Thread.currentThread() + " took " + (System.currentTimeMillis() - syncStartTime) + " ms.");
+
+ return task;
}
@Override
@@ -253,22 +340,67 @@ public class DefaultOperatorStateBackend implements OperatorStateBackend {
}
}
+ /**
+ *
+ * Implementation of operator list state.
+ *
+ * @param <S> the type of an operator state partition.
+ */
static final class PartitionableListState<S> implements ListState<S> {
+ /**
+ * The name of the state, as registered by the user
+ */
private final String name;
+
+ /**
+ * The type serializer for the elements in the state list
+ */
private final TypeSerializer<S> partitionStateSerializer;
+
+ /**
+ * The mode how elements in this state are assigned to tasks during restore
+ */
private final OperatorStateHandle.Mode assignmentMode;
- private final List<S> internalList;
+
+ /**
+ * The internal list the holds the elements of the state
+ */
+ private final ArrayList<S> internalList;
+
+ /**
+ * A serializer that allows to perfom deep copies of internalList
+ */
+ private final ArrayListSerializer<S> internalListCopySerializer;
public PartitionableListState(
String name,
TypeSerializer<S> partitionStateSerializer,
OperatorStateHandle.Mode assignmentMode) {
+ this(name, partitionStateSerializer, assignmentMode, new ArrayList<S>());
+ }
+
+ private PartitionableListState(
+ String name,
+ TypeSerializer<S> partitionStateSerializer,
+ OperatorStateHandle.Mode assignmentMode,
+ ArrayList<S> internalList) {
+
this.name = Preconditions.checkNotNull(name);
this.partitionStateSerializer = Preconditions.checkNotNull(partitionStateSerializer);
this.assignmentMode = Preconditions.checkNotNull(assignmentMode);
- this.internalList = new ArrayList<>();
+ this.internalList = Preconditions.checkNotNull(internalList);
+ this.internalListCopySerializer = new ArrayListSerializer<>(partitionStateSerializer);
+ }
+
+ private PartitionableListState(PartitionableListState<S> toCopy) {
+
+ this(
+ toCopy.name,
+ toCopy.partitionStateSerializer.duplicate(),
+ toCopy.assignmentMode,
+ toCopy.internalListCopySerializer.copy(toCopy.internalList));
}
public String getName() {
@@ -287,6 +419,10 @@ public class DefaultOperatorStateBackend implements OperatorStateBackend {
return internalList;
}
+ public PartitionableListState<S> deepCopy() {
+ return new PartitionableListState<>(this);
+ }
+
@Override
public void clear() {
internalList.clear();
http://git-wip-us.apache.org/repos/asf/flink/blob/6bdaf1e4/flink-runtime/src/main/java/org/apache/flink/runtime/state/OperatorStateHandle.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/OperatorStateHandle.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/OperatorStateHandle.java
index c59fbad..7c49338 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/OperatorStateHandle.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/OperatorStateHandle.java
@@ -32,8 +32,12 @@ import java.util.Map;
*/
public class OperatorStateHandle implements StreamStateHandle {
+ /**
+ * The modes that determine how an {@link OperatorStateHandle} is assigned to tasks during restore.
+ */
public enum Mode {
- SPLIT_DISTRIBUTE, BROADCAST
+ SPLIT_DISTRIBUTE, // The operator state partitions in the state handle are split and distributed to one task each.
+ BROADCAST // The operator state partitions are broadcasted to all task.
}
private static final long serialVersionUID = 35876522969227335L;
http://git-wip-us.apache.org/repos/asf/flink/blob/6bdaf1e4/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsStateBackend.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsStateBackend.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsStateBackend.java
index e27712c..e320bf3 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsStateBackend.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsStateBackend.java
@@ -27,7 +27,9 @@ import org.apache.flink.runtime.query.TaskKvStateRegistry;
import org.apache.flink.runtime.state.AbstractKeyedStateBackend;
import org.apache.flink.runtime.state.AbstractStateBackend;
import org.apache.flink.runtime.state.CheckpointStreamFactory;
+import org.apache.flink.runtime.state.DefaultOperatorStateBackend;
import org.apache.flink.runtime.state.KeyGroupRange;
+import org.apache.flink.runtime.state.OperatorStateBackend;
import org.apache.flink.runtime.state.heap.HeapKeyedStateBackend;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -310,6 +312,17 @@ public class FsStateBackend extends AbstractStateBackend {
}
@Override
+ public OperatorStateBackend createOperatorStateBackend(
+ Environment env,
+ String operatorIdentifier) throws Exception {
+
+ return new DefaultOperatorStateBackend(
+ env.getUserClassLoader(),
+ env.getExecutionConfig(),
+ asynchronousSnapshots);
+ }
+
+ @Override
public String toString() {
return "File State Backend @ " + basePath;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6bdaf1e4/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapKeyedStateBackend.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapKeyedStateBackend.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapKeyedStateBackend.java
index 9247ffa..38817cd 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapKeyedStateBackend.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapKeyedStateBackend.java
@@ -37,10 +37,10 @@ import org.apache.flink.core.memory.DataOutputViewStreamWrapper;
import org.apache.flink.migration.MigrationUtil;
import org.apache.flink.migration.runtime.state.KvStateSnapshot;
import org.apache.flink.migration.runtime.state.memory.MigrationRestoreSnapshot;
-import org.apache.flink.runtime.io.async.AbstractAsyncIOCallable;
-import org.apache.flink.runtime.io.async.AsyncStoppableTaskWithCallback;
import org.apache.flink.migration.state.MigrationKeyGroupStateHandle;
+import org.apache.flink.runtime.checkpoint.AbstractAsyncSnapshotIOCallable;
import org.apache.flink.runtime.checkpoint.CheckpointOptions;
+import org.apache.flink.runtime.io.async.AsyncStoppableTaskWithCallback;
import org.apache.flink.runtime.query.TaskKvStateRegistry;
import org.apache.flink.runtime.state.AbstractKeyedStateBackend;
import org.apache.flink.runtime.state.ArrayListSerializer;
@@ -72,7 +72,6 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.RunnableFuture;
-import java.util.concurrent.atomic.AtomicBoolean;
/**
* A {@link AbstractKeyedStateBackend} that keeps state on the Java Heap and will serialize state to
@@ -269,78 +268,50 @@ public class HeapKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
//--------------------------------------------------- this becomes the end of sync part
// implementation of the async IO operation, based on FutureTask
- final AbstractAsyncIOCallable<KeyedStateHandle, CheckpointStreamFactory.CheckpointStateOutputStream> ioCallable =
- new AbstractAsyncIOCallable<KeyedStateHandle, CheckpointStreamFactory.CheckpointStateOutputStream>() {
-
- AtomicBoolean open = new AtomicBoolean(false);
-
- @Override
- public CheckpointStreamFactory.CheckpointStateOutputStream openIOHandle() throws Exception {
- if (open.compareAndSet(false, true)) {
- CheckpointStreamFactory.CheckpointStateOutputStream stream =
- streamFactory.createCheckpointStateOutputStream(checkpointId, timestamp);
- try {
- cancelStreamRegistry.registerClosable(stream);
- return stream;
- } catch (Exception ex) {
- open.set(false);
- throw ex;
- }
- } else {
- throw new IOException("Operation already opened.");
+ final AbstractAsyncSnapshotIOCallable<KeyedStateHandle> ioCallable =
+ new AbstractAsyncSnapshotIOCallable<KeyedStateHandle>(
+ checkpointId,
+ timestamp,
+ streamFactory,
+ cancelStreamRegistry) {
+
+ @Override
+ public KeyGroupsStateHandle performOperation() throws Exception {
+ long asyncStartTime = System.currentTimeMillis();
+ CheckpointStreamFactory.CheckpointStateOutputStream stream = getIoHandle();
+ DataOutputViewStreamWrapper outView = new DataOutputViewStreamWrapper(stream);
+ serializationProxy.write(outView);
+
+ long[] keyGroupRangeOffsets = new long[keyGroupRange.getNumberOfKeyGroups()];
+
+ for (int keyGroupPos = 0; keyGroupPos < keyGroupRange.getNumberOfKeyGroups(); ++keyGroupPos) {
+ int keyGroupId = keyGroupRange.getKeyGroupId(keyGroupPos);
+ keyGroupRangeOffsets[keyGroupPos] = stream.getPos();
+ outView.writeInt(keyGroupId);
+
+ for (Map.Entry<String, StateTable<K, ?, ?>> kvState : stateTables.entrySet()) {
+ outView.writeShort(kVStateToId.get(kvState.getKey()));
+ cowStateStableSnapshots.get(kvState.getValue()).writeMappingsInKeyGroup(outView, keyGroupId);
}
}
- @Override
- public KeyGroupsStateHandle performOperation() throws Exception {
- long asyncStartTime = System.currentTimeMillis();
- CheckpointStreamFactory.CheckpointStateOutputStream stream = getIoHandle();
- DataOutputViewStreamWrapper outView = new DataOutputViewStreamWrapper(stream);
- serializationProxy.write(outView);
-
- long[] keyGroupRangeOffsets = new long[keyGroupRange.getNumberOfKeyGroups()];
-
- for (int keyGroupPos = 0; keyGroupPos < keyGroupRange.getNumberOfKeyGroups(); ++keyGroupPos) {
- int keyGroupId = keyGroupRange.getKeyGroupId(keyGroupPos);
- keyGroupRangeOffsets[keyGroupPos] = stream.getPos();
- outView.writeInt(keyGroupId);
-
- for (Map.Entry<String, StateTable<K, ?, ?>> kvState : stateTables.entrySet()) {
- outView.writeShort(kVStateToId.get(kvState.getKey()));
- cowStateStableSnapshots.get(kvState.getValue()).writeMappingsInKeyGroup(outView, keyGroupId);
- }
- }
-
- if (open.compareAndSet(true, false)) {
- StreamStateHandle streamStateHandle = stream.closeAndGetHandle();
- KeyGroupRangeOffsets offsets = new KeyGroupRangeOffsets(keyGroupRange, keyGroupRangeOffsets);
- final KeyGroupsStateHandle keyGroupsStateHandle = new KeyGroupsStateHandle(offsets, streamStateHandle);
+ final StreamStateHandle streamStateHandle = closeStreamAndGetStateHandle();
- if (asynchronousSnapshots) {
- LOG.info("Heap backend snapshot ({}, asynchronous part) in thread {} took {} ms.",
- streamFactory, Thread.currentThread(), (System.currentTimeMillis() - asyncStartTime));
- }
-
- return keyGroupsStateHandle;
- } else {
- throw new IOException("Checkpoint stream already closed.");
- }
+ if (asynchronousSnapshots) {
+ LOG.info("Heap backend snapshot ({}, asynchronous part) in thread {} took {} ms.",
+ streamFactory, Thread.currentThread(), (System.currentTimeMillis() - asyncStartTime));
}
- @Override
- public void done(boolean canceled) {
- if (open.compareAndSet(true, false)) {
- CheckpointStreamFactory.CheckpointStateOutputStream stream = getIoHandle();
- if (null != stream) {
- cancelStreamRegistry.unregisterClosable(stream);
- IOUtils.closeQuietly(stream);
- }
- }
- for (StateTableSnapshot snapshot : cowStateStableSnapshots.values()) {
- snapshot.release();
- }
+ if (streamStateHandle == null) {
+ return null;
}
- };
+
+ KeyGroupRangeOffsets offsets = new KeyGroupRangeOffsets(keyGroupRange, keyGroupRangeOffsets);
+ final KeyGroupsStateHandle keyGroupsStateHandle = new KeyGroupsStateHandle(offsets, streamStateHandle);
+
+ return keyGroupsStateHandle;
+ }
+ };
AsyncStoppableTaskWithCallback<KeyedStateHandle> task = AsyncStoppableTaskWithCallback.from(ioCallable);
http://git-wip-us.apache.org/repos/asf/flink/blob/6bdaf1e4/flink-runtime/src/main/java/org/apache/flink/runtime/state/memory/MemoryStateBackend.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/memory/MemoryStateBackend.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/memory/MemoryStateBackend.java
index f0bac1b..7ed1dea 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/memory/MemoryStateBackend.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/memory/MemoryStateBackend.java
@@ -25,7 +25,9 @@ import org.apache.flink.runtime.query.TaskKvStateRegistry;
import org.apache.flink.runtime.state.AbstractKeyedStateBackend;
import org.apache.flink.runtime.state.AbstractStateBackend;
import org.apache.flink.runtime.state.CheckpointStreamFactory;
+import org.apache.flink.runtime.state.DefaultOperatorStateBackend;
import org.apache.flink.runtime.state.KeyGroupRange;
+import org.apache.flink.runtime.state.OperatorStateBackend;
import org.apache.flink.runtime.state.heap.HeapKeyedStateBackend;
import java.io.IOException;
@@ -89,6 +91,17 @@ public class MemoryStateBackend extends AbstractStateBackend {
}
@Override
+ public OperatorStateBackend createOperatorStateBackend(
+ Environment env,
+ String operatorIdentifier) throws Exception {
+
+ return new DefaultOperatorStateBackend(
+ env.getUserClassLoader(),
+ env.getExecutionConfig(),
+ asynchronousSnapshots);
+ }
+
+ @Override
public String toString() {
return "MemoryStateBackend (data in heap memory / checkpoints to JobManager)";
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6bdaf1e4/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/BlockerCheckpointStreamFactory.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/BlockerCheckpointStreamFactory.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/BlockerCheckpointStreamFactory.java
new file mode 100644
index 0000000..6f892e2
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/BlockerCheckpointStreamFactory.java
@@ -0,0 +1,112 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.checkpoint;
+
+import org.apache.flink.core.testutils.OneShotLatch;
+import org.apache.flink.runtime.state.CheckpointStreamFactory;
+import org.apache.flink.runtime.state.memory.MemCheckpointStreamFactory;
+
+import java.io.IOException;
+
+/**
+ * A {@link CheckpointStreamFactory} for tests that creates streams that block on a latch to test concurrency in
+ * checkpointing.
+ */
+public class BlockerCheckpointStreamFactory implements CheckpointStreamFactory {
+
+ private final int maxSize;
+ private int afterNumberInvocations;
+ private OneShotLatch blocker;
+ private OneShotLatch waiter;
+
+ MemCheckpointStreamFactory.MemoryCheckpointOutputStream lastCreatedStream;
+
+ public MemCheckpointStreamFactory.MemoryCheckpointOutputStream getLastCreatedStream() {
+ return lastCreatedStream;
+ }
+
+ public BlockerCheckpointStreamFactory(int maxSize) {
+ this.maxSize = maxSize;
+ }
+
+ public void setAfterNumberInvocations(int afterNumberInvocations) {
+ this.afterNumberInvocations = afterNumberInvocations;
+ }
+
+ public void setBlockerLatch(OneShotLatch latch) {
+ this.blocker = latch;
+ }
+
+ public void setWaiterLatch(OneShotLatch latch) {
+ this.waiter = latch;
+ }
+
+ @Override
+ public MemCheckpointStreamFactory.MemoryCheckpointOutputStream createCheckpointStateOutputStream(long checkpointID, long timestamp) throws Exception {
+ waiter.trigger();
+ this.lastCreatedStream = new MemCheckpointStreamFactory.MemoryCheckpointOutputStream(maxSize) {
+
+ private int afterNInvocations = afterNumberInvocations;
+ private final OneShotLatch streamBlocker = blocker;
+ private final OneShotLatch streamWaiter = waiter;
+
+ @Override
+ public void write(int b) throws IOException {
+
+ if (afterNInvocations > 0) {
+ --afterNInvocations;
+ }
+
+ if (0 == afterNInvocations && null != streamBlocker) {
+ try {
+ streamBlocker.await();
+ } catch (InterruptedException ignored) {
+ }
+ }
+ try {
+ super.write(b);
+ } catch (IOException ex) {
+ if (null != streamWaiter) {
+ streamWaiter.trigger();
+ }
+ throw ex;
+ }
+
+ if (0 == afterNInvocations && null != streamWaiter) {
+ streamWaiter.trigger();
+ }
+ }
+
+ @Override
+ public void close() {
+ super.close();
+ if (null != streamWaiter) {
+ streamWaiter.trigger();
+ }
+ }
+ };
+
+ return lastCreatedStream;
+ }
+
+ @Override
+ public void close() throws Exception {
+
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/6bdaf1e4/flink-runtime/src/test/java/org/apache/flink/runtime/state/OperatorStateBackendTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/OperatorStateBackendTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/OperatorStateBackendTest.java
index 13a6307..c04ed8c 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/OperatorStateBackendTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/OperatorStateBackendTest.java
@@ -21,22 +21,30 @@ import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.core.testutils.OneShotLatch;
+import org.apache.flink.runtime.checkpoint.BlockerCheckpointStreamFactory;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer;
import org.apache.flink.runtime.checkpoint.CheckpointOptions;
import org.apache.flink.runtime.execution.Environment;
import org.apache.flink.runtime.state.DefaultOperatorStateBackend.PartitionableListState;
import org.apache.flink.runtime.state.memory.MemoryStateBackend;
-
import org.apache.flink.util.FutureUtil;
+import org.junit.Assert;
import org.junit.Test;
+import java.io.IOException;
import java.io.Serializable;
import java.io.File;
import java.util.Collections;
import java.util.Iterator;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
import java.util.concurrent.FutureTask;
import java.util.concurrent.RunnableFuture;
+import java.util.concurrent.TimeUnit;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
@@ -75,7 +83,7 @@ public class OperatorStateBackendTest {
final ExecutionConfig cfg = new ExecutionConfig();
cfg.registerTypeWithKryoSerializer(registeredType, com.esotericsoftware.kryo.serializers.JavaSerializer.class);
- final OperatorStateBackend operatorStateBackend = new DefaultOperatorStateBackend(classLoader, cfg);
+ final OperatorStateBackend operatorStateBackend = new DefaultOperatorStateBackend(classLoader, cfg, false);
ListStateDescriptor<File> stateDescriptor = new ListStateDescriptor<>("test", File.class);
ListStateDescriptor<String> stateDescriptor2 = new ListStateDescriptor<>("test2", String.class);
@@ -108,7 +116,7 @@ public class OperatorStateBackendTest {
@Test
public void testRegisterStates() throws Exception {
final OperatorStateBackend operatorStateBackend =
- new DefaultOperatorStateBackend(classLoader, new ExecutionConfig());
+ new DefaultOperatorStateBackend(classLoader, new ExecutionConfig(), false);
ListStateDescriptor<Serializable> stateDescriptor1 = new ListStateDescriptor<>("test1", new JavaSerializer<>());
ListStateDescriptor<Serializable> stateDescriptor2 = new ListStateDescriptor<>("test2", new JavaSerializer<>());
@@ -117,19 +125,19 @@ public class OperatorStateBackendTest {
assertNotNull(listState1);
assertEquals(1, operatorStateBackend.getRegisteredStateNames().size());
Iterator<Serializable> it = listState1.get().iterator();
- assertTrue(!it.hasNext());
+ assertFalse(it.hasNext());
listState1.add(42);
listState1.add(4711);
it = listState1.get().iterator();
assertEquals(42, it.next());
assertEquals(4711, it.next());
- assertTrue(!it.hasNext());
+ assertFalse(it.hasNext());
ListState<Serializable> listState2 = operatorStateBackend.getListState(stateDescriptor2);
assertNotNull(listState2);
assertEquals(2, operatorStateBackend.getRegisteredStateNames().size());
- assertTrue(!it.hasNext());
+ assertFalse(it.hasNext());
listState2.add(7);
listState2.add(13);
listState2.add(23);
@@ -138,12 +146,12 @@ public class OperatorStateBackendTest {
assertEquals(7, it.next());
assertEquals(13, it.next());
assertEquals(23, it.next());
- assertTrue(!it.hasNext());
+ assertFalse(it.hasNext());
ListState<Serializable> listState3 = operatorStateBackend.getUnionListState(stateDescriptor3);
assertNotNull(listState3);
assertEquals(3, operatorStateBackend.getRegisteredStateNames().size());
- assertTrue(!it.hasNext());
+ assertFalse(it.hasNext());
listState3.add(17);
listState3.add(3);
listState3.add(123);
@@ -152,7 +160,7 @@ public class OperatorStateBackendTest {
assertEquals(17, it.next());
assertEquals(3, it.next());
assertEquals(123, it.next());
- assertTrue(!it.hasNext());
+ assertFalse(it.hasNext());
ListState<Serializable> listState1b = operatorStateBackend.getListState(stateDescriptor1);
assertNotNull(listState1b);
@@ -161,19 +169,19 @@ public class OperatorStateBackendTest {
assertEquals(42, it.next());
assertEquals(4711, it.next());
assertEquals(123, it.next());
- assertTrue(!it.hasNext());
+ assertFalse(it.hasNext());
it = listState1.get().iterator();
assertEquals(42, it.next());
assertEquals(4711, it.next());
assertEquals(123, it.next());
- assertTrue(!it.hasNext());
+ assertFalse(it.hasNext());
it = listState1b.get().iterator();
assertEquals(42, it.next());
assertEquals(4711, it.next());
assertEquals(123, it.next());
- assertTrue(!it.hasNext());
+ assertFalse(it.hasNext());
try {
operatorStateBackend.getUnionListState(stateDescriptor2);
@@ -208,12 +216,10 @@ public class OperatorStateBackendTest {
}
@Test
- public void testSnapshotRestore() throws Exception {
+ public void testSnapshotRestoreSync() throws Exception {
AbstractStateBackend abstractStateBackend = new MemoryStateBackend(4096);
- OperatorStateBackend operatorStateBackend =
- abstractStateBackend.createOperatorStateBackend(createMockEnvironment(), "test-op-name");
-
+ OperatorStateBackend operatorStateBackend = abstractStateBackend.createOperatorStateBackend(createMockEnvironment(), "test-op-name");
ListStateDescriptor<Serializable> stateDescriptor1 = new ListStateDescriptor<>("test1", new JavaSerializer<>());
ListStateDescriptor<Serializable> stateDescriptor2 = new ListStateDescriptor<>("test2", new JavaSerializer<>());
ListStateDescriptor<Serializable> stateDescriptor3 = new ListStateDescriptor<>("test3", new JavaSerializer<>());
@@ -234,8 +240,9 @@ public class OperatorStateBackendTest {
listState3.add(20);
CheckpointStreamFactory streamFactory = abstractStateBackend.createStreamFactory(new JobID(), "testOperator");
- OperatorStateHandle stateHandle = FutureUtil.runIfNotDoneAndGet(
- operatorStateBackend.snapshot(1, 1, streamFactory, CheckpointOptions.forFullCheckpoint()));
+ RunnableFuture<OperatorStateHandle> runnableFuture =
+ operatorStateBackend.snapshot(1, 1, streamFactory, CheckpointOptions.forFullCheckpoint());
+ OperatorStateHandle stateHandle = FutureUtil.runIfNotDoneAndGet(runnableFuture);
try {
@@ -259,26 +266,278 @@ public class OperatorStateBackendTest {
Iterator<Serializable> it = listState1.get().iterator();
assertEquals(42, it.next());
assertEquals(4711, it.next());
- assertTrue(!it.hasNext());
+ assertFalse(it.hasNext());
it = listState2.get().iterator();
assertEquals(7, it.next());
assertEquals(13, it.next());
assertEquals(23, it.next());
- assertTrue(!it.hasNext());
+ assertFalse(it.hasNext());
it = listState3.get().iterator();
assertEquals(17, it.next());
assertEquals(18, it.next());
assertEquals(19, it.next());
assertEquals(20, it.next());
- assertTrue(!it.hasNext());
+ assertFalse(it.hasNext());
+
+ operatorStateBackend.close();
+ operatorStateBackend.dispose();
+ } finally {
+ stateHandle.discardState();
+ }
+ }
+
+ @Test
+ public void testSnapshotRestoreAsync() throws Exception {
+ DefaultOperatorStateBackend operatorStateBackend =
+ new DefaultOperatorStateBackend(OperatorStateBackendTest.class.getClassLoader(), new ExecutionConfig(), true);
+
+ ListStateDescriptor<MutableType> stateDescriptor1 =
+ new ListStateDescriptor<>("test1", new JavaSerializer<MutableType>());
+ ListStateDescriptor<MutableType> stateDescriptor2 =
+ new ListStateDescriptor<>("test2", new JavaSerializer<MutableType>());
+ ListStateDescriptor<MutableType> stateDescriptor3 =
+ new ListStateDescriptor<>("test3", new JavaSerializer<MutableType>());
+ ListState<MutableType> listState1 = operatorStateBackend.getListState(stateDescriptor1);
+ ListState<MutableType> listState2 = operatorStateBackend.getListState(stateDescriptor2);
+ ListState<MutableType> listState3 = operatorStateBackend.getUnionListState(stateDescriptor3);
+
+ listState1.add(MutableType.of(42));
+ listState1.add(MutableType.of(4711));
+
+ listState2.add(MutableType.of(7));
+ listState2.add(MutableType.of(13));
+ listState2.add(MutableType.of(23));
+
+ listState3.add(MutableType.of(17));
+ listState3.add(MutableType.of(18));
+ listState3.add(MutableType.of(19));
+ listState3.add(MutableType.of(20));
+
+ BlockerCheckpointStreamFactory streamFactory = new BlockerCheckpointStreamFactory(1024 * 1024);
+
+ OneShotLatch waiterLatch = new OneShotLatch();
+ OneShotLatch blockerLatch = new OneShotLatch();
+
+ streamFactory.setWaiterLatch(waiterLatch);
+ streamFactory.setBlockerLatch(blockerLatch);
+
+ RunnableFuture<OperatorStateHandle> runnableFuture =
+ operatorStateBackend.snapshot(1, 1, streamFactory, CheckpointOptions.forFullCheckpoint());
+
+ ExecutorService executorService = Executors.newFixedThreadPool(1);
+
+ executorService.submit(runnableFuture);
+
+ // wait until the async checkpoint is in the write code, then continue
+ waiterLatch.await();
+
+ // do some mutations to the state, to test if our snapshot will NOT reflect them
+
+ listState1.add(MutableType.of(77));
+
+ int n = 0;
+
+ for (MutableType mutableType : listState2.get()) {
+ if (++n == 2) {
+ // allow the write code to continue, so that we could do changes while state is written in parallel.
+ blockerLatch.trigger();
+ }
+ mutableType.setValue(mutableType.getValue() + 10);
+ }
+
+ listState3.clear();
+
+ operatorStateBackend.getListState(
+ new ListStateDescriptor<>("test4", new JavaSerializer<MutableType>()));
+
+ // run the snapshot
+ OperatorStateHandle stateHandle = runnableFuture.get();
+
+ try {
+
+ operatorStateBackend.close();
+ operatorStateBackend.dispose();
+
+ AbstractStateBackend abstractStateBackend = new MemoryStateBackend(4096);
+
+ //TODO this is temporarily casted to test already functionality that we do not yet expose through public API
+ operatorStateBackend = (DefaultOperatorStateBackend) abstractStateBackend.createOperatorStateBackend(
+ createMockEnvironment(),
+ "testOperator");
+
+ operatorStateBackend.restore(Collections.singletonList(stateHandle));
+
+ assertEquals(3, operatorStateBackend.getRegisteredStateNames().size());
+
+ listState1 = operatorStateBackend.getListState(stateDescriptor1);
+ listState2 = operatorStateBackend.getListState(stateDescriptor2);
+ listState3 = operatorStateBackend.getUnionListState(stateDescriptor3);
+
+ assertEquals(3, operatorStateBackend.getRegisteredStateNames().size());
+
+ Iterator<MutableType> it = listState1.get().iterator();
+ assertEquals(42, it.next().value);
+ assertEquals(4711, it.next().value);
+ assertFalse(it.hasNext());
+
+ it = listState2.get().iterator();
+ assertEquals(7, it.next().value);
+ assertEquals(13, it.next().value);
+ assertEquals(23, it.next().value);
+ assertFalse(it.hasNext());
+
+ it = listState3.get().iterator();
+ assertEquals(17, it.next().value);
+ assertEquals(18, it.next().value);
+ assertEquals(19, it.next().value);
+ assertEquals(20, it.next().value);
+ assertFalse(it.hasNext());
operatorStateBackend.close();
operatorStateBackend.dispose();
} finally {
stateHandle.discardState();
}
+
+ executorService.shutdown();
+ }
+
+ @Test
+ public void testSnapshotAsyncClose() throws Exception {
+ DefaultOperatorStateBackend operatorStateBackend =
+ new DefaultOperatorStateBackend(OperatorStateBackendTest.class.getClassLoader(), new ExecutionConfig(), true);
+
+ ListStateDescriptor<MutableType> stateDescriptor1 =
+ new ListStateDescriptor<>("test1", new JavaSerializer<MutableType>());
+
+ ListState<MutableType> listState1 = operatorStateBackend.getOperatorState(stateDescriptor1);
+
+
+ listState1.add(MutableType.of(42));
+ listState1.add(MutableType.of(4711));
+
+ BlockerCheckpointStreamFactory streamFactory = new BlockerCheckpointStreamFactory(1024 * 1024);
+
+ OneShotLatch waiterLatch = new OneShotLatch();
+ OneShotLatch blockerLatch = new OneShotLatch();
+
+ streamFactory.setWaiterLatch(waiterLatch);
+ streamFactory.setBlockerLatch(blockerLatch);
+
+ RunnableFuture<OperatorStateHandle> runnableFuture =
+ operatorStateBackend.snapshot(1, 1, streamFactory, CheckpointOptions.forFullCheckpoint());
+
+ ExecutorService executorService = Executors.newFixedThreadPool(1);
+
+ executorService.submit(runnableFuture);
+
+ // wait until the async checkpoint is in the write code, then continue
+ waiterLatch.await();
+
+ operatorStateBackend.close();
+
+ blockerLatch.trigger();
+
+ try {
+ runnableFuture.get(60, TimeUnit.SECONDS);
+ Assert.fail();
+ } catch (ExecutionException eex) {
+ Assert.assertTrue(eex.getCause() instanceof IOException);
+ }
+ }
+
+ @Test
+ public void testSnapshotAsyncCancel() throws Exception {
+ DefaultOperatorStateBackend operatorStateBackend =
+ new DefaultOperatorStateBackend(OperatorStateBackendTest.class.getClassLoader(), new ExecutionConfig(), true);
+
+ ListStateDescriptor<MutableType> stateDescriptor1 =
+ new ListStateDescriptor<>("test1", new JavaSerializer<MutableType>());
+
+ ListState<MutableType> listState1 = operatorStateBackend.getOperatorState(stateDescriptor1);
+
+
+ listState1.add(MutableType.of(42));
+ listState1.add(MutableType.of(4711));
+
+ BlockerCheckpointStreamFactory streamFactory = new BlockerCheckpointStreamFactory(1024 * 1024);
+
+ OneShotLatch waiterLatch = new OneShotLatch();
+ OneShotLatch blockerLatch = new OneShotLatch();
+
+ streamFactory.setWaiterLatch(waiterLatch);
+ streamFactory.setBlockerLatch(blockerLatch);
+
+ RunnableFuture<OperatorStateHandle> runnableFuture =
+ operatorStateBackend.snapshot(1, 1, streamFactory, CheckpointOptions.forFullCheckpoint());
+
+ ExecutorService executorService = Executors.newFixedThreadPool(1);
+
+ executorService.submit(runnableFuture);
+
+ // wait until the async checkpoint is in the write code, then continue
+ waiterLatch.await();
+
+ runnableFuture.cancel(true);
+
+ blockerLatch.trigger();
+
+ try {
+ runnableFuture.get(60, TimeUnit.SECONDS);
+ Assert.fail();
+ } catch (CancellationException ignore) {
+
+ }
+ }
+
+ static final class MutableType implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private int value;
+
+ public MutableType() {
+ this(0);
+ }
+
+ public MutableType(int value) {
+ this.value = value;
+ }
+
+ public int getValue() {
+ return value;
+ }
+
+ public void setValue(int value) {
+ this.value = value;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+
+ if (this == o) {
+ return true;
+ }
+
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+
+ MutableType that = (MutableType) o;
+
+ return value == that.value;
+ }
+
+ @Override
+ public int hashCode() {
+ return value;
+ }
+
+ static MutableType of(int value) {
+ return new MutableType(value);
+ }
}
// ------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/6bdaf1e4/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/BlockingCheckpointsTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/BlockingCheckpointsTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/BlockingCheckpointsTest.java
index e266ea1..bf5be79 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/BlockingCheckpointsTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/BlockingCheckpointsTest.java
@@ -53,7 +53,9 @@ import org.apache.flink.runtime.state.AbstractKeyedStateBackend;
import org.apache.flink.runtime.state.AbstractStateBackend;
import org.apache.flink.runtime.state.CheckpointStreamFactory;
import org.apache.flink.runtime.state.CheckpointStreamFactory.CheckpointStateOutputStream;
+import org.apache.flink.runtime.state.DefaultOperatorStateBackend;
import org.apache.flink.runtime.state.KeyGroupRange;
+import org.apache.flink.runtime.state.OperatorStateBackend;
import org.apache.flink.runtime.state.OperatorStateCheckpointOutputStream;
import org.apache.flink.runtime.state.StateSnapshotContext;
import org.apache.flink.runtime.state.StreamStateHandle;
@@ -187,6 +189,14 @@ public class BlockingCheckpointsTest {
throw new UnsupportedOperationException();
}
+
+ @Override
+ public OperatorStateBackend createOperatorStateBackend(Environment env, String operatorIdentifier) throws Exception {
+ return new DefaultOperatorStateBackend(
+ getClass().getClassLoader(),
+ new ExecutionConfig(),
+ true);
+ }
}
private static final class LockingOutputStreamFactory implements CheckpointStreamFactory {
http://git-wip-us.apache.org/repos/asf/flink/blob/6bdaf1e4/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/StateBackendITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/StateBackendITCase.java b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/StateBackendITCase.java
index 4677242..ce342c0 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/StateBackendITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/StateBackendITCase.java
@@ -32,6 +32,7 @@ import org.apache.flink.runtime.state.AbstractKeyedStateBackend;
import org.apache.flink.runtime.state.AbstractStateBackend;
import org.apache.flink.runtime.state.CheckpointStreamFactory;
import org.apache.flink.runtime.state.KeyGroupRange;
+import org.apache.flink.runtime.state.OperatorStateBackend;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.util.StreamingMultipleProgramsTestBase;
import org.junit.Test;
@@ -112,6 +113,14 @@ public class StateBackendITCase extends StreamingMultipleProgramsTestBase {
TaskKvStateRegistry kvStateRegistry) throws IOException {
throw new SuccessException();
}
+
+ @Override
+ public OperatorStateBackend createOperatorStateBackend(
+ Environment env,
+ String operatorIdentifier) throws Exception {
+
+ throw new UnsupportedOperationException();
+ }
}
static final class SuccessException extends IOException {