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 2019/03/05 15:29:30 UTC
[flink] branch master updated (fb256d4 -> e8daa49)
This is an automated email from the ASF dual-hosted git repository.
srichter pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git.
from fb256d4 [FLINK-11803][table-planner-blink] Improve FlinkTypeFactory for Blink (#7898)
new 94f84a5 [FLINK-11731] [State Backends] Make DefaultOperatorStateBackend follow the builder pattern
new e8daa49 [FLINK-11796] [State Backends] Remove Snapshotable interface
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../flink/streaming/tests/StubStateBackend.java | 9 +-
.../runtime/state/AbstractKeyedStateBackend.java | 3 +-
.../flink/runtime/state/AbstractStateBackend.java | 6 +-
.../runtime/state/DefaultOperatorStateBackend.java | 492 +--------------------
.../state/DefaultOperatorStateBackendBuilder.java | 98 ++++
...efaultOperatorStateBackendSnapshotStrategy.java | 213 +++++++++
.../flink/runtime/state/OperatorStateBackend.java | 5 +-
.../state/OperatorStateRestoreOperation.java | 219 +++++++++
.../runtime/state/PartitionableListState.java | 135 ++++++
.../apache/flink/runtime/state/Snapshotable.java | 41 --
.../apache/flink/runtime/state/StateBackend.java | 8 +-
.../runtime/state/filesystem/FsStateBackend.java | 13 +-
.../runtime/state/heap/HeapKeyedStateBackend.java | 6 -
.../state/heap/HeapKeyedStateBackendBuilder.java | 1 +
.../runtime/state/memory/MemoryStateBackend.java | 19 +-
.../CheckpointSettingsSerializableTest.java | 6 +-
.../runtime/state/OperatorStateBackendTest.java | 114 +++--
.../state/StateBackendMigrationTestBase.java | 13 +-
.../state/ttl/mock/MockKeyedStateBackend.java | 7 -
.../runtime/state/ttl/mock/MockStateBackend.java | 7 +-
.../streaming/state/RocksDBKeyedStateBackend.java | 6 -
.../streaming/state/RocksDBStateBackend.java | 19 +-
.../api/operators/BackendRestorerProcedure.java | 23 +-
.../operators/StreamTaskStateInitializerImpl.java | 21 +-
.../operators/BackendRestorerProcedureTest.java | 10 +-
.../StreamTaskStateInitializerImplTest.java | 11 +-
.../runtime/tasks/StreamTaskTerminationTest.java | 6 +-
.../tasks/TaskCheckpointingBehaviourTest.java | 97 ++--
.../runtime/tasks/TestSpyWrapperStateBackend.java | 8 +-
.../test/streaming/runtime/StateBackendITCase.java | 5 +-
30 files changed, 968 insertions(+), 653 deletions(-)
create mode 100644 flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackendBuilder.java
create mode 100644 flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackendSnapshotStrategy.java
create mode 100644 flink-runtime/src/main/java/org/apache/flink/runtime/state/OperatorStateRestoreOperation.java
create mode 100644 flink-runtime/src/main/java/org/apache/flink/runtime/state/PartitionableListState.java
delete mode 100644 flink-runtime/src/main/java/org/apache/flink/runtime/state/Snapshotable.java
[flink] 02/02: [FLINK-11796] [State Backends] Remove Snapshotable
interface
Posted by sr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
srichter pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit e8daa49a593edc401cd44761b25b1324b11be4a6
Author: Yu Li <li...@apache.org>
AuthorDate: Sun Mar 3 03:00:43 2019 +0800
[FLINK-11796] [State Backends] Remove Snapshotable interface
---
.../runtime/state/AbstractKeyedStateBackend.java | 3 +-
.../runtime/state/DefaultOperatorStateBackend.java | 8 +----
.../flink/runtime/state/OperatorStateBackend.java | 5 ++-
.../apache/flink/runtime/state/Snapshotable.java | 41 ----------------------
.../runtime/state/heap/HeapKeyedStateBackend.java | 6 ----
.../state/ttl/mock/MockKeyedStateBackend.java | 7 ----
.../streaming/state/RocksDBKeyedStateBackend.java | 6 ----
.../api/operators/BackendRestorerProcedure.java | 23 ++----------
.../StreamTaskStateInitializerImplTest.java | 6 ----
9 files changed, 7 insertions(+), 98 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/AbstractKeyedStateBackend.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/AbstractKeyedStateBackend.java
index 98e21ff..e28aeef 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/AbstractKeyedStateBackend.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/AbstractKeyedStateBackend.java
@@ -36,7 +36,6 @@ import org.apache.flink.util.Preconditions;
import java.io.Closeable;
import java.io.IOException;
-import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Set;
@@ -52,7 +51,7 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
*/
public abstract class AbstractKeyedStateBackend<K> implements
KeyedStateBackend<K>,
- Snapshotable<SnapshotResult<KeyedStateHandle>, Collection<KeyedStateHandle>>,
+ SnapshotStrategy<SnapshotResult<KeyedStateHandle>>,
Closeable,
CheckpointListener {
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 f6a0dba..48c8eb8 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
@@ -39,7 +39,6 @@ import javax.annotation.Nonnull;
import java.io.IOException;
import java.io.Serializable;
-import java.util.Collection;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.RunnableFuture;
@@ -246,13 +245,8 @@ public class DefaultOperatorStateBackend implements OperatorStateBackend {
}
// -------------------------------------------------------------------------------------------
- // Snapshot and restore
+ // Snapshot
// -------------------------------------------------------------------------------------------
-
- public void restore(Collection<OperatorStateHandle> restoreSnapshots) throws Exception {
- // all restore work done in builder and nothing to do here
- }
-
@Nonnull
@Override
public RunnableFuture<SnapshotResult<OperatorStateHandle>> snapshot(
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/OperatorStateBackend.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/OperatorStateBackend.java
index 3cbb351..4fb8024 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/OperatorStateBackend.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/OperatorStateBackend.java
@@ -22,16 +22,15 @@ import org.apache.flink.api.common.state.OperatorStateStore;
import org.apache.flink.util.Disposable;
import java.io.Closeable;
-import java.util.Collection;
/**
* Interface that combines both, the user facing {@link OperatorStateStore} interface and the system interface
- * {@link Snapshotable}
+ * {@link SnapshotStrategy}
*
*/
public interface OperatorStateBackend extends
OperatorStateStore,
- Snapshotable<SnapshotResult<OperatorStateHandle>, Collection<OperatorStateHandle>>,
+ SnapshotStrategy<SnapshotResult<OperatorStateHandle>>,
Closeable,
Disposable {
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/Snapshotable.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/Snapshotable.java
deleted file mode 100644
index 1677855..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/Snapshotable.java
+++ /dev/null
@@ -1,41 +0,0 @@
-/*
- * 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.state;
-
-import org.apache.flink.annotation.Internal;
-
-import javax.annotation.Nullable;
-
-/**
- * Interface for operators that can perform snapshots of their state.
- *
- * @param <S> Generic type of the state object that is created as handle to snapshots.
- * @param <R> Generic type of the state object that used in restore.
- */
-@Internal
-public interface Snapshotable<S extends StateObject, R> extends SnapshotStrategy<S> {
-
- /**
- * Restores state that was previously snapshotted from the provided parameters. Typically the parameters are state
- * handles from which the old state is read.
- *
- * @param state the old state to restore.
- */
- void restore(@Nullable R state) throws Exception;
-}
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 a3a0c29..d3dee0d 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
@@ -62,7 +62,6 @@ import org.slf4j.LoggerFactory;
import javax.annotation.Nonnull;
import java.io.IOException;
-import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.concurrent.RunnableFuture;
@@ -304,11 +303,6 @@ public class HeapKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
return snapshotRunner;
}
- @SuppressWarnings("deprecation")
- public void restore(Collection<KeyedStateHandle> restoredState) {
- // all restore work done in builder and nothing to do here
- }
-
@Override
public void notifyCheckpointComplete(long checkpointId) {
//Nothing to do
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/mock/MockKeyedStateBackend.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/mock/MockKeyedStateBackend.java
index 5803c8e..7cee4c8 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/mock/MockKeyedStateBackend.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/mock/MockKeyedStateBackend.java
@@ -55,7 +55,6 @@ import org.apache.flink.util.FlinkRuntimeException;
import javax.annotation.Nonnull;
import java.util.ArrayList;
-import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -185,12 +184,6 @@ public class MockKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
SnapshotResult.of(new MockKeyedStateHandle<>(copy(stateValues, stateSnapshotFilters))));
}
- @SuppressWarnings("unchecked")
- @Override
- public void restore(Collection<KeyedStateHandle> state) {
- // all restore work done in builder and nothing to do here
- }
-
static <K> Map<String, Map<K, Map<Object, Object>>> copy(
Map<String, Map<K, Map<Object, Object>>> stateValues, Map<String, StateSnapshotTransformer<Object>> stateSnapshotFilters) {
Map<String, Map<K, Map<Object, Object>>> snapshotStates = new HashMap<>();
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
index cc0726c..c6d3863 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
@@ -78,7 +78,6 @@ import javax.annotation.Nonnull;
import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
-import java.util.Collection;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@@ -450,11 +449,6 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
}
@Override
- public void restore(Collection<KeyedStateHandle> restoreState) {
- // all restore work done in builder and nothing to do here
- }
-
- @Override
public void notifyCheckpointComplete(long completedCheckpointId) throws Exception {
if (checkpointSnapshotStrategy != null) {
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/BackendRestorerProcedure.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/BackendRestorerProcedure.java
index 81b2bb7..53e100e 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/BackendRestorerProcedure.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/BackendRestorerProcedure.java
@@ -19,7 +19,6 @@
package org.apache.flink.streaming.api.operators;
import org.apache.flink.core.fs.CloseableRegistry;
-import org.apache.flink.runtime.state.Snapshotable;
import org.apache.flink.runtime.state.StateObject;
import org.apache.flink.util.Disposable;
import org.apache.flink.util.ExceptionUtils;
@@ -33,7 +32,6 @@ import org.slf4j.LoggerFactory;
import javax.annotation.Nonnull;
import java.io.Closeable;
-import java.io.IOException;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
@@ -50,7 +48,7 @@ import java.util.List;
* @param <S> type of the supplied snapshots from which the backend restores.
*/
public class BackendRestorerProcedure<
- T extends Closeable & Disposable & Snapshotable<?, Collection<S>>,
+ T extends Closeable & Disposable,
S extends StateObject> {
/** Logger for this class. */
@@ -140,30 +138,15 @@ public class BackendRestorerProcedure<
private T attemptCreateAndRestore(Collection<S> restoreState) throws Exception {
- // create a new, empty backend.
+ // create a new backend with necessary initialization.
final T backendInstance = instanceSupplier.apply(restoreState);
try {
// register the backend with the registry to participate in task lifecycle w.r.t. cancellation.
backendCloseableRegistry.registerCloseable(backendInstance);
-
- // attempt to restore from snapshot (or null if no state was checkpointed).
- // TODO we could remove this invocation when moving all backend's restore into builder
- backendInstance.restore(restoreState);
-
return backendInstance;
} catch (Exception ex) {
-
- // under failure, we need do close...
- if (backendCloseableRegistry.unregisterCloseable(backendInstance)) {
- try {
- backendInstance.close();
- } catch (IOException closeEx) {
- ex = ExceptionUtils.firstOrSuppressed(closeEx, ex);
- }
- }
-
- // ... and dispose, e.g. to release native resources.
+ // dispose the backend, e.g. to release native resources, if failed to register it into registry.
try {
backendInstance.dispose();
} catch (Exception disposeEx) {
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/StreamTaskStateInitializerImplTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/StreamTaskStateInitializerImplTest.java
index 22f6441..430c057 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/StreamTaskStateInitializerImplTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/StreamTaskStateInitializerImplTest.java
@@ -66,10 +66,8 @@ import java.util.Collection;
import java.util.Collections;
import java.util.Random;
-import static org.mockito.Matchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
-import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
/**
@@ -223,10 +221,6 @@ public class StreamTaskStateInitializerImplTest {
Assert.assertNotNull(keyedStateInputs);
Assert.assertNotNull(operatorStateInputs);
- // check that the expected job manager state was restored
- verify(operatorStateBackend).restore(eq(operatorSubtaskState.getManagedOperatorState()));
- verify(keyedStateBackend).restore(eq(operatorSubtaskState.getManagedKeyedState()));
-
int count = 0;
for (KeyGroupStatePartitionStreamProvider keyedStateInput : keyedStateInputs) {
++count;
[flink] 01/02: [FLINK-11731] [State Backends] Make
DefaultOperatorStateBackend follow the builder pattern
Posted by sr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
srichter pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 94f84a5c7b4876dee5b34e6a3725da7ee06a607f
Author: Yu Li <li...@apache.org>
AuthorDate: Sat Mar 2 18:19:21 2019 +0800
[FLINK-11731] [State Backends] Make DefaultOperatorStateBackend follow the builder pattern
This closes #7899.
---
.../flink/streaming/tests/StubStateBackend.java | 9 +-
.../flink/runtime/state/AbstractStateBackend.java | 6 +-
.../runtime/state/DefaultOperatorStateBackend.java | 486 +--------------------
.../state/DefaultOperatorStateBackendBuilder.java | 98 +++++
...efaultOperatorStateBackendSnapshotStrategy.java | 213 +++++++++
.../state/OperatorStateRestoreOperation.java | 219 ++++++++++
.../runtime/state/PartitionableListState.java | 135 ++++++
.../apache/flink/runtime/state/StateBackend.java | 8 +-
.../runtime/state/filesystem/FsStateBackend.java | 13 +-
.../state/heap/HeapKeyedStateBackendBuilder.java | 1 +
.../runtime/state/memory/MemoryStateBackend.java | 19 +-
.../CheckpointSettingsSerializableTest.java | 6 +-
.../runtime/state/OperatorStateBackendTest.java | 114 +++--
.../state/StateBackendMigrationTestBase.java | 13 +-
.../runtime/state/ttl/mock/MockStateBackend.java | 7 +-
.../streaming/state/RocksDBStateBackend.java | 19 +-
.../operators/StreamTaskStateInitializerImpl.java | 21 +-
.../operators/BackendRestorerProcedureTest.java | 10 +-
.../StreamTaskStateInitializerImplTest.java | 5 +-
.../runtime/tasks/StreamTaskTerminationTest.java | 6 +-
.../tasks/TaskCheckpointingBehaviourTest.java | 97 ++--
.../runtime/tasks/TestSpyWrapperStateBackend.java | 8 +-
.../test/streaming/runtime/StateBackendITCase.java | 5 +-
23 files changed, 962 insertions(+), 556 deletions(-)
diff --git a/flink-end-to-end-tests/flink-stream-state-ttl-test/src/main/java/org/apache/flink/streaming/tests/StubStateBackend.java b/flink-end-to-end-tests/flink-stream-state-ttl-test/src/main/java/org/apache/flink/streaming/tests/StubStateBackend.java
index dec4f2d..0d0c895 100644
--- a/flink-end-to-end-tests/flink-stream-state-ttl-test/src/main/java/org/apache/flink/streaming/tests/StubStateBackend.java
+++ b/flink-end-to-end-tests/flink-stream-state-ttl-test/src/main/java/org/apache/flink/streaming/tests/StubStateBackend.java
@@ -30,6 +30,7 @@ import org.apache.flink.runtime.state.CompletedCheckpointStorageLocation;
import org.apache.flink.runtime.state.KeyGroupRange;
import org.apache.flink.runtime.state.KeyedStateHandle;
import org.apache.flink.runtime.state.OperatorStateBackend;
+import org.apache.flink.runtime.state.OperatorStateHandle;
import org.apache.flink.runtime.state.StateBackend;
import org.apache.flink.runtime.state.ttl.TtlTimeProvider;
@@ -96,7 +97,11 @@ final class StubStateBackend implements StateBackend {
}
@Override
- public OperatorStateBackend createOperatorStateBackend(Environment env, String operatorIdentifier) throws Exception {
- return backend.createOperatorStateBackend(env, operatorIdentifier);
+ public OperatorStateBackend createOperatorStateBackend(
+ Environment env,
+ String operatorIdentifier,
+ @Nonnull Collection<OperatorStateHandle> stateHandles,
+ CloseableRegistry cancelStreamRegistry) throws Exception {
+ return backend.createOperatorStateBackend(env, operatorIdentifier, stateHandles, cancelStreamRegistry);
}
}
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 2343d83..8d94898 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
@@ -71,6 +71,8 @@ public abstract class AbstractStateBackend implements StateBackend, java.io.Seri
@Override
public abstract OperatorStateBackend createOperatorStateBackend(
- Environment env,
- String operatorIdentifier) throws Exception;
+ Environment env,
+ String operatorIdentifier,
+ @Nonnull Collection<OperatorStateHandle> stateHandles,
+ CloseableRegistry cancelStreamRegistry) throws Exception;
}
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 a3541b6..f6a0dba 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
@@ -18,6 +18,7 @@
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.BroadcastState;
@@ -26,20 +27,11 @@ import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.common.typeutils.TypeSerializerSchemaCompatibility;
-import org.apache.flink.api.common.typeutils.UnloadableDummyTypeSerializer;
import org.apache.flink.core.fs.CloseableRegistry;
-import org.apache.flink.core.fs.FSDataInputStream;
-import org.apache.flink.core.fs.FSDataOutputStream;
-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.CheckpointOptions;
-import org.apache.flink.runtime.state.metainfo.StateMetaInfoSnapshot;
import org.apache.flink.util.Preconditions;
import org.apache.flink.util.StateMigrationException;
-import org.apache.commons.io.IOUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -47,13 +39,9 @@ import javax.annotation.Nonnull;
import java.io.IOException;
import java.io.Serializable;
-import java.util.ArrayList;
import java.util.Collection;
-import java.util.HashMap;
-import java.util.List;
import java.util.Map;
import java.util.Set;
-import java.util.concurrent.FutureTask;
import java.util.concurrent.RunnableFuture;
/**
@@ -85,14 +73,9 @@ public class DefaultOperatorStateBackend implements OperatorStateBackend {
private final CloseableRegistry closeStreamOnCancelRegistry;
/**
- * Default serializer. Only used for the default operator state.
- */
- private final JavaSerializer<Serializable> javaSerializer;
-
- /**
- * The user code classloader.
+ * Default typeSerializer. Only used for the default operator state.
*/
- private final ClassLoader userClassloader;
+ private final JavaSerializer<Serializable> deprecatedDefaultJavaSerializer = new JavaSerializer<>();
/**
* The execution configuration.
@@ -100,11 +83,6 @@ public class DefaultOperatorStateBackend implements OperatorStateBackend {
private final ExecutionConfig executionConfig;
/**
- * Flag to de/activate asynchronous snapshots.
- */
- private final boolean asynchronousSnapshots;
-
- /**
* Cache of already accessed states.
*
* <p>In contrast to {@link #registeredOperatorStates} which may be repopulated
@@ -114,27 +92,27 @@ public class DefaultOperatorStateBackend implements OperatorStateBackend {
*
* @see <a href="https://issues.apache.org/jira/browse/FLINK-6849">FLINK-6849</a>
*/
- private final HashMap<String, PartitionableListState<?>> accessedStatesByName;
+ private final Map<String, PartitionableListState<?>> accessedStatesByName;
private final Map<String, BackendWritableBroadcastState<?, ?>> accessedBroadcastStatesByName;
private final AbstractSnapshotStrategy<OperatorStateHandle> snapshotStrategy;
public DefaultOperatorStateBackend(
- ClassLoader userClassLoader,
ExecutionConfig executionConfig,
- boolean asynchronousSnapshots) {
-
- this.closeStreamOnCancelRegistry = new CloseableRegistry();
- this.userClassloader = Preconditions.checkNotNull(userClassLoader);
+ CloseableRegistry closeStreamOnCancelRegistry,
+ Map<String, PartitionableListState<?>> registeredOperatorStates,
+ Map<String, BackendWritableBroadcastState<?, ?>> registeredBroadcastStates,
+ Map<String, PartitionableListState<?>> accessedStatesByName,
+ Map<String, BackendWritableBroadcastState<?, ?>> accessedBroadcastStatesByName,
+ AbstractSnapshotStrategy<OperatorStateHandle> snapshotStrategy) {
+ this.closeStreamOnCancelRegistry = closeStreamOnCancelRegistry;
this.executionConfig = executionConfig;
- this.javaSerializer = new JavaSerializer<>();
- this.registeredOperatorStates = new HashMap<>();
- this.registeredBroadcastStates = new HashMap<>();
- this.asynchronousSnapshots = asynchronousSnapshots;
- this.accessedStatesByName = new HashMap<>();
- this.accessedBroadcastStatesByName = new HashMap<>();
- this.snapshotStrategy = new DefaultOperatorStateBackendSnapshotStrategy();
+ this.registeredOperatorStates = registeredOperatorStates;
+ this.registeredBroadcastStates = registeredBroadcastStates;
+ this.accessedStatesByName = accessedStatesByName;
+ this.accessedBroadcastStatesByName = accessedBroadcastStatesByName;
+ this.snapshotStrategy = snapshotStrategy;
}
public ExecutionConfig getExecutionConfig() {
@@ -216,13 +194,13 @@ public class DefaultOperatorStateBackend implements OperatorStateBackend {
TypeSerializerSchemaCompatibility<K> keyCompatibility =
restoredBroadcastStateMetaInfo.updateKeySerializer(broadcastStateKeySerializer);
if (keyCompatibility.isIncompatible()) {
- throw new StateMigrationException("The new key serializer for broadcast state must not be incompatible.");
+ throw new StateMigrationException("The new key typeSerializer for broadcast state must not be incompatible.");
}
TypeSerializerSchemaCompatibility<V> valueCompatibility =
restoredBroadcastStateMetaInfo.updateValueSerializer(broadcastStateValueSerializer);
if (valueCompatibility.isIncompatible()) {
- throw new StateMigrationException("The new value serializer for broadcast state must not be incompatible.");
+ throw new StateMigrationException("The new value typeSerializer for broadcast state must not be incompatible.");
}
broadcastState.setStateMetaInfo(restoredBroadcastStateMetaInfo);
@@ -264,7 +242,7 @@ public class DefaultOperatorStateBackend implements OperatorStateBackend {
@Deprecated
@Override
public <T extends Serializable> ListState<T> getSerializableListState(String stateName) throws Exception {
- return (ListState<T>) getListState(new ListStateDescriptor<>(stateName, javaSerializer));
+ return (ListState<T>) getListState(new ListStateDescriptor<>(stateName, deprecatedDefaultJavaSerializer));
}
// -------------------------------------------------------------------------------------------
@@ -272,120 +250,7 @@ public class DefaultOperatorStateBackend implements OperatorStateBackend {
// -------------------------------------------------------------------------------------------
public void restore(Collection<OperatorStateHandle> restoreSnapshots) throws Exception {
-
- if (null == restoreSnapshots || restoreSnapshots.isEmpty()) {
- return;
- }
-
- for (OperatorStateHandle stateHandle : restoreSnapshots) {
-
- if (stateHandle == null) {
- continue;
- }
-
- FSDataInputStream in = stateHandle.openInputStream();
- closeStreamOnCancelRegistry.registerCloseable(in);
-
- ClassLoader restoreClassLoader = Thread.currentThread().getContextClassLoader();
-
- try {
- Thread.currentThread().setContextClassLoader(userClassloader);
- OperatorBackendSerializationProxy backendSerializationProxy =
- new OperatorBackendSerializationProxy(userClassloader);
-
- backendSerializationProxy.read(new DataInputViewStreamWrapper(in));
-
- List<StateMetaInfoSnapshot> restoredOperatorMetaInfoSnapshots =
- backendSerializationProxy.getOperatorStateMetaInfoSnapshots();
-
- // Recreate all PartitionableListStates from the meta info
- for (StateMetaInfoSnapshot restoredSnapshot : restoredOperatorMetaInfoSnapshots) {
-
- final RegisteredOperatorStateBackendMetaInfo<?> restoredMetaInfo =
- new RegisteredOperatorStateBackendMetaInfo<>(restoredSnapshot);
-
- if (restoredMetaInfo.getPartitionStateSerializer() instanceof UnloadableDummyTypeSerializer) {
-
- // must fail now if the previous serializer cannot be restored because there is no serializer
- // capable of reading previous state
- // TODO when eager state registration is in place, we can try to get a convert deserializer
- // TODO from the newly registered serializer instead of simply failing here
-
- throw new IOException("Unable to restore operator state [" + restoredSnapshot.getName() + "]." +
- " The previous serializer of the operator state must be present; the serializer could" +
- " have been removed from the classpath, or its implementation have changed and could" +
- " not be loaded. This is a temporary restriction that will be fixed in future versions.");
- }
-
- PartitionableListState<?> listState = registeredOperatorStates.get(restoredSnapshot.getName());
-
- if (null == listState) {
- listState = new PartitionableListState<>(restoredMetaInfo);
-
- registeredOperatorStates.put(listState.getStateMetaInfo().getName(), listState);
- } else {
- // TODO with eager state registration in place, check here for serializer migration strategies
- }
- }
-
- // ... and then get back the broadcast state.
- List<StateMetaInfoSnapshot> restoredBroadcastMetaInfoSnapshots =
- backendSerializationProxy.getBroadcastStateMetaInfoSnapshots();
-
- for (StateMetaInfoSnapshot restoredSnapshot : restoredBroadcastMetaInfoSnapshots) {
-
- final RegisteredBroadcastStateBackendMetaInfo<?, ?> restoredMetaInfo =
- new RegisteredBroadcastStateBackendMetaInfo<>(restoredSnapshot);
-
- if (restoredMetaInfo.getKeySerializer() instanceof UnloadableDummyTypeSerializer ||
- restoredMetaInfo.getValueSerializer() instanceof UnloadableDummyTypeSerializer) {
-
- // must fail now if the previous serializer cannot be restored because there is no serializer
- // capable of reading previous state
- // TODO when eager state registration is in place, we can try to get a convert deserializer
- // TODO from the newly registered serializer instead of simply failing here
-
- throw new IOException("Unable to restore broadcast state [" + restoredSnapshot.getName() + "]." +
- " The previous key and value serializers of the state must be present; the serializers could" +
- " have been removed from the classpath, or their implementations have changed and could" +
- " not be loaded. This is a temporary restriction that will be fixed in future versions.");
- }
-
- BackendWritableBroadcastState<? ,?> broadcastState = registeredBroadcastStates.get(restoredSnapshot.getName());
-
- if (broadcastState == null) {
- broadcastState = new HeapBroadcastState<>(restoredMetaInfo);
-
- registeredBroadcastStates.put(broadcastState.getStateMetaInfo().getName(), broadcastState);
- } else {
- // TODO with eager state registration in place, check here for serializer migration strategies
- }
- }
-
- // Restore all the states
- for (Map.Entry<String, OperatorStateHandle.StateMetaInfo> nameToOffsets :
- stateHandle.getStateNameToPartitionOffsets().entrySet()) {
-
- final String stateName = nameToOffsets.getKey();
-
- PartitionableListState<?> listStateForName = registeredOperatorStates.get(stateName);
- if (listStateForName == null) {
- BackendWritableBroadcastState<?, ?> broadcastStateForName = registeredBroadcastStates.get(stateName);
- Preconditions.checkState(broadcastStateForName != null, "Found state without " +
- "corresponding meta info: " + stateName);
- deserializeBroadcastStateValues(broadcastStateForName, in, nameToOffsets.getValue());
- } else {
- deserializeOperatorStateValues(listStateForName, in, nameToOffsets.getValue());
- }
- }
-
- } finally {
- Thread.currentThread().setContextClassLoader(restoreClassLoader);
- if (closeStreamOnCancelRegistry.unregisterCloseable(in)) {
- IOUtils.closeQuietly(in);
- }
- }
- }
+ // all restore work done in builder and nothing to do here
}
@Nonnull
@@ -405,112 +270,6 @@ public class DefaultOperatorStateBackend implements OperatorStateBackend {
return snapshotRunner;
}
- /**
- * Implementation of operator list state.
- *
- * @param <S> the type of an operator state partition.
- */
- static final class PartitionableListState<S> implements ListState<S> {
-
- /**
- * Meta information of the state, including state name, assignment mode, and serializer
- */
- private RegisteredOperatorStateBackendMetaInfo<S> stateMetaInfo;
-
- /**
- * The internal list the holds the elements of the state
- */
- private final ArrayList<S> internalList;
-
- /**
- * A serializer that allows to perform deep copies of internalList
- */
- private final ArrayListSerializer<S> internalListCopySerializer;
-
- PartitionableListState(RegisteredOperatorStateBackendMetaInfo<S> stateMetaInfo) {
- this(stateMetaInfo, new ArrayList<S>());
- }
-
- private PartitionableListState(
- RegisteredOperatorStateBackendMetaInfo<S> stateMetaInfo,
- ArrayList<S> internalList) {
-
- this.stateMetaInfo = Preconditions.checkNotNull(stateMetaInfo);
- this.internalList = Preconditions.checkNotNull(internalList);
- this.internalListCopySerializer = new ArrayListSerializer<>(stateMetaInfo.getPartitionStateSerializer());
- }
-
- private PartitionableListState(PartitionableListState<S> toCopy) {
-
- this(toCopy.stateMetaInfo.deepCopy(), toCopy.internalListCopySerializer.copy(toCopy.internalList));
- }
-
- public void setStateMetaInfo(RegisteredOperatorStateBackendMetaInfo<S> stateMetaInfo) {
- this.stateMetaInfo = stateMetaInfo;
- }
-
- public RegisteredOperatorStateBackendMetaInfo<S> getStateMetaInfo() {
- return stateMetaInfo;
- }
-
- public PartitionableListState<S> deepCopy() {
- return new PartitionableListState<>(this);
- }
-
- @Override
- public void clear() {
- internalList.clear();
- }
-
- @Override
- public Iterable<S> get() {
- return internalList;
- }
-
- @Override
- public void add(S value) {
- Preconditions.checkNotNull(value, "You cannot add null to a ListState.");
- internalList.add(value);
- }
-
- @Override
- public String toString() {
- return "PartitionableListState{" +
- "stateMetaInfo=" + stateMetaInfo +
- ", internalList=" + internalList +
- '}';
- }
-
- public long[] write(FSDataOutputStream out) throws IOException {
-
- long[] partitionOffsets = new long[internalList.size()];
-
- DataOutputView dov = new DataOutputViewStreamWrapper(out);
-
- for (int i = 0; i < internalList.size(); ++i) {
- S element = internalList.get(i);
- partitionOffsets[i] = out.getPos();
- getStateMetaInfo().getPartitionStateSerializer().serialize(element, dov);
- }
-
- return partitionOffsets;
- }
-
- @Override
- public void update(List<S> values) {
- internalList.clear();
-
- addAll(values);
- }
-
- @Override
- public void addAll(List<S> values) {
- if (values != null && !values.isEmpty()) {
- internalList.addAll(values);
- }
- }
- }
-
private <S> ListState<S> getListState(
ListStateDescriptor<S> stateDescriptor,
OperatorStateHandle.Mode mode) throws StateMigrationException {
@@ -567,7 +326,7 @@ public class DefaultOperatorStateBackend implements OperatorStateBackend {
TypeSerializerSchemaCompatibility<S> stateCompatibility =
restoredPartitionableListStateMetaInfo.updatePartitionStateSerializer(newPartitionStateSerializer);
if (stateCompatibility.isIncompatible()) {
- throw new StateMigrationException("The new state serializer for operator state must not be incompatible.");
+ throw new StateMigrationException("The new state typeSerializer for operator state must not be incompatible.");
}
partitionableListState.setStateMetaInfo(restoredPartitionableListStateMetaInfo);
@@ -577,47 +336,6 @@ public class DefaultOperatorStateBackend implements OperatorStateBackend {
return partitionableListState;
}
- private static <S> void deserializeOperatorStateValues(
- PartitionableListState<S> stateListForName,
- FSDataInputStream in,
- OperatorStateHandle.StateMetaInfo metaInfo) throws IOException {
-
- if (null != metaInfo) {
- long[] offsets = metaInfo.getOffsets();
- if (null != offsets) {
- DataInputView div = new DataInputViewStreamWrapper(in);
- TypeSerializer<S> serializer = stateListForName.getStateMetaInfo().getPartitionStateSerializer();
- for (long offset : offsets) {
- in.seek(offset);
- stateListForName.add(serializer.deserialize(div));
- }
- }
- }
- }
-
- private static <K, V> void deserializeBroadcastStateValues(
- final BackendWritableBroadcastState<K, V> broadcastStateForName,
- final FSDataInputStream in,
- final OperatorStateHandle.StateMetaInfo metaInfo) throws Exception {
-
- if (metaInfo != null) {
- long[] offsets = metaInfo.getOffsets();
- if (offsets != null) {
-
- TypeSerializer<K> keySerializer = broadcastStateForName.getStateMetaInfo().getKeySerializer();
- TypeSerializer<V> valueSerializer = broadcastStateForName.getStateMetaInfo().getValueSerializer();
-
- in.seek(offsets[0]);
-
- DataInputView div = new DataInputViewStreamWrapper(in);
- int size = div.readInt();
- for (int i = 0; i < size; i++) {
- broadcastStateForName.put(keySerializer.deserialize(div), valueSerializer.deserialize(div));
- }
- }
- }
- }
-
private static void checkStateNameAndMode(
String actualName,
String expectedName,
@@ -637,166 +355,4 @@ public class DefaultOperatorStateBackend implements OperatorStateBackend {
"registered with [" + expectedMode + "].");
}
- /**
- * Snapshot strategy for this backend.
- */
- private class DefaultOperatorStateBackendSnapshotStrategy extends AbstractSnapshotStrategy<OperatorStateHandle> {
-
- protected DefaultOperatorStateBackendSnapshotStrategy() {
- super("DefaultOperatorStateBackend snapshot");
- }
-
- @Nonnull
- @Override
- public RunnableFuture<SnapshotResult<OperatorStateHandle>> snapshot(
- final long checkpointId,
- final long timestamp,
- @Nonnull final CheckpointStreamFactory streamFactory,
- @Nonnull final CheckpointOptions checkpointOptions) throws IOException {
-
- if (registeredOperatorStates.isEmpty() && registeredBroadcastStates.isEmpty()) {
- return DoneFuture.of(SnapshotResult.empty());
- }
-
- final Map<String, PartitionableListState<?>> registeredOperatorStatesDeepCopies =
- new HashMap<>(registeredOperatorStates.size());
- final Map<String, BackendWritableBroadcastState<?, ?>> registeredBroadcastStatesDeepCopies =
- new HashMap<>(registeredBroadcastStates.size());
-
- ClassLoader snapshotClassLoader = Thread.currentThread().getContextClassLoader();
- Thread.currentThread().setContextClassLoader(userClassloader);
- try {
- // eagerly create deep copies of the list and the broadcast states (if any)
- // in the synchronous phase, so that we can use them in the async writing.
-
- if (!registeredOperatorStates.isEmpty()) {
- for (Map.Entry<String, PartitionableListState<?>> entry : registeredOperatorStates.entrySet()) {
- PartitionableListState<?> listState = entry.getValue();
- if (null != listState) {
- listState = listState.deepCopy();
- }
- registeredOperatorStatesDeepCopies.put(entry.getKey(), listState);
- }
- }
-
- if (!registeredBroadcastStates.isEmpty()) {
- for (Map.Entry<String, BackendWritableBroadcastState<?, ?>> entry : registeredBroadcastStates.entrySet()) {
- BackendWritableBroadcastState<?, ?> broadcastState = entry.getValue();
- if (null != broadcastState) {
- broadcastState = broadcastState.deepCopy();
- }
- registeredBroadcastStatesDeepCopies.put(entry.getKey(), broadcastState);
- }
- }
- } finally {
- Thread.currentThread().setContextClassLoader(snapshotClassLoader);
- }
-
- AsyncSnapshotCallable<SnapshotResult<OperatorStateHandle>> snapshotCallable =
- new AsyncSnapshotCallable<SnapshotResult<OperatorStateHandle>>() {
-
- @Override
- protected SnapshotResult<OperatorStateHandle> callInternal() throws Exception {
-
- CheckpointStreamFactory.CheckpointStateOutputStream localOut =
- streamFactory.createCheckpointStateOutputStream(CheckpointedStateScope.EXCLUSIVE);
- snapshotCloseableRegistry.registerCloseable(localOut);
-
- // get the registered operator state infos ...
- List<StateMetaInfoSnapshot> operatorMetaInfoSnapshots =
- new ArrayList<>(registeredOperatorStatesDeepCopies.size());
-
- for (Map.Entry<String, PartitionableListState<?>> entry :
- registeredOperatorStatesDeepCopies.entrySet()) {
- operatorMetaInfoSnapshots.add(entry.getValue().getStateMetaInfo().snapshot());
- }
-
- // ... get the registered broadcast operator state infos ...
- List<StateMetaInfoSnapshot> broadcastMetaInfoSnapshots =
- new ArrayList<>(registeredBroadcastStatesDeepCopies.size());
-
- for (Map.Entry<String, BackendWritableBroadcastState<?, ?>> entry :
- registeredBroadcastStatesDeepCopies.entrySet()) {
- broadcastMetaInfoSnapshots.add(entry.getValue().getStateMetaInfo().snapshot());
- }
-
- // ... write them all in the checkpoint stream ...
- DataOutputView dov = new DataOutputViewStreamWrapper(localOut);
-
- OperatorBackendSerializationProxy backendSerializationProxy =
- new OperatorBackendSerializationProxy(operatorMetaInfoSnapshots, broadcastMetaInfoSnapshots);
-
- backendSerializationProxy.write(dov);
-
- // ... and then go for the states ...
-
- // we put BOTH normal and broadcast state metadata here
- int initialMapCapacity =
- registeredOperatorStatesDeepCopies.size() + registeredBroadcastStatesDeepCopies.size();
- final Map<String, OperatorStateHandle.StateMetaInfo> writtenStatesMetaData =
- new HashMap<>(initialMapCapacity);
-
- for (Map.Entry<String, PartitionableListState<?>> entry :
- registeredOperatorStatesDeepCopies.entrySet()) {
-
- PartitionableListState<?> value = entry.getValue();
- long[] partitionOffsets = value.write(localOut);
- OperatorStateHandle.Mode mode = value.getStateMetaInfo().getAssignmentMode();
- writtenStatesMetaData.put(
- entry.getKey(),
- new OperatorStateHandle.StateMetaInfo(partitionOffsets, mode));
- }
-
- // ... and the broadcast states themselves ...
- for (Map.Entry<String, BackendWritableBroadcastState<?, ?>> entry :
- registeredBroadcastStatesDeepCopies.entrySet()) {
-
- BackendWritableBroadcastState<?, ?> value = entry.getValue();
- long[] partitionOffsets = {value.write(localOut)};
- OperatorStateHandle.Mode mode = value.getStateMetaInfo().getAssignmentMode();
- writtenStatesMetaData.put(
- entry.getKey(),
- new OperatorStateHandle.StateMetaInfo(partitionOffsets, mode));
- }
-
- // ... and, finally, create the state handle.
- OperatorStateHandle retValue = null;
-
- if (snapshotCloseableRegistry.unregisterCloseable(localOut)) {
-
- StreamStateHandle stateHandle = localOut.closeAndGetHandle();
-
- if (stateHandle != null) {
- retValue = new OperatorStreamStateHandle(writtenStatesMetaData, stateHandle);
- }
-
- return SnapshotResult.of(retValue);
- } else {
- throw new IOException("Stream was already unregistered.");
- }
- }
-
- @Override
- protected void cleanupProvidedResources() {
- // nothing to do
- }
-
- @Override
- protected void logAsyncSnapshotComplete(long startTime) {
- if (asynchronousSnapshots) {
- logAsyncCompleted(streamFactory, startTime);
- }
- }
- };
-
- final FutureTask<SnapshotResult<OperatorStateHandle>> task =
- snapshotCallable.toAsyncSnapshotFutureTask(closeStreamOnCancelRegistry);
-
- if (!asynchronousSnapshots) {
- task.run();
- }
-
- return task;
- }
- }
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackendBuilder.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackendBuilder.java
new file mode 100644
index 0000000..355dc3e
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackendBuilder.java
@@ -0,0 +1,98 @@
+/*
+ * 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.state;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.core.fs.CloseableRegistry;
+import org.apache.flink.util.IOUtils;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Builder class for {@link DefaultOperatorStateBackend} which handles all necessary initializations and clean ups.
+ */
+public class DefaultOperatorStateBackendBuilder implements
+ StateBackendBuilder<DefaultOperatorStateBackend, BackendBuildingException> {
+ /** The user code classloader. */
+ @VisibleForTesting
+ protected final ClassLoader userClassloader;
+ /** The execution configuration. */
+ @VisibleForTesting
+ protected final ExecutionConfig executionConfig;
+ /** Flag to de/activate asynchronous snapshots. */
+ @VisibleForTesting
+ protected final boolean asynchronousSnapshots;
+ /** State handles for restore. */
+ @VisibleForTesting
+ protected final Collection<OperatorStateHandle> restoreStateHandles;
+ @VisibleForTesting
+ protected final CloseableRegistry cancelStreamRegistry;
+
+
+ public DefaultOperatorStateBackendBuilder(
+ ClassLoader userClassloader,
+ ExecutionConfig executionConfig,
+ boolean asynchronousSnapshots,
+ Collection<OperatorStateHandle> stateHandles,
+ CloseableRegistry cancelStreamRegistry) {
+ this.userClassloader = userClassloader;
+ this.executionConfig = executionConfig;
+ this.asynchronousSnapshots = asynchronousSnapshots;
+ this.restoreStateHandles = stateHandles;
+ this.cancelStreamRegistry = cancelStreamRegistry;
+ }
+
+ @Override
+ public DefaultOperatorStateBackend build() throws BackendBuildingException {
+ Map<String, PartitionableListState<?>> registeredOperatorStates = new HashMap<>();
+ Map<String, BackendWritableBroadcastState<?, ?>> registeredBroadcastStates = new HashMap<>();
+ CloseableRegistry cancelStreamRegistryForBackend = new CloseableRegistry();
+ AbstractSnapshotStrategy<OperatorStateHandle> snapshotStrategy =
+ new DefaultOperatorStateBackendSnapshotStrategy(
+ userClassloader,
+ asynchronousSnapshots,
+ registeredOperatorStates,
+ registeredBroadcastStates,
+ cancelStreamRegistryForBackend);
+ OperatorStateRestoreOperation restoreOperation = new OperatorStateRestoreOperation(
+ cancelStreamRegistry,
+ userClassloader,
+ registeredOperatorStates,
+ registeredBroadcastStates,
+ restoreStateHandles
+ );
+ try {
+ restoreOperation.restore();
+ } catch (Exception e) {
+ IOUtils.closeQuietly(cancelStreamRegistryForBackend);
+ throw new BackendBuildingException("Failed when trying to restore operator state backend", e);
+ }
+ return new DefaultOperatorStateBackend(
+ executionConfig,
+ cancelStreamRegistryForBackend,
+ registeredOperatorStates,
+ registeredBroadcastStates,
+ new HashMap<>(),
+ new HashMap<>(),
+ snapshotStrategy
+ );
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackendSnapshotStrategy.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackendSnapshotStrategy.java
new file mode 100644
index 0000000..2948620
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackendSnapshotStrategy.java
@@ -0,0 +1,213 @@
+/*
+ * 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.state;
+
+import org.apache.flink.core.fs.CloseableRegistry;
+import org.apache.flink.core.memory.DataOutputView;
+import org.apache.flink.core.memory.DataOutputViewStreamWrapper;
+import org.apache.flink.runtime.checkpoint.CheckpointOptions;
+import org.apache.flink.runtime.state.metainfo.StateMetaInfoSnapshot;
+
+import javax.annotation.Nonnull;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.FutureTask;
+import java.util.concurrent.RunnableFuture;
+
+/**
+ * Snapshot strategy for this backend.
+ */
+class DefaultOperatorStateBackendSnapshotStrategy extends AbstractSnapshotStrategy<OperatorStateHandle> {
+ private final ClassLoader userClassLoader;
+ private final boolean asynchronousSnapshots;
+ private final Map<String, PartitionableListState<?>> registeredOperatorStates;
+ private final Map<String, BackendWritableBroadcastState<?, ?>> registeredBroadcastStates;
+ private final CloseableRegistry closeStreamOnCancelRegistry;
+
+ protected DefaultOperatorStateBackendSnapshotStrategy(
+ ClassLoader userClassLoader,
+ boolean asynchronousSnapshots,
+ Map<String, PartitionableListState<?>> registeredOperatorStates,
+ Map<String, BackendWritableBroadcastState<?, ?>> registeredBroadcastStates,
+ CloseableRegistry closeStreamOnCancelRegistry) {
+ super("DefaultOperatorStateBackend snapshot");
+ this.userClassLoader = userClassLoader;
+ this.asynchronousSnapshots = asynchronousSnapshots;
+ this.registeredOperatorStates = registeredOperatorStates;
+ this.registeredBroadcastStates = registeredBroadcastStates;
+ this.closeStreamOnCancelRegistry = closeStreamOnCancelRegistry;
+ }
+
+ @Nonnull
+ @Override
+ public RunnableFuture<SnapshotResult<OperatorStateHandle>> snapshot(
+ final long checkpointId,
+ final long timestamp,
+ @Nonnull final CheckpointStreamFactory streamFactory,
+ @Nonnull final CheckpointOptions checkpointOptions) throws IOException {
+
+ if (registeredOperatorStates.isEmpty() && registeredBroadcastStates.isEmpty()) {
+ return DoneFuture.of(SnapshotResult.empty());
+ }
+
+ final Map<String, PartitionableListState<?>> registeredOperatorStatesDeepCopies =
+ new HashMap<>(registeredOperatorStates.size());
+ final Map<String, BackendWritableBroadcastState<?, ?>> registeredBroadcastStatesDeepCopies =
+ new HashMap<>(registeredBroadcastStates.size());
+
+ ClassLoader snapshotClassLoader = Thread.currentThread().getContextClassLoader();
+ Thread.currentThread().setContextClassLoader(userClassLoader);
+ try {
+ // eagerly create deep copies of the list and the broadcast states (if any)
+ // in the synchronous phase, so that we can use them in the async writing.
+
+ if (!registeredOperatorStates.isEmpty()) {
+ for (Map.Entry<String, PartitionableListState<?>> entry : registeredOperatorStates.entrySet()) {
+ PartitionableListState<?> listState = entry.getValue();
+ if (null != listState) {
+ listState = listState.deepCopy();
+ }
+ registeredOperatorStatesDeepCopies.put(entry.getKey(), listState);
+ }
+ }
+
+ if (!registeredBroadcastStates.isEmpty()) {
+ for (Map.Entry<String, BackendWritableBroadcastState<?, ?>> entry : registeredBroadcastStates.entrySet()) {
+ BackendWritableBroadcastState<?, ?> broadcastState = entry.getValue();
+ if (null != broadcastState) {
+ broadcastState = broadcastState.deepCopy();
+ }
+ registeredBroadcastStatesDeepCopies.put(entry.getKey(), broadcastState);
+ }
+ }
+ } finally {
+ Thread.currentThread().setContextClassLoader(snapshotClassLoader);
+ }
+
+ AsyncSnapshotCallable<SnapshotResult<OperatorStateHandle>> snapshotCallable =
+ new AsyncSnapshotCallable<SnapshotResult<OperatorStateHandle>>() {
+
+ @Override
+ protected SnapshotResult<OperatorStateHandle> callInternal() throws Exception {
+
+ CheckpointStreamFactory.CheckpointStateOutputStream localOut =
+ streamFactory.createCheckpointStateOutputStream(CheckpointedStateScope.EXCLUSIVE);
+ snapshotCloseableRegistry.registerCloseable(localOut);
+
+ // get the registered operator state infos ...
+ List<StateMetaInfoSnapshot> operatorMetaInfoSnapshots =
+ new ArrayList<>(registeredOperatorStatesDeepCopies.size());
+
+ for (Map.Entry<String, PartitionableListState<?>> entry :
+ registeredOperatorStatesDeepCopies.entrySet()) {
+ operatorMetaInfoSnapshots.add(entry.getValue().getStateMetaInfo().snapshot());
+ }
+
+ // ... get the registered broadcast operator state infos ...
+ List<StateMetaInfoSnapshot> broadcastMetaInfoSnapshots =
+ new ArrayList<>(registeredBroadcastStatesDeepCopies.size());
+
+ for (Map.Entry<String, BackendWritableBroadcastState<?, ?>> entry :
+ registeredBroadcastStatesDeepCopies.entrySet()) {
+ broadcastMetaInfoSnapshots.add(entry.getValue().getStateMetaInfo().snapshot());
+ }
+
+ // ... write them all in the checkpoint stream ...
+ DataOutputView dov = new DataOutputViewStreamWrapper(localOut);
+
+ OperatorBackendSerializationProxy backendSerializationProxy =
+ new OperatorBackendSerializationProxy(operatorMetaInfoSnapshots, broadcastMetaInfoSnapshots);
+
+ backendSerializationProxy.write(dov);
+
+ // ... and then go for the states ...
+
+ // we put BOTH normal and broadcast state metadata here
+ int initialMapCapacity =
+ registeredOperatorStatesDeepCopies.size() + registeredBroadcastStatesDeepCopies.size();
+ final Map<String, OperatorStateHandle.StateMetaInfo> writtenStatesMetaData =
+ new HashMap<>(initialMapCapacity);
+
+ for (Map.Entry<String, PartitionableListState<?>> entry :
+ registeredOperatorStatesDeepCopies.entrySet()) {
+
+ PartitionableListState<?> value = entry.getValue();
+ long[] partitionOffsets = value.write(localOut);
+ OperatorStateHandle.Mode mode = value.getStateMetaInfo().getAssignmentMode();
+ writtenStatesMetaData.put(
+ entry.getKey(),
+ new OperatorStateHandle.StateMetaInfo(partitionOffsets, mode));
+ }
+
+ // ... and the broadcast states themselves ...
+ for (Map.Entry<String, BackendWritableBroadcastState<?, ?>> entry :
+ registeredBroadcastStatesDeepCopies.entrySet()) {
+
+ BackendWritableBroadcastState<?, ?> value = entry.getValue();
+ long[] partitionOffsets = {value.write(localOut)};
+ OperatorStateHandle.Mode mode = value.getStateMetaInfo().getAssignmentMode();
+ writtenStatesMetaData.put(
+ entry.getKey(),
+ new OperatorStateHandle.StateMetaInfo(partitionOffsets, mode));
+ }
+
+ // ... and, finally, create the state handle.
+ OperatorStateHandle retValue = null;
+
+ if (snapshotCloseableRegistry.unregisterCloseable(localOut)) {
+
+ StreamStateHandle stateHandle = localOut.closeAndGetHandle();
+
+ if (stateHandle != null) {
+ retValue = new OperatorStreamStateHandle(writtenStatesMetaData, stateHandle);
+ }
+
+ return SnapshotResult.of(retValue);
+ } else {
+ throw new IOException("Stream was already unregistered.");
+ }
+ }
+
+ @Override
+ protected void cleanupProvidedResources() {
+ // nothing to do
+ }
+
+ @Override
+ protected void logAsyncSnapshotComplete(long startTime) {
+ if (asynchronousSnapshots) {
+ logAsyncCompleted(streamFactory, startTime);
+ }
+ }
+ };
+
+ final FutureTask<SnapshotResult<OperatorStateHandle>> task =
+ snapshotCallable.toAsyncSnapshotFutureTask(closeStreamOnCancelRegistry);
+
+ if (!asynchronousSnapshots) {
+ task.run();
+ }
+
+ return task;
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/OperatorStateRestoreOperation.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/OperatorStateRestoreOperation.java
new file mode 100644
index 0000000..c34d6da
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/OperatorStateRestoreOperation.java
@@ -0,0 +1,219 @@
+/*
+ * 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.state;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.UnloadableDummyTypeSerializer;
+import org.apache.flink.core.fs.CloseableRegistry;
+import org.apache.flink.core.fs.FSDataInputStream;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataInputViewStreamWrapper;
+import org.apache.flink.runtime.state.metainfo.StateMetaInfoSnapshot;
+import org.apache.flink.util.Preconditions;
+
+import javax.annotation.Nonnull;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Implementation of operator state restore operation.
+ */
+public class OperatorStateRestoreOperation implements RestoreOperation<Void> {
+ private final CloseableRegistry closeStreamOnCancelRegistry;
+ private final ClassLoader userClassloader;
+ private final Map<String, PartitionableListState<?>> registeredOperatorStates;
+ private final Map<String, BackendWritableBroadcastState<?, ?>> registeredBroadcastStates;
+ private final Collection<OperatorStateHandle> stateHandles;
+
+ public OperatorStateRestoreOperation(
+ CloseableRegistry closeStreamOnCancelRegistry,
+ ClassLoader userClassloader,
+ Map<String, PartitionableListState<?>> registeredOperatorStates,
+ Map<String, BackendWritableBroadcastState<?, ?>> registeredBroadcastStates,
+ @Nonnull Collection<OperatorStateHandle> stateHandles) {
+ this.closeStreamOnCancelRegistry = closeStreamOnCancelRegistry;
+ this.userClassloader = userClassloader;
+ this.registeredOperatorStates = registeredOperatorStates;
+ this.registeredBroadcastStates = registeredBroadcastStates;
+ this.stateHandles = stateHandles;
+ }
+
+ @Override
+ public Void restore() throws Exception {
+ if (stateHandles.isEmpty()) {
+ return null;
+ }
+
+ for (OperatorStateHandle stateHandle : stateHandles) {
+
+ if (stateHandle == null) {
+ continue;
+ }
+
+ FSDataInputStream in = stateHandle.openInputStream();
+ closeStreamOnCancelRegistry.registerCloseable(in);
+
+ ClassLoader restoreClassLoader = Thread.currentThread().getContextClassLoader();
+
+ try {
+ Thread.currentThread().setContextClassLoader(userClassloader);
+ OperatorBackendSerializationProxy backendSerializationProxy =
+ new OperatorBackendSerializationProxy(userClassloader);
+
+ backendSerializationProxy.read(new DataInputViewStreamWrapper(in));
+
+ List<StateMetaInfoSnapshot> restoredOperatorMetaInfoSnapshots =
+ backendSerializationProxy.getOperatorStateMetaInfoSnapshots();
+
+ // Recreate all PartitionableListStates from the meta info
+ for (StateMetaInfoSnapshot restoredSnapshot : restoredOperatorMetaInfoSnapshots) {
+
+ final RegisteredOperatorStateBackendMetaInfo<?> restoredMetaInfo =
+ new RegisteredOperatorStateBackendMetaInfo<>(restoredSnapshot);
+
+ if (restoredMetaInfo.getPartitionStateSerializer() instanceof UnloadableDummyTypeSerializer) {
+
+ // must fail now if the previous typeSerializer cannot be restored because there is no typeSerializer
+ // capable of reading previous state
+ // TODO when eager state registration is in place, we can try to get a convert deserializer
+ // TODO from the newly registered typeSerializer instead of simply failing here
+
+ throw new IOException("Unable to restore operator state [" + restoredSnapshot.getName() + "]." +
+ " The previous typeSerializer of the operator state must be present; the typeSerializer could" +
+ " have been removed from the classpath, or its implementation have changed and could" +
+ " not be loaded. This is a temporary restriction that will be fixed in future versions.");
+ }
+
+ PartitionableListState<?> listState = registeredOperatorStates.get(restoredSnapshot.getName());
+
+ if (null == listState) {
+ listState = new PartitionableListState<>(restoredMetaInfo);
+
+ registeredOperatorStates.put(listState.getStateMetaInfo().getName(), listState);
+ } else {
+ // TODO with eager state registration in place, check here for typeSerializer migration strategies
+ }
+ }
+
+ // ... and then get back the broadcast state.
+ List<StateMetaInfoSnapshot> restoredBroadcastMetaInfoSnapshots =
+ backendSerializationProxy.getBroadcastStateMetaInfoSnapshots();
+
+ for (StateMetaInfoSnapshot restoredSnapshot : restoredBroadcastMetaInfoSnapshots) {
+
+ final RegisteredBroadcastStateBackendMetaInfo<?, ?> restoredMetaInfo =
+ new RegisteredBroadcastStateBackendMetaInfo<>(restoredSnapshot);
+
+ if (restoredMetaInfo.getKeySerializer() instanceof UnloadableDummyTypeSerializer ||
+ restoredMetaInfo.getValueSerializer() instanceof UnloadableDummyTypeSerializer) {
+
+ // must fail now if the previous typeSerializer cannot be restored because there is no typeSerializer
+ // capable of reading previous state
+ // TODO when eager state registration is in place, we can try to get a convert deserializer
+ // TODO from the newly registered typeSerializer instead of simply failing here
+
+ throw new IOException("Unable to restore broadcast state [" + restoredSnapshot.getName() + "]." +
+ " The previous key and value serializers of the state must be present; the serializers could" +
+ " have been removed from the classpath, or their implementations have changed and could" +
+ " not be loaded. This is a temporary restriction that will be fixed in future versions.");
+ }
+
+ BackendWritableBroadcastState<?, ?> broadcastState = registeredBroadcastStates.get(restoredSnapshot.getName());
+
+ if (broadcastState == null) {
+ broadcastState = new HeapBroadcastState<>(restoredMetaInfo);
+
+ registeredBroadcastStates.put(broadcastState.getStateMetaInfo().getName(), broadcastState);
+ } else {
+ // TODO with eager state registration in place, check here for typeSerializer migration strategies
+ }
+ }
+
+ // Restore all the states
+ for (Map.Entry<String, OperatorStateHandle.StateMetaInfo> nameToOffsets :
+ stateHandle.getStateNameToPartitionOffsets().entrySet()) {
+
+ final String stateName = nameToOffsets.getKey();
+
+ PartitionableListState<?> listStateForName = registeredOperatorStates.get(stateName);
+ if (listStateForName == null) {
+ BackendWritableBroadcastState<?, ?> broadcastStateForName = registeredBroadcastStates.get(stateName);
+ Preconditions.checkState(broadcastStateForName != null, "Found state without " +
+ "corresponding meta info: " + stateName);
+ deserializeBroadcastStateValues(broadcastStateForName, in, nameToOffsets.getValue());
+ } else {
+ deserializeOperatorStateValues(listStateForName, in, nameToOffsets.getValue());
+ }
+ }
+
+ } finally {
+ Thread.currentThread().setContextClassLoader(restoreClassLoader);
+ if (closeStreamOnCancelRegistry.unregisterCloseable(in)) {
+ IOUtils.closeQuietly(in);
+ }
+ }
+ }
+ return null;
+ }
+
+ private <S> void deserializeOperatorStateValues(
+ PartitionableListState<S> stateListForName,
+ FSDataInputStream in,
+ OperatorStateHandle.StateMetaInfo metaInfo) throws IOException {
+
+ if (null != metaInfo) {
+ long[] offsets = metaInfo.getOffsets();
+ if (null != offsets) {
+ DataInputView div = new DataInputViewStreamWrapper(in);
+ TypeSerializer<S> serializer = stateListForName.getStateMetaInfo().getPartitionStateSerializer();
+ for (long offset : offsets) {
+ in.seek(offset);
+ stateListForName.add(serializer.deserialize(div));
+ }
+ }
+ }
+ }
+
+ private <K, V> void deserializeBroadcastStateValues(
+ final BackendWritableBroadcastState<K, V> broadcastStateForName,
+ final FSDataInputStream in,
+ final OperatorStateHandle.StateMetaInfo metaInfo) throws Exception {
+
+ if (metaInfo != null) {
+ long[] offsets = metaInfo.getOffsets();
+ if (offsets != null) {
+
+ TypeSerializer<K> keySerializer = broadcastStateForName.getStateMetaInfo().getKeySerializer();
+ TypeSerializer<V> valueSerializer = broadcastStateForName.getStateMetaInfo().getValueSerializer();
+
+ in.seek(offsets[0]);
+
+ DataInputView div = new DataInputViewStreamWrapper(in);
+ int size = div.readInt();
+ for (int i = 0; i < size; i++) {
+ broadcastStateForName.put(keySerializer.deserialize(div), valueSerializer.deserialize(div));
+ }
+ }
+ }
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/PartitionableListState.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/PartitionableListState.java
new file mode 100644
index 0000000..6be694c
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/PartitionableListState.java
@@ -0,0 +1,135 @@
+/*
+ * 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.state;
+
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.core.fs.FSDataOutputStream;
+import org.apache.flink.core.memory.DataOutputView;
+import org.apache.flink.core.memory.DataOutputViewStreamWrapper;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Implementation of operator list state.
+ *
+ * @param <S> the type of an operator state partition.
+ */
+public final class PartitionableListState<S> implements ListState<S> {
+
+ /**
+ * Meta information of the state, including state name, assignment mode, and typeSerializer
+ */
+ private RegisteredOperatorStateBackendMetaInfo<S> stateMetaInfo;
+
+ /**
+ * The internal list the holds the elements of the state
+ */
+ private final ArrayList<S> internalList;
+
+ /**
+ * A typeSerializer that allows to perform deep copies of internalList
+ */
+ private final ArrayListSerializer<S> internalListCopySerializer;
+
+ PartitionableListState(RegisteredOperatorStateBackendMetaInfo<S> stateMetaInfo) {
+ this(stateMetaInfo, new ArrayList<S>());
+ }
+
+ private PartitionableListState(
+ RegisteredOperatorStateBackendMetaInfo<S> stateMetaInfo,
+ ArrayList<S> internalList) {
+
+ this.stateMetaInfo = Preconditions.checkNotNull(stateMetaInfo);
+ this.internalList = Preconditions.checkNotNull(internalList);
+ this.internalListCopySerializer = new ArrayListSerializer<>(stateMetaInfo.getPartitionStateSerializer());
+ }
+
+ private PartitionableListState(PartitionableListState<S> toCopy) {
+
+ this(toCopy.stateMetaInfo.deepCopy(), toCopy.internalListCopySerializer.copy(toCopy.internalList));
+ }
+
+ public void setStateMetaInfo(RegisteredOperatorStateBackendMetaInfo<S> stateMetaInfo) {
+ this.stateMetaInfo = stateMetaInfo;
+ }
+
+ public RegisteredOperatorStateBackendMetaInfo<S> getStateMetaInfo() {
+ return stateMetaInfo;
+ }
+
+ public PartitionableListState<S> deepCopy() {
+ return new PartitionableListState<>(this);
+ }
+
+ @Override
+ public void clear() {
+ internalList.clear();
+ }
+
+ @Override
+ public Iterable<S> get() {
+ return internalList;
+ }
+
+ @Override
+ public void add(S value) {
+ Preconditions.checkNotNull(value, "You cannot add null to a ListState.");
+ internalList.add(value);
+ }
+
+ @Override
+ public String toString() {
+ return "PartitionableListState{" +
+ "stateMetaInfo=" + stateMetaInfo +
+ ", internalList=" + internalList +
+ '}';
+ }
+
+ public long[] write(FSDataOutputStream out) throws IOException {
+
+ long[] partitionOffsets = new long[internalList.size()];
+
+ DataOutputView dov = new DataOutputViewStreamWrapper(out);
+
+ for (int i = 0; i < internalList.size(); ++i) {
+ S element = internalList.get(i);
+ partitionOffsets[i] = out.getPos();
+ getStateMetaInfo().getPartitionStateSerializer().serialize(element, dov);
+ }
+
+ return partitionOffsets;
+ }
+
+ @Override
+ public void update(List<S> values) {
+ internalList.clear();
+
+ addAll(values);
+ }
+
+ @Override
+ public void addAll(List<S> values) {
+ if (values != null && !values.isEmpty()) {
+ internalList.addAll(values);
+ }
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/StateBackend.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/StateBackend.java
index 5eac82f..0fccd60 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/StateBackend.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/StateBackend.java
@@ -259,10 +259,16 @@ public interface StateBackend extends java.io.Serializable {
*
* @param env The runtime environment of the executing task.
* @param operatorIdentifier The identifier of the operator whose state should be stored.
+ * @param stateHandles The state handles for restore.
+ * @param cancelStreamRegistry The registry to register streams to close if task canceled.
*
* @return The OperatorStateBackend for operator identified by the job and operator identifier.
*
* @throws Exception This method may forward all exceptions that occur while instantiating the backend.
*/
- OperatorStateBackend createOperatorStateBackend(Environment env, String operatorIdentifier) throws Exception;
+ OperatorStateBackend createOperatorStateBackend(
+ Environment env,
+ String operatorIdentifier,
+ @Nonnull Collection<OperatorStateHandle> stateHandles,
+ CloseableRegistry cancelStreamRegistry) throws Exception;
}
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 0511dd3..f516fc2 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
@@ -34,11 +34,12 @@ import org.apache.flink.runtime.state.AbstractStateBackend;
import org.apache.flink.runtime.state.BackendBuildingException;
import org.apache.flink.runtime.state.CheckpointStorage;
import org.apache.flink.runtime.state.ConfigurableStateBackend;
-import org.apache.flink.runtime.state.DefaultOperatorStateBackend;
+import org.apache.flink.runtime.state.DefaultOperatorStateBackendBuilder;
import org.apache.flink.runtime.state.KeyGroupRange;
import org.apache.flink.runtime.state.KeyedStateHandle;
import org.apache.flink.runtime.state.LocalRecoveryConfig;
import org.apache.flink.runtime.state.OperatorStateBackend;
+import org.apache.flink.runtime.state.OperatorStateHandle;
import org.apache.flink.runtime.state.TaskStateManager;
import org.apache.flink.runtime.state.heap.HeapKeyedStateBackendBuilder;
import org.apache.flink.runtime.state.heap.HeapPriorityQueueSetFactory;
@@ -491,12 +492,16 @@ public class FsStateBackend extends AbstractFileStateBackend implements Configur
@Override
public OperatorStateBackend createOperatorStateBackend(
Environment env,
- String operatorIdentifier) {
+ String operatorIdentifier,
+ @Nonnull Collection<OperatorStateHandle> stateHandles,
+ CloseableRegistry cancelStreamRegistry) throws BackendBuildingException {
- return new DefaultOperatorStateBackend(
+ return new DefaultOperatorStateBackendBuilder(
env.getUserClassLoader(),
env.getExecutionConfig(),
- isUsingAsynchronousSnapshots());
+ isUsingAsynchronousSnapshots(),
+ stateHandles,
+ cancelStreamRegistry).build();
}
// ------------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapKeyedStateBackendBuilder.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapKeyedStateBackendBuilder.java
index 3047fd0..c1bd0bb 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapKeyedStateBackendBuilder.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapKeyedStateBackendBuilder.java
@@ -126,6 +126,7 @@ public class HeapKeyedStateBackendBuilder<K> extends AbstractKeyedStateBackendBu
try {
restoreOperation.restore();
} catch (Exception e) {
+ backend.dispose();
throw new BackendBuildingException("Failed when trying to restore heap backend", e);
}
return backend;
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 03e11ae..b2ff0e8 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
@@ -33,10 +33,11 @@ import org.apache.flink.runtime.state.AbstractStateBackend;
import org.apache.flink.runtime.state.BackendBuildingException;
import org.apache.flink.runtime.state.CheckpointStorage;
import org.apache.flink.runtime.state.ConfigurableStateBackend;
-import org.apache.flink.runtime.state.DefaultOperatorStateBackend;
+import org.apache.flink.runtime.state.DefaultOperatorStateBackendBuilder;
import org.apache.flink.runtime.state.KeyGroupRange;
import org.apache.flink.runtime.state.KeyedStateHandle;
import org.apache.flink.runtime.state.OperatorStateBackend;
+import org.apache.flink.runtime.state.OperatorStateHandle;
import org.apache.flink.runtime.state.TaskStateManager;
import org.apache.flink.runtime.state.filesystem.AbstractFileStateBackend;
import org.apache.flink.runtime.state.heap.HeapKeyedStateBackendBuilder;
@@ -300,13 +301,17 @@ public class MemoryStateBackend extends AbstractFileStateBackend implements Conf
@Override
public OperatorStateBackend createOperatorStateBackend(
- Environment env,
- String operatorIdentifier) throws Exception {
+ Environment env,
+ String operatorIdentifier,
+ @Nonnull Collection<OperatorStateHandle> stateHandles,
+ CloseableRegistry cancelStreamRegistry) throws Exception {
- return new DefaultOperatorStateBackend(
- env.getUserClassLoader(),
- env.getExecutionConfig(),
- isUsingAsynchronousSnapshots());
+ return new DefaultOperatorStateBackendBuilder(
+ env.getUserClassLoader(),
+ env.getExecutionConfig(),
+ isUsingAsynchronousSnapshots(),
+ stateHandles,
+ cancelStreamRegistry).build();
}
@Override
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointSettingsSerializableTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointSettingsSerializableTest.java
index cc4436d..62259e3 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointSettingsSerializableTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointSettingsSerializableTest.java
@@ -43,6 +43,7 @@ import org.apache.flink.runtime.state.CompletedCheckpointStorageLocation;
import org.apache.flink.runtime.state.KeyedStateHandle;
import org.apache.flink.runtime.state.KeyGroupRange;
import org.apache.flink.runtime.state.OperatorStateBackend;
+import org.apache.flink.runtime.state.OperatorStateHandle;
import org.apache.flink.runtime.state.StateBackend;
import org.apache.flink.runtime.state.ttl.TtlTimeProvider;
import org.apache.flink.runtime.testingUtils.TestingUtils;
@@ -184,7 +185,10 @@ public class CheckpointSettingsSerializableTest extends TestLogger {
@Override
public OperatorStateBackend createOperatorStateBackend(
- Environment env, String operatorIdentifier) throws Exception {
+ Environment env,
+ String operatorIdentifier,
+ @Nonnull Collection<OperatorStateHandle> stateHandles,
+ CloseableRegistry cancelStreamRegistry) throws Exception {
throw new UnsupportedOperationException();
}
}
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 f842fb2..9ea2f98 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
@@ -28,6 +28,7 @@ import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
import org.apache.flink.api.common.typeutils.base.IntSerializer;
import org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer;
+import org.apache.flink.core.fs.CloseableRegistry;
import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.core.memory.DataOutputView;
import org.apache.flink.core.testutils.OneShotLatch;
@@ -35,19 +36,19 @@ import org.apache.flink.runtime.checkpoint.CheckpointOptions;
import org.apache.flink.runtime.checkpoint.StateObjectCollection;
import org.apache.flink.runtime.concurrent.FutureUtils;
import org.apache.flink.runtime.execution.Environment;
-import org.apache.flink.runtime.state.DefaultOperatorStateBackend.PartitionableListState;
import org.apache.flink.runtime.state.memory.MemCheckpointStreamFactory;
import org.apache.flink.runtime.state.memory.MemoryStateBackend;
import org.apache.flink.runtime.util.BlockerCheckpointStreamFactory;
import org.apache.flink.runtime.util.BlockingCheckpointOutputStream;
import org.apache.flink.util.Preconditions;
-
import org.junit.Assert;
import org.junit.Test;
import java.io.File;
import java.io.IOException;
import java.io.Serializable;
+import java.util.Collection;
+import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
@@ -71,13 +72,15 @@ import static org.mockito.Mockito.when;
public class OperatorStateBackendTest {
private final ClassLoader classLoader = getClass().getClassLoader();
+ private final Collection<OperatorStateHandle> emptyStateHandles = Collections.emptyList();
@Test
public void testCreateOnAbstractStateBackend() throws Exception {
// we use the memory state backend as a subclass of the AbstractStateBackend
final AbstractStateBackend abstractStateBackend = new MemoryStateBackend();
+ CloseableRegistry cancelStreamRegistry = new CloseableRegistry();
final OperatorStateBackend operatorStateBackend = abstractStateBackend.createOperatorStateBackend(
- createMockEnvironment(), "test-operator");
+ createMockEnvironment(), "test-operator", emptyStateHandles, cancelStreamRegistry);
assertNotNull(operatorStateBackend);
assertTrue(operatorStateBackend.getRegisteredStateNames().isEmpty());
@@ -97,7 +100,14 @@ public class OperatorStateBackendTest {
final ExecutionConfig cfg = new ExecutionConfig();
cfg.registerTypeWithKryoSerializer(registeredType, com.esotericsoftware.kryo.serializers.JavaSerializer.class);
- final OperatorStateBackend operatorStateBackend = new DefaultOperatorStateBackend(classLoader, cfg, false);
+ final OperatorStateBackend operatorStateBackend =
+ new DefaultOperatorStateBackendBuilder(
+ classLoader,
+ cfg,
+ false,
+ emptyStateHandles,
+ new CloseableRegistry())
+ .build();
ListStateDescriptor<File> stateDescriptor = new ListStateDescriptor<>("test", File.class);
ListStateDescriptor<String> stateDescriptor2 = new ListStateDescriptor<>("test2", String.class);
@@ -130,7 +140,12 @@ public class OperatorStateBackendTest {
@Test
public void testRegisterStates() throws Exception {
final OperatorStateBackend operatorStateBackend =
- new DefaultOperatorStateBackend(classLoader, new ExecutionConfig(), false);
+ new DefaultOperatorStateBackendBuilder(
+ classLoader,
+ new ExecutionConfig(),
+ false,
+ emptyStateHandles,
+ new CloseableRegistry()).build();
ListStateDescriptor<Serializable> stateDescriptor1 = new ListStateDescriptor<>("test1", new JavaSerializer<>());
ListStateDescriptor<Serializable> stateDescriptor2 = new ListStateDescriptor<>("test2", new JavaSerializer<>());
@@ -219,7 +234,8 @@ public class OperatorStateBackendTest {
AbstractStateBackend abstractStateBackend = new MemoryStateBackend(4096);
final Environment env = createMockEnvironment();
- OperatorStateBackend operatorStateBackend = abstractStateBackend.createOperatorStateBackend(env, "test-op-name");
+ CloseableRegistry cancelStreamRegistry = new CloseableRegistry();
+ OperatorStateBackend operatorStateBackend = abstractStateBackend.createOperatorStateBackend(env, "test-op-name", emptyStateHandles, cancelStreamRegistry);
AtomicInteger copyCounter = new AtomicInteger(0);
TypeSerializer<Integer> serializer = new VerifyingIntSerializer(env.getUserClassLoader(), copyCounter);
@@ -353,9 +369,10 @@ public class OperatorStateBackendTest {
@Test
public void testSnapshotEmpty() throws Exception {
final AbstractStateBackend abstractStateBackend = new MemoryStateBackend(4096);
+ CloseableRegistry cancelStreamRegistry = new CloseableRegistry();
final OperatorStateBackend operatorStateBackend =
- abstractStateBackend.createOperatorStateBackend(createMockEnvironment(), "testOperator");
+ abstractStateBackend.createOperatorStateBackend(createMockEnvironment(), "testOperator", emptyStateHandles, cancelStreamRegistry);
CheckpointStreamFactory streamFactory = new MemCheckpointStreamFactory(4096);
@@ -371,8 +388,12 @@ public class OperatorStateBackendTest {
public void testSnapshotBroadcastStateWithEmptyOperatorState() throws Exception {
final AbstractStateBackend abstractStateBackend = new MemoryStateBackend(4096);
- final OperatorStateBackend operatorStateBackend =
- abstractStateBackend.createOperatorStateBackend(createMockEnvironment(), "testOperator");
+ OperatorStateBackend operatorStateBackend =
+ abstractStateBackend.createOperatorStateBackend(
+ createMockEnvironment(),
+ "testOperator",
+ emptyStateHandles,
+ new CloseableRegistry());
final MapStateDescriptor<Integer, Integer> broadcastStateDesc = new MapStateDescriptor<>(
"test-broadcast", BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO);
@@ -398,7 +419,10 @@ public class OperatorStateBackendTest {
final Map<Integer, Integer> retrieved = new HashMap<>();
- operatorStateBackend.restore(StateObjectCollection.singleton(stateHandle));
+ operatorStateBackend = recreateOperatorStateBackend(
+ operatorStateBackend,
+ abstractStateBackend,
+ StateObjectCollection.singleton(stateHandle));
BroadcastState<Integer, Integer> retrievedState = operatorStateBackend.getBroadcastState(broadcastStateDesc);
for (Map.Entry<Integer, Integer> e: retrievedState.entries()) {
retrieved.put(e.getKey(), e.getValue());
@@ -406,7 +430,7 @@ public class OperatorStateBackendTest {
assertEquals(expected, retrieved);
// remove an element from both expected and stored state.
- broadcastState.remove(1);
+ retrievedState.remove(1);
expected.remove(1);
snapshot = operatorStateBackend.snapshot(1L, 1L, streamFactory, CheckpointOptions.forCheckpointWithDefaultLocation());
@@ -416,7 +440,10 @@ public class OperatorStateBackendTest {
stateHandle = snapshotResult.getJobManagerOwnedSnapshot();
retrieved.clear();
- operatorStateBackend.restore(StateObjectCollection.singleton(stateHandle));
+ operatorStateBackend = recreateOperatorStateBackend(
+ operatorStateBackend,
+ abstractStateBackend,
+ StateObjectCollection.singleton(stateHandle));
retrievedState = operatorStateBackend.getBroadcastState(broadcastStateDesc);
for (Map.Entry<Integer, Integer> e: retrievedState.immutableEntries()) {
retrieved.put(e.getKey(), e.getValue());
@@ -424,7 +451,7 @@ public class OperatorStateBackendTest {
assertEquals(expected, retrieved);
// remove all elements from both expected and stored state.
- broadcastState.clear();
+ retrievedState.clear();
expected.clear();
snapshot = operatorStateBackend.snapshot(2L, 2L, streamFactory, CheckpointOptions.forCheckpointWithDefaultLocation());
@@ -435,7 +462,10 @@ public class OperatorStateBackendTest {
stateHandle = snapshotResult.getJobManagerOwnedSnapshot();
retrieved.clear();
- operatorStateBackend.restore(StateObjectCollection.singleton(stateHandle));
+ operatorStateBackend = recreateOperatorStateBackend(
+ operatorStateBackend,
+ abstractStateBackend,
+ StateObjectCollection.singleton(stateHandle));
retrievedState = operatorStateBackend.getBroadcastState(broadcastStateDesc);
for (Map.Entry<Integer, Integer> e: retrievedState.immutableEntries()) {
retrieved.put(e.getKey(), e.getValue());
@@ -459,7 +489,11 @@ public class OperatorStateBackendTest {
public void testSnapshotRestoreSync() throws Exception {
AbstractStateBackend abstractStateBackend = new MemoryStateBackend(2 * 4096);
- OperatorStateBackend operatorStateBackend = abstractStateBackend.createOperatorStateBackend(createMockEnvironment(), "test-op-name");
+ OperatorStateBackend operatorStateBackend = abstractStateBackend.createOperatorStateBackend(
+ createMockEnvironment(),
+ "test-op-name",
+ emptyStateHandles,
+ new CloseableRegistry());
ListStateDescriptor<Serializable> stateDescriptor1 = new ListStateDescriptor<>("test1", new JavaSerializer<>());
ListStateDescriptor<Serializable> stateDescriptor2 = new ListStateDescriptor<>("test2", new JavaSerializer<>());
ListStateDescriptor<Serializable> stateDescriptor3 = new ListStateDescriptor<>("test3", new JavaSerializer<>());
@@ -507,9 +541,9 @@ public class OperatorStateBackendTest {
operatorStateBackend = abstractStateBackend.createOperatorStateBackend(
createMockEnvironment(),
- "testOperator");
-
- operatorStateBackend.restore(StateObjectCollection.singleton(stateHandle));
+ "testOperator",
+ StateObjectCollection.singleton(stateHandle),
+ new CloseableRegistry());
assertEquals(3, operatorStateBackend.getRegisteredStateNames().size());
assertEquals(3, operatorStateBackend.getRegisteredBroadcastStateNames().size());
@@ -574,7 +608,12 @@ public class OperatorStateBackendTest {
@Test
public void testSnapshotRestoreAsync() throws Exception {
OperatorStateBackend operatorStateBackend =
- new DefaultOperatorStateBackend(OperatorStateBackendTest.class.getClassLoader(), new ExecutionConfig(), true);
+ new DefaultOperatorStateBackendBuilder(
+ OperatorStateBackendTest.class.getClassLoader(),
+ new ExecutionConfig(),
+ true,
+ emptyStateHandles,
+ new CloseableRegistry()).build();
ListStateDescriptor<MutableType> stateDescriptor1 =
new ListStateDescriptor<>("test1", new JavaSerializer<MutableType>());
@@ -665,12 +704,13 @@ public class OperatorStateBackendTest {
operatorStateBackend.dispose();
AbstractStateBackend abstractStateBackend = new MemoryStateBackend(4096);
+ CloseableRegistry cancelStreamRegistry = new CloseableRegistry();
operatorStateBackend = abstractStateBackend.createOperatorStateBackend(
- createMockEnvironment(),
- "testOperator");
-
- operatorStateBackend.restore(StateObjectCollection.singleton(stateHandle));
+ createMockEnvironment(),
+ "testOperator",
+ StateObjectCollection.singleton(stateHandle),
+ cancelStreamRegistry);
assertEquals(3, operatorStateBackend.getRegisteredStateNames().size());
assertEquals(3, operatorStateBackend.getRegisteredBroadcastStateNames().size());
@@ -737,7 +777,12 @@ public class OperatorStateBackendTest {
@Test
public void testSnapshotAsyncClose() throws Exception {
DefaultOperatorStateBackend operatorStateBackend =
- new DefaultOperatorStateBackend(OperatorStateBackendTest.class.getClassLoader(), new ExecutionConfig(), true);
+ new DefaultOperatorStateBackendBuilder(
+ OperatorStateBackendTest.class.getClassLoader(),
+ new ExecutionConfig(),
+ true,
+ emptyStateHandles,
+ new CloseableRegistry()).build();
ListStateDescriptor<MutableType> stateDescriptor1 =
new ListStateDescriptor<>("test1", new JavaSerializer<MutableType>());
@@ -786,7 +831,12 @@ public class OperatorStateBackendTest {
@Test
public void testSnapshotAsyncCancel() throws Exception {
DefaultOperatorStateBackend operatorStateBackend =
- new DefaultOperatorStateBackend(OperatorStateBackendTest.class.getClassLoader(), new ExecutionConfig(), true);
+ new DefaultOperatorStateBackendBuilder(
+ OperatorStateBackendTest.class.getClassLoader(),
+ new ExecutionConfig(),
+ true,
+ emptyStateHandles,
+ new CloseableRegistry()).build();
ListStateDescriptor<MutableType> stateDescriptor1 =
new ListStateDescriptor<>("test1", new JavaSerializer<MutableType>());
@@ -889,4 +939,18 @@ public class OperatorStateBackendTest {
when(env.getUserClassLoader()).thenReturn(OperatorStateBackendTest.class.getClassLoader());
return env;
}
+
+ private static OperatorStateBackend recreateOperatorStateBackend(
+ OperatorStateBackend oldOperatorStateBackend,
+ AbstractStateBackend abstractStateBackend,
+ Collection<OperatorStateHandle> toRestore
+ ) throws Exception {
+ oldOperatorStateBackend.close();
+ oldOperatorStateBackend.dispose();
+ return abstractStateBackend.createOperatorStateBackend(
+ createMockEnvironment(),
+ "testOperator",
+ toRestore,
+ new CloseableRegistry());
+ }
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateBackendMigrationTestBase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateBackendMigrationTestBase.java
index 6a01583..90f9db1 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateBackendMigrationTestBase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateBackendMigrationTestBase.java
@@ -28,6 +28,7 @@ import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.common.typeutils.TypeSerializerSchemaCompatibility;
import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
import org.apache.flink.api.common.typeutils.base.IntSerializer;
+import org.apache.flink.core.fs.CloseableRegistry;
import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.core.memory.DataOutputView;
import org.apache.flink.runtime.checkpoint.CheckpointOptions;
@@ -46,6 +47,7 @@ import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import java.io.IOException;
+import java.util.Collection;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
@@ -1059,12 +1061,17 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
// -------------------------------------------------------------------------------
private OperatorStateBackend createOperatorStateBackend() throws Exception {
- return getStateBackend().createOperatorStateBackend(new DummyEnvironment(), "test_op");
+ return getStateBackend().createOperatorStateBackend(
+ new DummyEnvironment(), "test_op", Collections.emptyList(), new CloseableRegistry());
+ }
+
+ private OperatorStateBackend createOperatorStateBackend(Collection<OperatorStateHandle> state) throws Exception {
+ return getStateBackend().createOperatorStateBackend(
+ new DummyEnvironment(), "test_op", state, new CloseableRegistry());
}
private OperatorStateBackend restoreOperatorStateBackend(OperatorStateHandle state) throws Exception {
- OperatorStateBackend operatorStateBackend = createOperatorStateBackend();
- operatorStateBackend.restore(StateObjectCollection.singleton(state));
+ OperatorStateBackend operatorStateBackend = createOperatorStateBackend(StateObjectCollection.singleton(state));
return operatorStateBackend;
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/mock/MockStateBackend.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/mock/MockStateBackend.java
index 117db95..bdf07bf 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/mock/MockStateBackend.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/mock/MockStateBackend.java
@@ -38,6 +38,7 @@ import org.apache.flink.runtime.state.CompletedCheckpointStorageLocation;
import org.apache.flink.runtime.state.KeyedStateHandle;
import org.apache.flink.runtime.state.KeyGroupRange;
import org.apache.flink.runtime.state.OperatorStateBackend;
+import org.apache.flink.runtime.state.OperatorStateHandle;
import org.apache.flink.runtime.state.ttl.TtlTimeProvider;
import javax.annotation.Nonnull;
@@ -142,7 +143,11 @@ public class MockStateBackend extends AbstractStateBackend {
}
@Override
- public OperatorStateBackend createOperatorStateBackend(Environment env, String operatorIdentifier) {
+ public OperatorStateBackend createOperatorStateBackend(
+ Environment env,
+ String operatorIdentifier,
+ @Nonnull Collection<OperatorStateHandle> stateHandles,
+ CloseableRegistry cancelStreamRegistry) {
throw new UnsupportedOperationException();
}
}
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackend.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackend.java
index 62e30ac..c5604e5 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackend.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackend.java
@@ -34,11 +34,12 @@ import org.apache.flink.runtime.state.AbstractStateBackend;
import org.apache.flink.runtime.state.CheckpointStorage;
import org.apache.flink.runtime.state.CompletedCheckpointStorageLocation;
import org.apache.flink.runtime.state.ConfigurableStateBackend;
-import org.apache.flink.runtime.state.DefaultOperatorStateBackend;
+import org.apache.flink.runtime.state.DefaultOperatorStateBackendBuilder;
import org.apache.flink.runtime.state.KeyGroupRange;
import org.apache.flink.runtime.state.KeyedStateHandle;
import org.apache.flink.runtime.state.LocalRecoveryConfig;
import org.apache.flink.runtime.state.OperatorStateBackend;
+import org.apache.flink.runtime.state.OperatorStateHandle;
import org.apache.flink.runtime.state.StateBackend;
import org.apache.flink.runtime.state.StreamCompressionDecorator;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
@@ -521,15 +522,19 @@ public class RocksDBStateBackend extends AbstractStateBackend implements Configu
@Override
public OperatorStateBackend createOperatorStateBackend(
- Environment env,
- String operatorIdentifier) throws Exception {
+ Environment env,
+ String operatorIdentifier,
+ @Nonnull Collection<OperatorStateHandle> stateHandles,
+ CloseableRegistry cancelStreamRegistry) 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);
+ return new DefaultOperatorStateBackendBuilder(
+ env.getUserClassLoader(),
+ env.getExecutionConfig(),
+ asyncSnapshots,
+ stateHandles,
+ cancelStreamRegistry).build();
}
private OptionsFactory configureOptionsFactory(
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamTaskStateInitializerImpl.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamTaskStateInitializerImpl.java
index 768a4f5..5264e50 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamTaskStateInitializerImpl.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamTaskStateInitializerImpl.java
@@ -236,14 +236,29 @@ public class StreamTaskStateInitializerImpl implements StreamTaskStateInitialize
String logDescription = "operator state backend for " + operatorIdentifierText;
+ // Now restore processing is included in backend building/constructing process, so we need to make sure
+ // each stream constructed in restore could also be closed in case of task cancel, for example the data
+ // input stream opened for serDe during restore.
+ CloseableRegistry cancelStreamRegistryForRestore = new CloseableRegistry();
+ backendCloseableRegistry.registerCloseable(cancelStreamRegistryForRestore);
BackendRestorerProcedure<OperatorStateBackend, OperatorStateHandle> backendRestorer =
new BackendRestorerProcedure<>(
- (stateHandles) -> stateBackend.createOperatorStateBackend(environment, operatorIdentifierText),
+ (stateHandles) -> stateBackend.createOperatorStateBackend(
+ environment,
+ operatorIdentifierText,
+ stateHandles,
+ cancelStreamRegistryForRestore),
backendCloseableRegistry,
logDescription);
- return backendRestorer.createAndRestore(
- prioritizedOperatorSubtaskStates.getPrioritizedManagedOperatorState());
+ try {
+ return backendRestorer.createAndRestore(
+ prioritizedOperatorSubtaskStates.getPrioritizedManagedOperatorState());
+ } finally {
+ if (backendCloseableRegistry.unregisterCloseable(cancelStreamRegistryForRestore)) {
+ IOUtils.closeQuietly(cancelStreamRegistryForRestore);
+ }
+ }
}
protected <K> AbstractKeyedStateBackend<K> keyedStatedBackend(
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/BackendRestorerProcedureTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/BackendRestorerProcedureTest.java
index 876d35e..6203481 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/BackendRestorerProcedureTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/BackendRestorerProcedureTest.java
@@ -26,7 +26,7 @@ import org.apache.flink.core.testutils.OneShotLatch;
import org.apache.flink.runtime.checkpoint.CheckpointOptions;
import org.apache.flink.runtime.checkpoint.StateObjectCollection;
import org.apache.flink.runtime.state.CheckpointStreamFactory;
-import org.apache.flink.runtime.state.DefaultOperatorStateBackend;
+import org.apache.flink.runtime.state.DefaultOperatorStateBackendBuilder;
import org.apache.flink.runtime.state.OperatorStateBackend;
import org.apache.flink.runtime.state.OperatorStateHandle;
import org.apache.flink.runtime.state.SnapshotResult;
@@ -59,10 +59,12 @@ import static org.powermock.api.mockito.PowerMockito.when;
public class BackendRestorerProcedureTest extends TestLogger {
private final FunctionWithException<Collection<OperatorStateHandle>, OperatorStateBackend, Exception> backendSupplier =
- (stateHandles) -> new DefaultOperatorStateBackend(
+ (stateHandles) -> new DefaultOperatorStateBackendBuilder(
getClass().getClassLoader(),
new ExecutionConfig(),
- true);
+ true,
+ stateHandles,
+ new CloseableRegistry()).build();
/**
* Tests that the restore procedure follows the order of the iterator and will retries failed attempts if there are
@@ -75,7 +77,7 @@ public class BackendRestorerProcedureTest extends TestLogger {
CheckpointStreamFactory checkpointStreamFactory = new MemCheckpointStreamFactory(1024);
ListStateDescriptor<Integer> stateDescriptor = new ListStateDescriptor<>("test-state", Integer.class);
- OperatorStateBackend originalBackend = backendSupplier.apply(null);
+ OperatorStateBackend originalBackend = backendSupplier.apply(Collections.emptyList());
SnapshotResult<OperatorStateHandle> snapshotResult;
try {
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/StreamTaskStateInitializerImplTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/StreamTaskStateInitializerImplTest.java
index b66a7b4..22f6441 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/StreamTaskStateInitializerImplTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/StreamTaskStateInitializerImplTest.java
@@ -156,7 +156,10 @@ public class StreamTaskStateInitializerImplTest {
@Override
public OperatorStateBackend createOperatorStateBackend(
- Environment env, String operatorIdentifier) throws Exception {
+ Environment env,
+ String operatorIdentifier,
+ @Nonnull Collection<OperatorStateHandle> stateHandles,
+ CloseableRegistry cancelStreamRegistry) throws Exception {
return mock(OperatorStateBackend.class);
}
});
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTerminationTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTerminationTest.java
index e448bad..54f4855 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTerminationTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTerminationTest.java
@@ -279,7 +279,11 @@ public class StreamTaskTerminationTest extends TestLogger {
}
@Override
- public OperatorStateBackend createOperatorStateBackend(Environment env, String operatorIdentifier) throws Exception {
+ public OperatorStateBackend createOperatorStateBackend(
+ Environment env,
+ String operatorIdentifier,
+ @Nonnull Collection<OperatorStateHandle> stateHandles,
+ CloseableRegistry cancelStreamRegistry) throws Exception {
OperatorStateBackend operatorStateBackend = mock(OperatorStateBackend.class);
when(operatorStateBackend.snapshot(anyLong(), anyLong(), any(CheckpointStreamFactory.class), any(CheckpointOptions.class)))
.thenReturn(new FutureTask<>(new BlockingCallable()));
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/TaskCheckpointingBehaviourTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/TaskCheckpointingBehaviourTest.java
index 96def7a..5c43564 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/TaskCheckpointingBehaviourTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/TaskCheckpointingBehaviourTest.java
@@ -22,6 +22,7 @@ import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.CloseableRegistry;
import org.apache.flink.core.testutils.OneShotLatch;
import org.apache.flink.runtime.blob.BlobCacheService;
import org.apache.flink.runtime.blob.PermanentBlobCache;
@@ -54,10 +55,12 @@ import org.apache.flink.runtime.jobgraph.tasks.InputSplitProvider;
import org.apache.flink.runtime.memory.MemoryManager;
import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
import org.apache.flink.runtime.query.TaskKvStateRegistry;
+import org.apache.flink.runtime.state.AbstractSnapshotStrategy;
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.DefaultOperatorStateBackendBuilder;
import org.apache.flink.runtime.state.OperatorStateBackend;
import org.apache.flink.runtime.state.OperatorStateCheckpointOutputStream;
import org.apache.flink.runtime.state.OperatorStateHandle;
@@ -87,7 +90,9 @@ import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import java.io.IOException;
+import java.util.Collection;
import java.util.Collections;
+import java.util.HashMap;
import java.util.concurrent.FutureTask;
import java.util.concurrent.RunnableFuture;
@@ -303,22 +308,42 @@ public class TaskCheckpointingBehaviourTest extends TestLogger {
private static final long serialVersionUID = -1915780414440060539L;
@Override
- public OperatorStateBackend createOperatorStateBackend(Environment env, String operatorIdentifier) throws Exception {
- return new DefaultOperatorStateBackend(
+ public OperatorStateBackend createOperatorStateBackend(
+ Environment env,
+ String operatorIdentifier,
+ @Nonnull Collection<OperatorStateHandle> stateHandles,
+ CloseableRegistry cancelStreamRegistry) throws Exception {
+ return new DefaultOperatorStateBackendBuilder(
env.getUserClassLoader(),
env.getExecutionConfig(),
- true) {
- @Nonnull
+ true,
+ stateHandles,
+ cancelStreamRegistry) {
@Override
- public RunnableFuture<SnapshotResult<OperatorStateHandle>> snapshot(
- long checkpointId,
- long timestamp,
- @Nonnull CheckpointStreamFactory streamFactory,
- @Nonnull CheckpointOptions checkpointOptions) throws Exception {
-
- throw new Exception("Sync part snapshot exception.");
+ @SuppressWarnings("unchecked")
+ public DefaultOperatorStateBackend build() {
+ return new DefaultOperatorStateBackend(
+ executionConfig,
+ cancelStreamRegistry,
+ new HashMap<>(),
+ new HashMap<>(),
+ new HashMap<>(),
+ new HashMap<>(),
+ mock(AbstractSnapshotStrategy.class)
+ ) {
+ @Nonnull
+ @Override
+ public RunnableFuture<SnapshotResult<OperatorStateHandle>> snapshot(
+ long checkpointId,
+ long timestamp,
+ @Nonnull CheckpointStreamFactory streamFactory,
+ @Nonnull CheckpointOptions checkpointOptions) throws Exception {
+
+ throw new Exception("Sync part snapshot exception.");
+ }
+ };
}
- };
+ }.build();
}
@Override
@@ -333,24 +358,44 @@ public class TaskCheckpointingBehaviourTest extends TestLogger {
private static final long serialVersionUID = -7613628662587098470L;
@Override
- public OperatorStateBackend createOperatorStateBackend(Environment env, String operatorIdentifier) throws Exception {
- return new DefaultOperatorStateBackend(
+ public OperatorStateBackend createOperatorStateBackend(
+ Environment env,
+ String operatorIdentifier,
+ @Nonnull Collection<OperatorStateHandle> stateHandles,
+ CloseableRegistry cancelStreamRegistry) throws Exception {
+ return new DefaultOperatorStateBackendBuilder(
env.getUserClassLoader(),
env.getExecutionConfig(),
- true) {
- @Nonnull
+ true,
+ stateHandles,
+ cancelStreamRegistry) {
@Override
- public RunnableFuture<SnapshotResult<OperatorStateHandle>> snapshot(
- long checkpointId,
- long timestamp,
- @Nonnull CheckpointStreamFactory streamFactory,
- @Nonnull CheckpointOptions checkpointOptions) throws Exception {
-
- return new FutureTask<>(() -> {
- throw new Exception("Async part snapshot exception.");
- });
+ @SuppressWarnings("unchecked")
+ public DefaultOperatorStateBackend build() {
+ return new DefaultOperatorStateBackend(
+ executionConfig,
+ cancelStreamRegistry,
+ new HashMap<>(),
+ new HashMap<>(),
+ new HashMap<>(),
+ new HashMap<>(),
+ mock(AbstractSnapshotStrategy.class)
+ ) {
+ @Nonnull
+ @Override
+ public RunnableFuture<SnapshotResult<OperatorStateHandle>> snapshot(
+ long checkpointId,
+ long timestamp,
+ @Nonnull CheckpointStreamFactory streamFactory,
+ @Nonnull CheckpointOptions checkpointOptions) throws Exception {
+
+ return new FutureTask<>(() -> {
+ throw new Exception("Async part snapshot exception.");
+ });
+ }
+ };
}
- };
+ }.build();
}
@Override
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/TestSpyWrapperStateBackend.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/TestSpyWrapperStateBackend.java
index 4160260..3fd42d7 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/TestSpyWrapperStateBackend.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/TestSpyWrapperStateBackend.java
@@ -31,6 +31,7 @@ import org.apache.flink.runtime.state.CompletedCheckpointStorageLocation;
import org.apache.flink.runtime.state.KeyGroupRange;
import org.apache.flink.runtime.state.KeyedStateHandle;
import org.apache.flink.runtime.state.OperatorStateBackend;
+import org.apache.flink.runtime.state.OperatorStateHandle;
import org.apache.flink.runtime.state.ttl.TtlTimeProvider;
import org.apache.flink.util.Preconditions;
@@ -81,8 +82,11 @@ public class TestSpyWrapperStateBackend extends AbstractStateBackend {
@Override
public OperatorStateBackend createOperatorStateBackend(
- Environment env, String operatorIdentifier) throws Exception {
- return spy(delegate.createOperatorStateBackend(env, operatorIdentifier));
+ Environment env,
+ String operatorIdentifier,
+ @Nonnull Collection<OperatorStateHandle> stateHandles,
+ CloseableRegistry cancelStreamRegistry) throws Exception {
+ return spy(delegate.createOperatorStateBackend(env, operatorIdentifier, stateHandles, cancelStreamRegistry));
}
@Override
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 4c71e14..224c237 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
@@ -36,6 +36,7 @@ import org.apache.flink.runtime.state.CompletedCheckpointStorageLocation;
import org.apache.flink.runtime.state.KeyGroupRange;
import org.apache.flink.runtime.state.KeyedStateHandle;
import org.apache.flink.runtime.state.OperatorStateBackend;
+import org.apache.flink.runtime.state.OperatorStateHandle;
import org.apache.flink.runtime.state.StateBackend;
import org.apache.flink.runtime.state.memory.MemoryBackendCheckpointStorage;
import org.apache.flink.runtime.state.ttl.TtlTimeProvider;
@@ -129,7 +130,9 @@ public class StateBackendITCase extends AbstractTestBase {
@Override
public OperatorStateBackend createOperatorStateBackend(
Environment env,
- String operatorIdentifier) throws Exception {
+ String operatorIdentifier,
+ @Nonnull Collection<OperatorStateHandle> stateHandles,
+ CloseableRegistry cancelStreamRegistry) throws Exception {
throw new SuccessException();
}