You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2018/02/01 15:46:47 UTC
[15/22] flink git commit: [FLINK-8531] [checkpoints] (part 6) Tasks
resolve CheckpointStreamFactory from CheckpointStorage and Checkpoint
Location Reference to persist checkpoint data.
http://git-wip-us.apache.org/repos/asf/flink/blob/1887187f/flink-runtime/src/test/java/org/apache/flink/runtime/state/testutils/BackendForTestStream.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/testutils/BackendForTestStream.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/testutils/BackendForTestStream.java
new file mode 100644
index 0000000..fdb04cc
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/testutils/BackendForTestStream.java
@@ -0,0 +1,125 @@
+/*
+ * 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.testutils;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.state.CheckpointStorage;
+import org.apache.flink.runtime.state.CheckpointStorageLocation;
+import org.apache.flink.runtime.state.CheckpointStorageLocationReference;
+import org.apache.flink.runtime.state.CheckpointStreamFactory;
+import org.apache.flink.runtime.state.CheckpointStreamFactory.CheckpointStateOutputStream;
+import org.apache.flink.runtime.state.StreamStateHandle;
+import org.apache.flink.runtime.state.memory.MemoryStateBackend;
+import org.apache.flink.util.function.SupplierWithException;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * A test backends that allows you to supply a specific test stream.
+ */
+@SuppressWarnings({"serial"})
+public class BackendForTestStream extends MemoryStateBackend {
+
+ private static final long serialVersionUID = 1L;
+
+ private final TestFactory streamFactory;
+
+ public BackendForTestStream(TestFactory streamFactory) {
+ this.streamFactory = checkNotNull(streamFactory);
+ }
+
+ public BackendForTestStream(StreamFactory streamSupplier) {
+ this(new TestFactory(streamSupplier));
+ }
+
+ // make no reconfiguration!
+ @Override
+ public MemoryStateBackend configure(Configuration config) {
+ return this;
+ }
+
+ @Override
+ public CheckpointStorage createCheckpointStorage(JobID jobId) throws IOException {
+ return new TestCheckpointStorage();
+ }
+
+ // ------------------------------------------------------------------------
+
+ public interface StreamFactory
+ extends SupplierWithException<CheckpointStateOutputStream, Exception>, java.io.Serializable {}
+
+ // ------------------------------------------------------------------------
+
+ private final class TestCheckpointStorage implements CheckpointStorage {
+
+ @Override
+ public boolean supportsHighlyAvailableStorage() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean hasDefaultSavepointLocation() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public StreamStateHandle resolveCheckpoint(String pointer) throws IOException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public CheckpointStorageLocation initializeLocationForCheckpoint(long checkpointId) throws IOException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public CheckpointStorageLocation initializeLocationForSavepoint(long checkpointId, @Nullable String externalLocationPointer) throws IOException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public CheckpointStreamFactory resolveCheckpointStorageLocation(long checkpointId, CheckpointStorageLocationReference reference) throws IOException {
+ return streamFactory;
+ }
+
+ @Override
+ public CheckpointStateOutputStream createTaskOwnedStateStream() throws IOException {
+ throw new UnsupportedOperationException();
+ }
+ }
+
+ private static final class TestFactory implements CheckpointStreamFactory, java.io.Serializable {
+
+ private final StreamFactory streamFactory;
+
+ TestFactory(StreamFactory streamFactory) {
+ this.streamFactory = streamFactory;
+ }
+
+ @Override
+ public CheckpointStateOutputStream createCheckpointStateOutputStream(long checkpointID, long timestamp) throws Exception {
+ return streamFactory.get();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/1887187f/flink-runtime/src/test/java/org/apache/flink/runtime/state/testutils/TestCheckpointStreamFactory.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/testutils/TestCheckpointStreamFactory.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/testutils/TestCheckpointStreamFactory.java
new file mode 100644
index 0000000..57a1d3b
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/testutils/TestCheckpointStreamFactory.java
@@ -0,0 +1,43 @@
+/*
+ * 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.testutils;
+
+import org.apache.flink.runtime.state.CheckpointStreamFactory;
+
+import java.util.function.Supplier;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * A bridge between factory lambdas (like {@link Supplier} and the
+ * CheckpointStreamFactory class.
+ */
+public class TestCheckpointStreamFactory implements CheckpointStreamFactory {
+
+ private final Supplier<CheckpointStateOutputStream> supplier;
+
+ public TestCheckpointStreamFactory(Supplier<CheckpointStateOutputStream> supplier) {
+ this.supplier = checkNotNull(supplier);
+ }
+
+ @Override
+ public CheckpointStateOutputStream createCheckpointStateOutputStream(long checkpointId, long timestamp) {
+ return supplier.get();
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/1887187f/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java
index c6bc144..7db157c 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java
@@ -34,7 +34,6 @@ import org.apache.flink.metrics.Counter;
import org.apache.flink.metrics.Gauge;
import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.runtime.checkpoint.CheckpointOptions;
-import org.apache.flink.runtime.checkpoint.CheckpointType;
import org.apache.flink.runtime.execution.Environment;
import org.apache.flink.runtime.jobgraph.OperatorID;
import org.apache.flink.runtime.metrics.groups.OperatorMetricGroup;
@@ -72,9 +71,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.Closeable;
-import java.io.IOException;
import java.io.Serializable;
-import java.nio.charset.StandardCharsets;
import java.util.ConcurrentModificationException;
import java.util.HashMap;
import java.util.Map;
@@ -119,11 +116,6 @@ public abstract class AbstractStreamOperator<OUT>
/** The runtime context for UDFs. */
private transient StreamingRuntimeContext runtimeContext;
- // ----------------- general state -------------------
-
- /** The factory that give this operator access to checkpoint storage. */
- private transient CheckpointStreamFactory checkpointStreamFactory;
-
// ---------------- key/value state ------------------
/**
@@ -243,7 +235,6 @@ public abstract class AbstractStreamOperator<OUT>
}
timeServiceManager = context.internalTimerServiceManager();
- checkpointStreamFactory = context.checkpointStreamFactory();
CloseableIterable<KeyGroupStatePartitionStreamProvider> keyedStateInputs = context.rawKeyedStateInputs();
CloseableIterable<StatePartitionStreamProvider> operatorStateInputs = context.rawOperatorStateInputs();
@@ -352,15 +343,17 @@ public abstract class AbstractStreamOperator<OUT>
}
@Override
- public final OperatorSnapshotResult snapshotState(long checkpointId, long timestamp, CheckpointOptions checkpointOptions) throws Exception {
+ public final OperatorSnapshotResult snapshotState(
+ long checkpointId,
+ long timestamp,
+ CheckpointOptions checkpointOptions,
+ CheckpointStreamFactory factory) throws Exception {
KeyGroupRange keyGroupRange = null != keyedStateBackend ?
keyedStateBackend.getKeyGroupRange() : KeyGroupRange.EMPTY_KEY_GROUP_RANGE;
OperatorSnapshotResult snapshotInProgress = new OperatorSnapshotResult();
- CheckpointStreamFactory factory = getCheckpointStreamFactory(checkpointOptions);
-
try (StateSnapshotContextSynchronousImpl snapshotContext = new StateSnapshotContextSynchronousImpl(
checkpointId,
timestamp,
@@ -450,37 +443,6 @@ public abstract class AbstractStreamOperator<OUT>
}
}
- /**
- * Returns a checkpoint stream factory for the provided options.
- *
- * <p>For {@link CheckpointType#CHECKPOINT} this returns the shared
- * factory of this operator.
- *
- * <p>For {@link CheckpointType#SAVEPOINT} it creates a custom factory per
- * savepoint.
- *
- * @param checkpointOptions Options for the checkpoint
- * @return Checkpoint stream factory for the checkpoints
- * @throws IOException Failures while creating a new stream factory are forwarded
- */
- @VisibleForTesting
- CheckpointStreamFactory getCheckpointStreamFactory(CheckpointOptions checkpointOptions) throws IOException {
- CheckpointType checkpointType = checkpointOptions.getCheckpointType();
- if (checkpointType == CheckpointType.CHECKPOINT) {
- return checkpointStreamFactory;
- } else if (checkpointType == CheckpointType.SAVEPOINT) {
-
- // temporary fix: hard-code back conversion of the location reference to a string
- String targetAsString = new String(
- checkpointOptions.getTargetLocation().getReferenceBytes(),
- StandardCharsets.UTF_8);
-
- return container.createSavepointStreamFactory(this, targetAsString);
- } else {
- throw new IllegalStateException("Unknown checkpoint type " + checkpointType);
- }
- }
-
// ------------------------------------------------------------------------
// Properties and Services
// ------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/1887187f/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamOperator.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamOperator.java
index 07adfe4..8450396 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamOperator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamOperator.java
@@ -22,6 +22,7 @@ import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.runtime.checkpoint.CheckpointOptions;
import org.apache.flink.runtime.jobgraph.OperatorID;
import org.apache.flink.runtime.state.CheckpointListener;
+import org.apache.flink.runtime.state.CheckpointStreamFactory;
import org.apache.flink.streaming.api.graph.StreamConfig;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.streaming.runtime.tasks.StreamTask;
@@ -101,7 +102,8 @@ public interface StreamOperator<OUT> extends CheckpointListener, KeyContext, Ser
OperatorSnapshotResult snapshotState(
long checkpointId,
long timestamp,
- CheckpointOptions checkpointOptions) throws Exception;
+ CheckpointOptions checkpointOptions,
+ CheckpointStreamFactory storageLocation) throws Exception;
/**
* Provides a context to initialize all state in the operator.
http://git-wip-us.apache.org/repos/asf/flink/blob/1887187f/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamOperatorStateContext.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamOperatorStateContext.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamOperatorStateContext.java
index 52afd2a..75ead44 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamOperatorStateContext.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamOperatorStateContext.java
@@ -19,7 +19,6 @@
package org.apache.flink.streaming.api.operators;
import org.apache.flink.runtime.state.AbstractKeyedStateBackend;
-import org.apache.flink.runtime.state.CheckpointStreamFactory;
import org.apache.flink.runtime.state.KeyGroupStatePartitionStreamProvider;
import org.apache.flink.runtime.state.OperatorStateBackend;
import org.apache.flink.runtime.state.StatePartitionStreamProvider;
@@ -54,11 +53,6 @@ public interface StreamOperatorStateContext {
InternalTimeServiceManager<?, ?> internalTimerServiceManager();
/**
- * Returns the checkpoint stream factory for the stream operator.
- */
- CheckpointStreamFactory checkpointStreamFactory();
-
- /**
* Returns an iterable to obtain input streams for previously stored operator state partitions that are assigned to
* this stream operator.
*/
http://git-wip-us.apache.org/repos/asf/flink/blob/1887187f/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamTaskStateInitializerImpl.java
----------------------------------------------------------------------
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 ff00a8c..9afd2b7 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
@@ -28,7 +28,6 @@ import org.apache.flink.runtime.checkpoint.OperatorSubtaskState;
import org.apache.flink.runtime.execution.Environment;
import org.apache.flink.runtime.jobgraph.OperatorID;
import org.apache.flink.runtime.state.AbstractKeyedStateBackend;
-import org.apache.flink.runtime.state.CheckpointStreamFactory;
import org.apache.flink.runtime.state.DefaultOperatorStateBackend;
import org.apache.flink.runtime.state.KeyGroupRange;
import org.apache.flink.runtime.state.KeyGroupRangeAssignment;
@@ -127,8 +126,7 @@ public class StreamTaskStateInitializerImpl implements StreamTaskStateInitialize
OperatorStateBackend operatorStateBackend = null;
CloseableIterable<KeyGroupStatePartitionStreamProvider> rawKeyedStateInputs = null;
CloseableIterable<StatePartitionStreamProvider> rawOperatorStateInputs = null;
- CheckpointStreamFactory checkpointStreamFactory = null;
- InternalTimeServiceManager<?, ?> timeServiceManager = null;
+ InternalTimeServiceManager<?, ?> timeServiceManager;
try {
@@ -152,9 +150,6 @@ public class StreamTaskStateInitializerImpl implements StreamTaskStateInitialize
rawOperatorStateInputs = rawOperatorStateInputs(operatorSubtaskStateFromJobManager);
streamTaskCloseableRegistry.registerCloseable(rawOperatorStateInputs);
- // -------------- Checkpoint Stream Factory --------------
- checkpointStreamFactory = streamFactory(operatorIdentifierText);
-
// -------------- Internal Timer Service Manager --------------
timeServiceManager = internalTimeServiceManager(keyedStatedBackend, keyContext, rawKeyedStateInputs);
@@ -166,8 +161,7 @@ public class StreamTaskStateInitializerImpl implements StreamTaskStateInitialize
keyedStatedBackend,
timeServiceManager,
rawOperatorStateInputs,
- rawKeyedStateInputs,
- checkpointStreamFactory);
+ rawKeyedStateInputs);
} catch (Exception ex) {
// cleanup if something went wrong before results got published.
@@ -259,10 +253,6 @@ public class StreamTaskStateInitializerImpl implements StreamTaskStateInitialize
backendCloseableRegistry);
}
- protected CheckpointStreamFactory streamFactory(String operatorIdentifierText) throws IOException {
- return stateBackend.createStreamFactory(environment.getJobID(), operatorIdentifierText);
- }
-
protected CloseableIterable<StatePartitionStreamProvider> rawOperatorStateInputs(
OperatorSubtaskState operatorSubtaskStateFromJobManager) {
@@ -589,16 +579,13 @@ public class StreamTaskStateInitializerImpl implements StreamTaskStateInitialize
private final CloseableIterable<StatePartitionStreamProvider> rawOperatorStateInputs;
private final CloseableIterable<KeyGroupStatePartitionStreamProvider> rawKeyedStateInputs;
- private final CheckpointStreamFactory checkpointStreamFactory;
-
StreamOperatorStateContextImpl(
boolean restored,
OperatorStateBackend operatorStateBackend,
AbstractKeyedStateBackend<?> keyedStateBackend,
InternalTimeServiceManager<?, ?> internalTimeServiceManager,
CloseableIterable<StatePartitionStreamProvider> rawOperatorStateInputs,
- CloseableIterable<KeyGroupStatePartitionStreamProvider> rawKeyedStateInputs,
- CheckpointStreamFactory checkpointStreamFactory) {
+ CloseableIterable<KeyGroupStatePartitionStreamProvider> rawKeyedStateInputs) {
this.restored = restored;
this.operatorStateBackend = operatorStateBackend;
@@ -606,7 +593,6 @@ public class StreamTaskStateInitializerImpl implements StreamTaskStateInitialize
this.internalTimeServiceManager = internalTimeServiceManager;
this.rawOperatorStateInputs = rawOperatorStateInputs;
this.rawKeyedStateInputs = rawKeyedStateInputs;
- this.checkpointStreamFactory = checkpointStreamFactory;
}
@Override
@@ -630,11 +616,6 @@ public class StreamTaskStateInitializerImpl implements StreamTaskStateInitialize
}
@Override
- public CheckpointStreamFactory checkpointStreamFactory() {
- return checkpointStreamFactory;
- }
-
- @Override
public CloseableIterable<StatePartitionStreamProvider> rawOperatorStateInputs() {
return rawOperatorStateInputs;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/1887187f/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericWriteAheadSink.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericWriteAheadSink.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericWriteAheadSink.java
index 89ef327..27db126 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericWriteAheadSink.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericWriteAheadSink.java
@@ -24,6 +24,7 @@ import org.apache.flink.core.fs.FSDataInputStream;
import org.apache.flink.core.memory.DataInputViewStreamWrapper;
import org.apache.flink.core.memory.DataOutputViewStreamWrapper;
import org.apache.flink.runtime.io.disk.InputViewIterator;
+import org.apache.flink.runtime.state.CheckpointStorage;
import org.apache.flink.runtime.state.CheckpointStreamFactory;
import org.apache.flink.runtime.state.StateInitializationContext;
import org.apache.flink.runtime.state.StateSnapshotContext;
@@ -65,7 +66,7 @@ public abstract class GenericWriteAheadSink<IN> extends AbstractStreamOperator<I
protected final TypeSerializer<IN> serializer;
private transient CheckpointStreamFactory.CheckpointStateOutputStream out;
- private transient CheckpointStreamFactory checkpointStreamFactory;
+ private transient CheckpointStorage checkpointStorage;
private transient ListState<PendingCheckpoint> checkpointedState;
@@ -116,8 +117,7 @@ public abstract class GenericWriteAheadSink<IN> extends AbstractStreamOperator<I
committer.setOperatorId(id);
committer.open();
- checkpointStreamFactory = getContainingTask()
- .createCheckpointStreamFactory(this);
+ checkpointStorage = getContainingTask().getCheckpointStorage();
cleanRestoredHandles();
}
@@ -274,7 +274,7 @@ public abstract class GenericWriteAheadSink<IN> extends AbstractStreamOperator<I
IN value = element.getValue();
// generate initial operator state
if (out == null) {
- out = checkpointStreamFactory.createCheckpointStateOutputStream(0, 0);
+ out = checkpointStorage.createTaskOwnedStateStream();
}
serializer.serialize(value, new DataOutputViewStreamWrapper(out));
}
http://git-wip-us.apache.org/repos/asf/flink/blob/1887187f/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
index 801e990..4a85fb9 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
@@ -34,6 +34,7 @@ import org.apache.flink.runtime.io.network.api.serialization.EventSerializer;
import org.apache.flink.runtime.io.network.api.writer.ResultPartitionWriter;
import org.apache.flink.runtime.jobgraph.OperatorID;
import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
+import org.apache.flink.runtime.state.CheckpointStorage;
import org.apache.flink.runtime.state.CheckpointStreamFactory;
import org.apache.flink.runtime.state.StateBackend;
import org.apache.flink.runtime.state.StateBackendLoader;
@@ -61,7 +62,6 @@ import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
import java.io.Closeable;
-import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
@@ -141,6 +141,9 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>>
/** Our state backend. We use this to create checkpoint streams and a keyed state backend. */
protected StateBackend stateBackend;
+ /** The external storage where checkpoint data is persisted. */
+ private CheckpointStorage checkpointStorage;
+
/**
* The internal {@link ProcessingTimeService} used to define the current
* processing time (default = {@code System.currentTimeMillis()}) and
@@ -246,6 +249,7 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>>
asynchronousCheckpointExceptionHandler = new AsyncCheckpointExceptionHandler(this);
stateBackend = createStateBackend();
+ checkpointStorage = stateBackend.createCheckpointStorage(getEnvironment().getJobID());
accumulatorMap = getEnvironment().getAccumulatorRegistry().getUserMap();
@@ -527,6 +531,10 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>>
return lock;
}
+ public CheckpointStorage getCheckpointStorage() {
+ return checkpointStorage;
+ }
+
public StreamConfig getConfiguration() {
return configuration;
}
@@ -686,10 +694,15 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>>
CheckpointOptions checkpointOptions,
CheckpointMetrics checkpointMetrics) throws Exception {
+ CheckpointStreamFactory storage = checkpointStorage.resolveCheckpointStorageLocation(
+ checkpointMetaData.getCheckpointId(),
+ checkpointOptions.getTargetLocation());
+
CheckpointingOperation checkpointingOperation = new CheckpointingOperation(
this,
checkpointMetaData,
checkpointOptions,
+ storage,
checkpointMetrics);
checkpointingOperation.executeCheckpointing();
@@ -720,27 +733,6 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>>
LOG);
}
- /**
- * This is only visible because
- * {@link org.apache.flink.streaming.runtime.operators.GenericWriteAheadSink} uses the
- * checkpoint stream factory to write write-ahead logs. <b>This should not be used for
- * anything else.</b>
- */
- public CheckpointStreamFactory createCheckpointStreamFactory(
- StreamOperator<?> operator) throws IOException {
- return stateBackend.createStreamFactory(
- getEnvironment().getJobID(),
- createOperatorIdentifier(operator));
- }
-
- public CheckpointStreamFactory createSavepointStreamFactory(
- StreamOperator<?> operator, String targetLocation) throws IOException {
- return stateBackend.createSavepointStreamFactory(
- getEnvironment().getJobID(),
- createOperatorIdentifier(operator),
- targetLocation);
- }
-
protected CheckpointExceptionHandlerFactory createCheckpointExceptionHandlerFactory() {
return new CheckpointExceptionHandlerFactory();
}
@@ -883,7 +875,6 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>>
checkpointMetaData.getCheckpointId());
}
} catch (Exception e) {
- e.printStackTrace();
// the state is completed if an exception occurred in the acknowledgeCheckpoint call
// in order to clean up, we have to set it to RUNNING again.
asyncCheckpointState.compareAndSet(
@@ -960,6 +951,7 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>>
private final CheckpointMetaData checkpointMetaData;
private final CheckpointOptions checkpointOptions;
private final CheckpointMetrics checkpointMetrics;
+ private final CheckpointStreamFactory storageLocation;
private final StreamOperator<?>[] allOperators;
@@ -974,12 +966,14 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>>
StreamTask<?, ?> owner,
CheckpointMetaData checkpointMetaData,
CheckpointOptions checkpointOptions,
+ CheckpointStreamFactory checkpointStorageLocation,
CheckpointMetrics checkpointMetrics) {
this.owner = Preconditions.checkNotNull(owner);
this.checkpointMetaData = Preconditions.checkNotNull(checkpointMetaData);
this.checkpointOptions = Preconditions.checkNotNull(checkpointOptions);
this.checkpointMetrics = Preconditions.checkNotNull(checkpointMetrics);
+ this.storageLocation = Preconditions.checkNotNull(checkpointStorageLocation);
this.allOperators = owner.operatorChain.getAllOperators();
this.operatorSnapshotsInProgress = new HashMap<>(allOperators.length);
}
@@ -1050,7 +1044,8 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>>
OperatorSnapshotResult snapshotInProgress = op.snapshotState(
checkpointMetaData.getCheckpointId(),
checkpointMetaData.getTimestamp(),
- checkpointOptions);
+ checkpointOptions,
+ storageLocation);
operatorSnapshotsInProgress.put(op.getOperatorID(), snapshotInProgress);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/1887187f/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/AbstractStreamOperatorTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/AbstractStreamOperatorTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/AbstractStreamOperatorTest.java
index ee0f8de..97a0182 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/AbstractStreamOperatorTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/AbstractStreamOperatorTest.java
@@ -35,6 +35,7 @@ import org.apache.flink.runtime.state.OperatorStateHandle;
import org.apache.flink.runtime.state.StateSnapshotContextSynchronousImpl;
import org.apache.flink.runtime.state.VoidNamespace;
import org.apache.flink.runtime.state.VoidNamespaceSerializer;
+import org.apache.flink.runtime.state.memory.MemCheckpointStreamFactory;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.streaming.runtime.tasks.OperatorStateHandles;
import org.apache.flink.streaming.runtime.tasks.StreamTask;
@@ -501,10 +502,14 @@ public class AbstractStreamOperatorTest {
when(containingTask.getCancelables()).thenReturn(closeableRegistry);
AbstractStreamOperator<Void> operator = mock(AbstractStreamOperator.class);
- when(operator.snapshotState(anyLong(), anyLong(), any(CheckpointOptions.class))).thenCallRealMethod();
+ when(operator.snapshotState(anyLong(), anyLong(), any(CheckpointOptions.class), any(CheckpointStreamFactory.class))).thenCallRealMethod();
doReturn(containingTask).when(operator).getContainingTask();
- operator.snapshotState(checkpointId, timestamp, CheckpointOptions.forCheckpointWithDefaultLocation());
+ operator.snapshotState(
+ checkpointId,
+ timestamp,
+ CheckpointOptions.forCheckpointWithDefaultLocation(),
+ new MemCheckpointStreamFactory(Integer.MAX_VALUE));
verify(context).close();
}
@@ -530,14 +535,18 @@ public class AbstractStreamOperatorTest {
when(containingTask.getCancelables()).thenReturn(closeableRegistry);
AbstractStreamOperator<Void> operator = mock(AbstractStreamOperator.class);
- when(operator.snapshotState(anyLong(), anyLong(), any(CheckpointOptions.class))).thenCallRealMethod();
+ when(operator.snapshotState(anyLong(), anyLong(), any(CheckpointOptions.class), any(CheckpointStreamFactory.class))).thenCallRealMethod();
doReturn(containingTask).when(operator).getContainingTask();
// lets fail when calling the actual snapshotState method
doThrow(failingException).when(operator).snapshotState(eq(context));
try {
- operator.snapshotState(checkpointId, timestamp, CheckpointOptions.forCheckpointWithDefaultLocation());
+ operator.snapshotState(
+ checkpointId,
+ timestamp,
+ CheckpointOptions.forCheckpointWithDefaultLocation(),
+ new MemCheckpointStreamFactory(Integer.MAX_VALUE));
fail("Exception expected.");
} catch (Exception e) {
assertEquals(failingException, e.getCause());
@@ -577,16 +586,11 @@ public class AbstractStreamOperatorTest {
when(containingTask.getCancelables()).thenReturn(closeableRegistry);
AbstractStreamOperator<Void> operator = mock(AbstractStreamOperator.class);
- when(operator.snapshotState(anyLong(), anyLong(), any(CheckpointOptions.class))).thenCallRealMethod();
+ when(operator.snapshotState(anyLong(), anyLong(), any(CheckpointOptions.class), any(CheckpointStreamFactory.class))).thenCallRealMethod();
doCallRealMethod().when(operator).close();
doCallRealMethod().when(operator).dispose();
- // The amount of mocking in this test makes it necessary to make the
- // getCheckpointStreamFactory method visible for the test and to
- // overwrite its behaviour.
- when(operator.getCheckpointStreamFactory(any(CheckpointOptions.class))).thenReturn(streamFactory);
-
doReturn(containingTask).when(operator).getContainingTask();
RunnableFuture<OperatorStateHandle> futureManagedOperatorStateHandle = mock(RunnableFuture.class);
@@ -602,7 +606,7 @@ public class AbstractStreamOperatorTest {
when(keyedStateBackend.snapshot(
eq(checkpointId),
eq(timestamp),
- eq(streamFactory),
+ any(CheckpointStreamFactory.class),
eq(CheckpointOptions.forCheckpointWithDefaultLocation()))).thenThrow(failingException);
closeableRegistry.registerCloseable(operatorStateBackend);
@@ -610,10 +614,13 @@ public class AbstractStreamOperatorTest {
Whitebox.setInternalState(operator, "operatorStateBackend", operatorStateBackend);
Whitebox.setInternalState(operator, "keyedStateBackend", keyedStateBackend);
- Whitebox.setInternalState(operator, "checkpointStreamFactory", streamFactory);
try {
- operator.snapshotState(checkpointId, timestamp, CheckpointOptions.forCheckpointWithDefaultLocation());
+ operator.snapshotState(
+ checkpointId,
+ timestamp,
+ CheckpointOptions.forCheckpointWithDefaultLocation(),
+ new MemCheckpointStreamFactory(Integer.MAX_VALUE));
fail("Exception expected.");
} catch (Exception e) {
assertEquals(failingException, e.getCause());
http://git-wip-us.apache.org/repos/asf/flink/blob/1887187f/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/AbstractUdfStreamOperatorLifecycleTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/AbstractUdfStreamOperatorLifecycleTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/AbstractUdfStreamOperatorLifecycleTest.java
index 84c74d6..f82e403 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/AbstractUdfStreamOperatorLifecycleTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/AbstractUdfStreamOperatorLifecycleTest.java
@@ -91,7 +91,7 @@ public class AbstractUdfStreamOperatorLifecycleTest {
"setup[class org.apache.flink.streaming.runtime.tasks.StreamTask, class " +
"org.apache.flink.streaming.api.graph.StreamConfig, interface " +
"org.apache.flink.streaming.api.operators.Output], " +
- "snapshotState[long, long, class org.apache.flink.runtime.checkpoint.CheckpointOptions]]";
+ "snapshotState[long, long, class org.apache.flink.runtime.checkpoint.CheckpointOptions, interface org.apache.flink.runtime.state.CheckpointStreamFactory]]";
private static final String ALL_METHODS_RICH_FUNCTION = "[close[], getIterationRuntimeContext[], getRuntimeContext[]" +
", open[class org.apache.flink.configuration.Configuration], setRuntimeContext[interface " +
http://git-wip-us.apache.org/repos/asf/flink/blob/1887187f/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/StreamTaskStateInitializerImplTest.java
----------------------------------------------------------------------
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 3495bc0..6b69707 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
@@ -33,7 +33,6 @@ import org.apache.flink.runtime.operators.testutils.DummyEnvironment;
import org.apache.flink.runtime.query.TaskKvStateRegistry;
import org.apache.flink.runtime.state.AbstractKeyedStateBackend;
import org.apache.flink.runtime.state.CheckpointStorage;
-import org.apache.flink.runtime.state.CheckpointStreamFactory;
import org.apache.flink.runtime.state.KeyGroupRange;
import org.apache.flink.runtime.state.KeyGroupStatePartitionStreamProvider;
import org.apache.flink.runtime.state.OperatorStateBackend;
@@ -52,8 +51,6 @@ import org.apache.flink.util.CloseableIterable;
import org.junit.Assert;
import org.junit.Test;
-import javax.annotation.Nullable;
-
import java.io.Closeable;
import java.io.IOException;
import java.util.Collections;
@@ -107,14 +104,9 @@ public class StreamTaskStateInitializerImplTest {
any(Environment.class),
any(String.class));
- verify(stateBackend).createStreamFactory(
- any(JobID.class),
- any(String.class));
-
OperatorStateBackend operatorStateBackend = stateContext.operatorStateBackend();
AbstractKeyedStateBackend<?> keyedStateBackend = stateContext.keyedStateBackend();
InternalTimeServiceManager<?, ?> timeServiceManager = stateContext.internalTimerServiceManager();
- CheckpointStreamFactory streamFactory = stateContext.checkpointStreamFactory();
CloseableIterable<KeyGroupStatePartitionStreamProvider> keyedStateInputs = stateContext.rawKeyedStateInputs();
CloseableIterable<StatePartitionStreamProvider> operatorStateInputs = stateContext.rawOperatorStateInputs();
@@ -122,7 +114,6 @@ public class StreamTaskStateInitializerImplTest {
Assert.assertNotNull(operatorStateBackend);
Assert.assertNotNull(keyedStateBackend);
Assert.assertNotNull(timeServiceManager);
- Assert.assertNotNull(streamFactory);
Assert.assertNotNull(keyedStateInputs);
Assert.assertNotNull(operatorStateInputs);
@@ -158,18 +149,6 @@ public class StreamTaskStateInitializerImplTest {
}
@Override
- public CheckpointStreamFactory createStreamFactory(
- JobID jobId, String operatorIdentifier) throws IOException {
- return mock(CheckpointStreamFactory.class);
- }
-
- @Override
- public CheckpointStreamFactory createSavepointStreamFactory(
- JobID jobId, String operatorIdentifier, @Nullable String targetLocation) throws IOException {
- return mock(CheckpointStreamFactory.class);
- }
-
- @Override
public <K> AbstractKeyedStateBackend<K> createKeyedStateBackend(
Environment env,
JobID jobID,
@@ -243,14 +222,9 @@ public class StreamTaskStateInitializerImplTest {
any(Environment.class),
any(String.class));
- verify(mockingBackend).createStreamFactory(
- any(JobID.class),
- any(String.class));
-
OperatorStateBackend operatorStateBackend = stateContext.operatorStateBackend();
AbstractKeyedStateBackend<?> keyedStateBackend = stateContext.keyedStateBackend();
InternalTimeServiceManager<?, ?> timeServiceManager = stateContext.internalTimerServiceManager();
- CheckpointStreamFactory streamFactory = stateContext.checkpointStreamFactory();
CloseableIterable<KeyGroupStatePartitionStreamProvider> keyedStateInputs = stateContext.rawKeyedStateInputs();
CloseableIterable<StatePartitionStreamProvider> operatorStateInputs = stateContext.rawOperatorStateInputs();
@@ -260,7 +234,6 @@ public class StreamTaskStateInitializerImplTest {
Assert.assertNotNull(keyedStateBackend);
// this is deactivated on purpose so that it does not attempt to consume the raw keyed state.
Assert.assertNull(timeServiceManager);
- Assert.assertNotNull(streamFactory);
Assert.assertNotNull(keyedStateInputs);
Assert.assertNotNull(operatorStateInputs);
http://git-wip-us.apache.org/repos/asf/flink/blob/1887187f/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTerminationTest.java
----------------------------------------------------------------------
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 ab97a18..f88857b 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
@@ -60,6 +60,7 @@ import org.apache.flink.runtime.state.OperatorStateHandle;
import org.apache.flink.runtime.state.StateBackend;
import org.apache.flink.runtime.state.StreamStateHandle;
import org.apache.flink.runtime.state.TestTaskStateManager;
+import org.apache.flink.runtime.state.memory.MemoryBackendCheckpointStorage;
import org.apache.flink.runtime.taskmanager.CheckpointResponder;
import org.apache.flink.runtime.taskmanager.Task;
import org.apache.flink.runtime.taskmanager.TaskManagerActions;
@@ -75,8 +76,6 @@ import org.apache.flink.util.TestLogger;
import org.junit.Test;
-import javax.annotation.Nullable;
-
import java.io.IOException;
import java.util.Collections;
import java.util.concurrent.Callable;
@@ -258,17 +257,7 @@ public class StreamTaskTerminationTest extends TestLogger {
@Override
public CheckpointStorage createCheckpointStorage(JobID jobId) throws IOException {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public CheckpointStreamFactory createStreamFactory(JobID jobId, String operatorIdentifier) throws IOException {
- return mock(CheckpointStreamFactory.class);
- }
-
- @Override
- public CheckpointStreamFactory createSavepointStreamFactory(JobID jobId, String operatorIdentifier, @Nullable String targetLocation) throws IOException {
- return null;
+ return new MemoryBackendCheckpointStorage(jobId, null, null, Integer.MAX_VALUE);
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/1887187f/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java
index 5fb683f..9ee35ee 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java
@@ -64,6 +64,7 @@ import org.apache.flink.runtime.operators.testutils.MockInputSplitProvider;
import org.apache.flink.runtime.query.TaskKvStateRegistry;
import org.apache.flink.runtime.state.AbstractKeyedStateBackend;
import org.apache.flink.runtime.state.AbstractStateBackend;
+import org.apache.flink.runtime.state.CheckpointStorage;
import org.apache.flink.runtime.state.CheckpointStreamFactory;
import org.apache.flink.runtime.state.DoneFuture;
import org.apache.flink.runtime.state.KeyGroupRange;
@@ -74,11 +75,11 @@ import org.apache.flink.runtime.state.OperatorStateHandle;
import org.apache.flink.runtime.state.StateBackendFactory;
import org.apache.flink.runtime.state.StateInitializationContext;
import org.apache.flink.runtime.state.StatePartitionStreamProvider;
-import org.apache.flink.runtime.state.StreamStateHandle;
import org.apache.flink.runtime.state.TaskLocalStateStore;
import org.apache.flink.runtime.state.TaskStateManager;
import org.apache.flink.runtime.state.TaskStateManagerImpl;
import org.apache.flink.runtime.state.TestTaskStateManager;
+import org.apache.flink.runtime.state.memory.MemoryBackendCheckpointStorage;
import org.apache.flink.runtime.state.memory.MemoryStateBackend;
import org.apache.flink.runtime.taskmanager.CheckpointResponder;
import org.apache.flink.runtime.taskmanager.Task;
@@ -346,9 +347,9 @@ public class StreamTaskTest extends TestLogger {
final Exception testException = new Exception("Test exception");
- when(streamOperator1.snapshotState(anyLong(), anyLong(), any(CheckpointOptions.class))).thenReturn(operatorSnapshotResult1);
- when(streamOperator2.snapshotState(anyLong(), anyLong(), any(CheckpointOptions.class))).thenReturn(operatorSnapshotResult2);
- when(streamOperator3.snapshotState(anyLong(), anyLong(), any(CheckpointOptions.class))).thenThrow(testException);
+ when(streamOperator1.snapshotState(anyLong(), anyLong(), any(CheckpointOptions.class), any(CheckpointStreamFactory.class))).thenReturn(operatorSnapshotResult1);
+ when(streamOperator2.snapshotState(anyLong(), anyLong(), any(CheckpointOptions.class), any(CheckpointStreamFactory.class))).thenReturn(operatorSnapshotResult2);
+ when(streamOperator3.snapshotState(anyLong(), anyLong(), any(CheckpointOptions.class), any(CheckpointStreamFactory.class))).thenThrow(testException);
OperatorID operatorID1 = new OperatorID();
OperatorID operatorID2 = new OperatorID();
@@ -369,6 +370,7 @@ public class StreamTaskTest extends TestLogger {
Whitebox.setInternalState(streamTask, "operatorChain", operatorChain);
Whitebox.setInternalState(streamTask, "cancelables", new CloseableRegistry());
Whitebox.setInternalState(streamTask, "configuration", new StreamConfig(new Configuration()));
+ Whitebox.setInternalState(streamTask, "checkpointStorage", new MemoryBackendCheckpointStorage(new JobID(), null, null, Integer.MAX_VALUE));
CheckpointExceptionHandlerFactory checkpointExceptionHandlerFactory = new CheckpointExceptionHandlerFactory();
CheckpointExceptionHandler checkpointExceptionHandler =
@@ -423,9 +425,9 @@ public class StreamTaskTest extends TestLogger {
when(operatorSnapshotResult3.getOperatorStateRawFuture()).thenReturn(failingFuture);
- when(streamOperator1.snapshotState(anyLong(), anyLong(), any(CheckpointOptions.class))).thenReturn(operatorSnapshotResult1);
- when(streamOperator2.snapshotState(anyLong(), anyLong(), any(CheckpointOptions.class))).thenReturn(operatorSnapshotResult2);
- when(streamOperator3.snapshotState(anyLong(), anyLong(), any(CheckpointOptions.class))).thenReturn(operatorSnapshotResult3);
+ when(streamOperator1.snapshotState(anyLong(), anyLong(), any(CheckpointOptions.class), any(CheckpointStreamFactory.class))).thenReturn(operatorSnapshotResult1);
+ when(streamOperator2.snapshotState(anyLong(), anyLong(), any(CheckpointOptions.class), any(CheckpointStreamFactory.class))).thenReturn(operatorSnapshotResult2);
+ when(streamOperator3.snapshotState(anyLong(), anyLong(), any(CheckpointOptions.class), any(CheckpointStreamFactory.class))).thenReturn(operatorSnapshotResult3);
OperatorID operatorID1 = new OperatorID();
OperatorID operatorID2 = new OperatorID();
@@ -445,6 +447,7 @@ public class StreamTaskTest extends TestLogger {
Whitebox.setInternalState(streamTask, "cancelables", new CloseableRegistry());
Whitebox.setInternalState(streamTask, "asyncOperationsThreadPool", new DirectExecutorService());
Whitebox.setInternalState(streamTask, "configuration", new StreamConfig(new Configuration()));
+ Whitebox.setInternalState(streamTask, "checkpointStorage", new MemoryBackendCheckpointStorage(new JobID(), null, null, Integer.MAX_VALUE));
CheckpointExceptionHandlerFactory checkpointExceptionHandlerFactory = new CheckpointExceptionHandlerFactory();
CheckpointExceptionHandler checkpointExceptionHandler =
@@ -529,24 +532,14 @@ public class StreamTaskTest extends TestLogger {
new DoneFuture<>(managedOperatorStateHandle),
new DoneFuture<>(rawOperatorStateHandle));
- when(streamOperator.snapshotState(anyLong(), anyLong(), any(CheckpointOptions.class))).thenReturn(operatorSnapshotResult);
+ when(streamOperator.snapshotState(anyLong(), anyLong(), any(CheckpointOptions.class), any(CheckpointStreamFactory.class))).thenReturn(operatorSnapshotResult);
StreamOperator<?>[] streamOperators = {streamOperator};
OperatorChain<Void, AbstractStreamOperator<Void>> operatorChain = mock(OperatorChain.class);
when(operatorChain.getAllOperators()).thenReturn(streamOperators);
- StreamStateHandle streamStateHandle = mock(StreamStateHandle.class);
-
- CheckpointStreamFactory.CheckpointStateOutputStream outStream = mock(CheckpointStreamFactory.CheckpointStateOutputStream.class);
-
- when(outStream.closeAndGetHandle()).thenReturn(streamStateHandle);
-
- CheckpointStreamFactory mockStreamFactory = mock(CheckpointStreamFactory.class);
- when(mockStreamFactory.createCheckpointStateOutputStream(anyLong(), anyLong())).thenReturn(outStream);
-
- AbstractStateBackend mockStateBackend = mock(AbstractStateBackend.class);
- when(mockStateBackend.createStreamFactory(any(JobID.class), anyString())).thenReturn(mockStreamFactory);
+ CheckpointStorage checkpointStorage = new MemoryBackendCheckpointStorage(new JobID(), null, null, Integer.MAX_VALUE);
Whitebox.setInternalState(streamTask, "isRunning", true);
Whitebox.setInternalState(streamTask, "lock", new Object());
@@ -554,7 +547,7 @@ public class StreamTaskTest extends TestLogger {
Whitebox.setInternalState(streamTask, "cancelables", new CloseableRegistry());
Whitebox.setInternalState(streamTask, "asyncOperationsThreadPool", Executors.newFixedThreadPool(1));
Whitebox.setInternalState(streamTask, "configuration", new StreamConfig(new Configuration()));
- Whitebox.setInternalState(streamTask, "stateBackend", mockStateBackend);
+ Whitebox.setInternalState(streamTask, "checkpointStorage", checkpointStorage);
streamTask.triggerCheckpoint(checkpointMetaData, CheckpointOptions.forCheckpointWithDefaultLocation());
@@ -658,24 +651,14 @@ public class StreamTaskTest extends TestLogger {
new DoneFuture<>(managedOperatorStateHandle),
new DoneFuture<>(rawOperatorStateHandle));
- when(streamOperator.snapshotState(anyLong(), anyLong(), any(CheckpointOptions.class))).thenReturn(operatorSnapshotResult);
+ when(streamOperator.snapshotState(anyLong(), anyLong(), any(CheckpointOptions.class), any(CheckpointStreamFactory.class))).thenReturn(operatorSnapshotResult);
StreamOperator<?>[] streamOperators = {streamOperator};
OperatorChain<Void, AbstractStreamOperator<Void>> operatorChain = mock(OperatorChain.class);
when(operatorChain.getAllOperators()).thenReturn(streamOperators);
- StreamStateHandle streamStateHandle = mock(StreamStateHandle.class);
-
- CheckpointStreamFactory.CheckpointStateOutputStream outStream = mock(CheckpointStreamFactory.CheckpointStateOutputStream.class);
-
- when(outStream.closeAndGetHandle()).thenReturn(streamStateHandle);
-
- CheckpointStreamFactory mockStreamFactory = mock(CheckpointStreamFactory.class);
- when(mockStreamFactory.createCheckpointStateOutputStream(anyLong(), anyLong())).thenReturn(outStream);
-
- AbstractStateBackend mockStateBackend = mock(AbstractStateBackend.class);
- when(mockStateBackend.createStreamFactory(any(JobID.class), anyString())).thenReturn(mockStreamFactory);
+ CheckpointStorage checkpointStorage = new MemoryBackendCheckpointStorage(new JobID(), null, null, Integer.MAX_VALUE);
ExecutorService executor = Executors.newFixedThreadPool(1);
@@ -685,7 +668,7 @@ public class StreamTaskTest extends TestLogger {
Whitebox.setInternalState(streamTask, "cancelables", new CloseableRegistry());
Whitebox.setInternalState(streamTask, "asyncOperationsThreadPool", executor);
Whitebox.setInternalState(streamTask, "configuration", new StreamConfig(new Configuration()));
- Whitebox.setInternalState(streamTask, "stateBackend", mockStateBackend);
+ Whitebox.setInternalState(streamTask, "checkpointStorage", checkpointStorage);
streamTask.triggerCheckpoint(checkpointMetaData, CheckpointOptions.forCheckpointWithDefaultLocation());
@@ -773,7 +756,7 @@ public class StreamTaskTest extends TestLogger {
// mock the returned empty snapshot result (all state handles are null)
OperatorSnapshotResult statelessOperatorSnapshotResult = new OperatorSnapshotResult();
- when(statelessOperator.snapshotState(anyLong(), anyLong(), any(CheckpointOptions.class)))
+ when(statelessOperator.snapshotState(anyLong(), anyLong(), any(CheckpointOptions.class), any(CheckpointStreamFactory.class)))
.thenReturn(statelessOperatorSnapshotResult);
// set up the task
@@ -787,6 +770,7 @@ public class StreamTaskTest extends TestLogger {
Whitebox.setInternalState(streamTask, "cancelables", new CloseableRegistry());
Whitebox.setInternalState(streamTask, "configuration", new StreamConfig(new Configuration()));
Whitebox.setInternalState(streamTask, "asyncOperationsThreadPool", Executors.newCachedThreadPool());
+ Whitebox.setInternalState(streamTask, "checkpointStorage", new MemoryBackendCheckpointStorage(new JobID(), null, null, Integer.MAX_VALUE));
streamTask.triggerCheckpoint(checkpointMetaData, CheckpointOptions.forCheckpointWithDefaultLocation());
checkpointCompletedLatch.await(30, TimeUnit.SECONDS);
@@ -904,7 +888,7 @@ public class StreamTaskTest extends TestLogger {
public Future<ExecutionState> notifyWhenExecutionState(ExecutionState executionState) {
synchronized (priorityQueue) {
if (this.executionState != null && this.executionState.ordinal() >= executionState.ordinal()) {
- return Futures.successful(executionState);
+ return Futures.<ExecutionState>successful(executionState);
} else {
Promise<ExecutionState> promise = new Promise.DefaultPromise<ExecutionState>();
@@ -1160,11 +1144,6 @@ public class StreamTaskTest extends TestLogger {
}
@Override
- public CheckpointStreamFactory checkpointStreamFactory() {
- return context.checkpointStreamFactory();
- }
-
- @Override
public CloseableIterable<StatePartitionStreamProvider> rawOperatorStateInputs() {
return replaceWithSpy(context.rawOperatorStateInputs());
}
http://git-wip-us.apache.org/repos/asf/flink/blob/1887187f/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/TaskCheckpointingBehaviourTest.java
----------------------------------------------------------------------
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 1525848..6df33b7 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
@@ -61,10 +61,12 @@ import org.apache.flink.runtime.state.DefaultOperatorStateBackend;
import org.apache.flink.runtime.state.OperatorStateBackend;
import org.apache.flink.runtime.state.OperatorStateCheckpointOutputStream;
import org.apache.flink.runtime.state.OperatorStateHandle;
+import org.apache.flink.runtime.state.StateBackend;
import org.apache.flink.runtime.state.StateSnapshotContext;
import org.apache.flink.runtime.state.StreamStateHandle;
import org.apache.flink.runtime.state.TestTaskStateManager;
import org.apache.flink.runtime.state.memory.MemoryStateBackend;
+import org.apache.flink.runtime.state.testutils.BackendForTestStream;
import org.apache.flink.runtime.taskmanager.CheckpointResponder;
import org.apache.flink.runtime.taskmanager.Task;
import org.apache.flink.runtime.taskmanager.TaskManagerActions;
@@ -128,8 +130,10 @@ public class TaskCheckpointingBehaviourTest extends TestLogger {
@Test
public void testBlockingNonInterruptibleCheckpoint() throws Exception {
+ StateBackend lockingStateBackend = new BackendForTestStream(LockingOutputStream::new);
+
Task task =
- createTask(new TestOperator(), new LockingStreamStateBackend(), mock(CheckpointResponder.class), true);
+ createTask(new TestOperator(), lockingStateBackend, mock(CheckpointResponder.class), true);
// start the task and wait until it is in "restore"
task.startTaskThread();
@@ -182,7 +186,7 @@ public class TaskCheckpointingBehaviourTest extends TestLogger {
private static Task createTask(
StreamOperator<?> op,
- AbstractStateBackend backend,
+ StateBackend backend,
CheckpointResponder checkpointResponder,
boolean failOnCheckpointErrors) throws IOException {
@@ -352,41 +356,9 @@ public class TaskCheckpointingBehaviourTest extends TestLogger {
}
// ------------------------------------------------------------------------
- // state backend with locking output stream.
+ // locking output stream.
// ------------------------------------------------------------------------
- private static class LockingStreamStateBackend extends MemoryStateBackend {
-
- private static final long serialVersionUID = 1L;
-
- @Override
- public CheckpointStreamFactory createStreamFactory(JobID jobId, String operatorIdentifier) throws IOException {
- return new LockingOutputStreamFactory();
- }
-
- @Override
- public OperatorStateBackend createOperatorStateBackend(Environment env, String operatorIdentifier) throws Exception {
- return new DefaultOperatorStateBackend(
- getClass().getClassLoader(),
- new ExecutionConfig(),
- true);
- }
-
- @Override
- public LockingStreamStateBackend configure(Configuration config) {
- // retain this instance, no re-configuration!
- return this;
- }
- }
-
- private static final class LockingOutputStreamFactory implements CheckpointStreamFactory {
-
- @Override
- public CheckpointStateOutputStream createCheckpointStateOutputStream(long checkpointID, long timestamp) {
- return new LockingOutputStream();
- }
- }
-
private static final class LockingOutputStream extends CheckpointStateOutputStream {
private final Object lock = new Object();
http://git-wip-us.apache.org/repos/asf/flink/blob/1887187f/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/AbstractStreamOperatorTestHarness.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/AbstractStreamOperatorTestHarness.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/AbstractStreamOperatorTestHarness.java
index 2602603..966d205 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/AbstractStreamOperatorTestHarness.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/AbstractStreamOperatorTestHarness.java
@@ -35,7 +35,8 @@ import org.apache.flink.runtime.execution.Environment;
import org.apache.flink.runtime.jobgraph.OperatorID;
import org.apache.flink.runtime.operators.testutils.MockEnvironment;
import org.apache.flink.runtime.operators.testutils.MockInputSplitProvider;
-import org.apache.flink.runtime.state.CheckpointStreamFactory;
+import org.apache.flink.runtime.state.CheckpointStorage;
+import org.apache.flink.runtime.state.CheckpointStorageLocationReference;
import org.apache.flink.runtime.state.KeyGroupRange;
import org.apache.flink.runtime.state.KeyedStateHandle;
import org.apache.flink.runtime.state.OperatorStateHandle;
@@ -67,6 +68,7 @@ import org.apache.flink.util.Preconditions;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
+import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
@@ -113,6 +115,7 @@ public class AbstractStreamOperatorTestHarness<OUT> implements AutoCloseable {
// use this as default for tests
protected StateBackend stateBackend = new MemoryStateBackend();
+ private CheckpointStorage checkpointStorage = stateBackend.createCheckpointStorage(new JobID());
private final Object checkpointLock;
@@ -219,18 +222,7 @@ public class AbstractStreamOperatorTestHarness<OUT> implements AutoCloseable {
}
}).when(mockTask).handleAsyncException(any(String.class), any(Throwable.class));
- try {
- doAnswer(new Answer<CheckpointStreamFactory>() {
- @Override
- public CheckpointStreamFactory answer(InvocationOnMock invocationOnMock) throws Throwable {
-
- final StreamOperator<?> operator = (StreamOperator<?>) invocationOnMock.getArguments()[0];
- return stateBackend.createStreamFactory(new JobID(), operator.getClass().getSimpleName());
- }
- }).when(mockTask).createCheckpointStreamFactory(any(StreamOperator.class));
- } catch (Exception e) {
- throw new RuntimeException(e.getMessage(), e);
- }
+ when(mockTask.getCheckpointStorage()).thenAnswer((invocationOnMock) -> this.checkpointStorage);
doAnswer(new Answer<ProcessingTimeService>() {
@Override
@@ -253,6 +245,12 @@ public class AbstractStreamOperatorTestHarness<OUT> implements AutoCloseable {
public void setStateBackend(StateBackend stateBackend) {
this.stateBackend = stateBackend;
+
+ try {
+ this.checkpointStorage = stateBackend.createCheckpointStorage(new JobID());
+ } catch (IOException e) {
+ throw new RuntimeException(e.getMessage(), e);
+ }
}
public Object getCheckpointLock() {
@@ -470,14 +468,15 @@ public class AbstractStreamOperatorTestHarness<OUT> implements AutoCloseable {
}
/**
- * Calls {@link StreamOperator#snapshotState(long, long, CheckpointOptions)}.
+ * Calls {@link StreamOperator#snapshotState(long, long, CheckpointOptions, org.apache.flink.runtime.state.CheckpointStreamFactory)}.
*/
public OperatorStateHandles snapshot(long checkpointId, long timestamp) throws Exception {
OperatorSnapshotResult operatorStateResult = operator.snapshotState(
checkpointId,
timestamp,
- CheckpointOptions.forCheckpointWithDefaultLocation());
+ CheckpointOptions.forCheckpointWithDefaultLocation(),
+ checkpointStorage.resolveCheckpointStorageLocation(checkpointId, CheckpointStorageLocationReference.getDefault()));
KeyedStateHandle keyedManaged = FutureUtil.runIfNotDoneAndGet(operatorStateResult.getKeyedStateManagedFuture());
KeyedStateHandle keyedRaw = FutureUtil.runIfNotDoneAndGet(operatorStateResult.getKeyedStateRawFuture());
http://git-wip-us.apache.org/repos/asf/flink/blob/1887187f/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/StateBackendITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/StateBackendITCase.java b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/StateBackendITCase.java
index 9c023d4..df8d40e 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
@@ -30,7 +30,6 @@ import org.apache.flink.runtime.execution.Environment;
import org.apache.flink.runtime.query.TaskKvStateRegistry;
import org.apache.flink.runtime.state.AbstractKeyedStateBackend;
import org.apache.flink.runtime.state.CheckpointStorage;
-import org.apache.flink.runtime.state.CheckpointStreamFactory;
import org.apache.flink.runtime.state.KeyGroupRange;
import org.apache.flink.runtime.state.OperatorStateBackend;
import org.apache.flink.runtime.state.StateBackend;
@@ -101,19 +100,7 @@ public class StateBackendITCase extends AbstractTestBase {
@Override
public CheckpointStorage createCheckpointStorage(JobID jobId) throws IOException {
- return new MemoryBackendCheckpointStorage(jobId);
- }
-
- @Override
- public CheckpointStreamFactory createStreamFactory(JobID jobId,
- String operatorIdentifier) throws IOException {
- throw new SuccessException();
- }
-
- @Override
- public CheckpointStreamFactory createSavepointStreamFactory(JobID jobId,
- String operatorIdentifier, String targetLocation) throws IOException {
- throw new SuccessException();
+ return new MemoryBackendCheckpointStorage(jobId, null, null, 1_000_000);
}
@Override