You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by pn...@apache.org on 2021/08/10 06:35:29 UTC
[flink] 09/11: [FLINK-23408] Emit end of data for
stop-with-savepoint --drain
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit e3010c16947ed8da2ecb7d89a3aa08dacecc524a
Author: Dawid Wysakowicz <dw...@apache.org>
AuthorDate: Thu Jul 29 16:12:56 2021 +0200
[FLINK-23408] Emit end of data for stop-with-savepoint --drain
---
.../connectors/kinesis/FlinkKinesisITCase.java | 2 +
.../flink/runtime/checkpoint/CheckpointType.java | 4 +
.../streaming/api/operators/SourceOperator.java | 81 +++++++---
.../streaming/api/operators/StreamSource.java | 19 +--
.../runtime/io/StreamTaskSourceInput.java | 4 +
.../runtime/tasks/MultipleInputStreamTask.java | 27 +++-
.../streaming/runtime/tasks/OperatorChain.java | 10 +-
.../runtime/tasks/SourceOperatorStreamTask.java | 35 +++--
.../streaming/runtime/tasks/SourceStreamTask.java | 107 +++++++++----
.../runtime/tasks/StreamOperatorWrapper.java | 11 +-
.../flink/streaming/runtime/tasks/StreamTask.java | 139 +++++++++++------
.../api/operators/SourceOperatorTest.java | 24 +++
.../api/operators/source/CollectingDataOutput.java | 2 +-
...tStreamTaskChainedSourcesCheckpointingTest.java | 82 ++++++++++
.../runtime/tasks/MultipleInputStreamTaskTest.java | 42 +++++-
.../tasks/SourceOperatorStreamTaskTest.java | 5 +-
.../runtime/tasks/SourceTaskTerminationTest.java | 10 +-
.../runtime/tasks/StreamOperatorWrapperTest.java | 4 +-
.../tasks/StreamTaskMailboxTestHarness.java | 10 +-
.../streaming/runtime/tasks/StreamTaskTest.java | 15 +-
.../runtime/tasks/SynchronousCheckpointITCase.java | 4 +-
.../JobMasterStopWithSavepointITCase.java | 10 +-
.../flink/test/checkpointing/SavepointITCase.java | 166 +++++++++++++++++++++
23 files changed, 641 insertions(+), 172 deletions(-)
diff --git a/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/FlinkKinesisITCase.java b/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/FlinkKinesisITCase.java
index 90a3c8d..fa01244 100644
--- a/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/FlinkKinesisITCase.java
+++ b/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/FlinkKinesisITCase.java
@@ -32,6 +32,7 @@ import org.apache.flink.util.TestLogger;
import org.junit.Before;
import org.junit.ClassRule;
+import org.junit.Ignore;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
@@ -52,6 +53,7 @@ import static org.hamcrest.Matchers.lessThan;
import static org.junit.Assert.assertThat;
/** IT cases for using Kinesis consumer/producer based on Kinesalite. */
+@Ignore("See FLINK-23528")
public class FlinkKinesisITCase extends TestLogger {
public static final String TEST_STREAM = "test_stream";
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointType.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointType.java
index 1c5424e..b378ae1 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointType.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointType.java
@@ -62,6 +62,10 @@ public enum CheckpointType {
}
public boolean shouldAdvanceToEndOfTime() {
+ return shouldDrain();
+ }
+
+ public boolean shouldDrain() {
return getPostCheckpointAction() == PostCheckpointAction.TERMINATE;
}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java
index 9ceb987..a24ba61 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java
@@ -32,6 +32,7 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.io.InputStatus;
import org.apache.flink.core.io.SimpleVersionedSerializer;
import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.runtime.io.AvailabilityProvider;
import org.apache.flink.runtime.operators.coordination.OperatorEvent;
import org.apache.flink.runtime.operators.coordination.OperatorEventGateway;
import org.apache.flink.runtime.operators.coordination.OperatorEventHandler;
@@ -128,7 +129,18 @@ public class SourceOperator<OUT, SplitT extends SourceSplit> extends AbstractStr
*/
private TimestampsAndWatermarks<OUT> eventTimeLogic;
- private boolean dataFinished = false;
+ /** A mode to control the behaviour of the {@link #emitNext(DataOutput)} method. */
+ private OperatingMode operatingMode;
+
+ private final CompletableFuture<Void> emittedEndOfData = new CompletableFuture<>();
+ private final CompletableFuture<Void> forcedStop = new CompletableFuture<>();
+
+ private enum OperatingMode {
+ OUTPUT_NOT_INITIALIZED,
+ READING,
+ SOURCE_STOPPED,
+ DATA_FINISHED
+ }
public SourceOperator(
FunctionWithException<SourceReaderContext, SourceReader<OUT, SplitT>, Exception>
@@ -149,6 +161,7 @@ public class SourceOperator<OUT, SplitT extends SourceSplit> extends AbstractStr
this.configuration = checkNotNull(configuration);
this.localHostname = checkNotNull(localHostname);
this.emitProgressiveWatermarks = emitProgressiveWatermarks;
+ this.operatingMode = OperatingMode.OUTPUT_NOT_INITIALIZED;
}
/**
@@ -270,6 +283,17 @@ public class SourceOperator<OUT, SplitT extends SourceSplit> extends AbstractStr
super.finish();
}
+ public CompletableFuture<Void> stop() {
+ switch (operatingMode) {
+ case OUTPUT_NOT_INITIALIZED:
+ case READING:
+ this.operatingMode = OperatingMode.SOURCE_STOPPED;
+ forcedStop.complete(null);
+ break;
+ }
+ return emittedEndOfData;
+ }
+
@Override
public void close() throws Exception {
if (sourceReader != null) {
@@ -283,17 +307,29 @@ public class SourceOperator<OUT, SplitT extends SourceSplit> extends AbstractStr
// guarding an assumptions we currently make due to the fact that certain classes
// assume a constant output, this assumption does not need to stand if we emitted all
// records. In that case the output will change to FinishedDataOutput
- assert lastInvokedOutput == output || lastInvokedOutput == null || dataFinished;
-
- // short circuit the common case (every invocation except the first)
- if (currentMainOutput != null) {
- return convertToInternalStatus(sourceReader.pollNext(currentMainOutput));
+ assert lastInvokedOutput == output
+ || lastInvokedOutput == null
+ || this.operatingMode == OperatingMode.DATA_FINISHED;
+
+ switch (operatingMode) {
+ case OUTPUT_NOT_INITIALIZED:
+ // this creates a batch or streaming output based on the runtime mode
+ currentMainOutput = eventTimeLogic.createMainOutput(output);
+ lastInvokedOutput = output;
+ this.operatingMode = OperatingMode.READING;
+ return convertToInternalStatus(sourceReader.pollNext(currentMainOutput));
+ case READING:
+ // short circuit the common case (every invocation except the first)
+ return convertToInternalStatus(sourceReader.pollNext(currentMainOutput));
+ case SOURCE_STOPPED:
+ this.operatingMode = OperatingMode.DATA_FINISHED;
+ emittedEndOfData.complete(null);
+ return DataInputStatus.END_OF_DATA;
+ case DATA_FINISHED:
+ return DataInputStatus.END_OF_INPUT;
+ default:
+ throw new IllegalStateException("Unknown operating mode: " + operatingMode);
}
-
- // this creates a batch or streaming output based on the runtime mode
- currentMainOutput = eventTimeLogic.createMainOutput(output);
- lastInvokedOutput = output;
- return convertToInternalStatus(sourceReader.pollNext(currentMainOutput));
}
private DataInputStatus convertToInternalStatus(InputStatus inputStatus) {
@@ -303,12 +339,9 @@ public class SourceOperator<OUT, SplitT extends SourceSplit> extends AbstractStr
case NOTHING_AVAILABLE:
return DataInputStatus.NOTHING_AVAILABLE;
case END_OF_INPUT:
- if (dataFinished) {
- return DataInputStatus.END_OF_INPUT;
- } else {
- dataFinished = true;
- return DataInputStatus.END_OF_DATA;
- }
+ this.operatingMode = OperatingMode.DATA_FINISHED;
+ emittedEndOfData.complete(null);
+ return DataInputStatus.END_OF_DATA;
default:
throw new IllegalArgumentException("Unknown input status: " + inputStatus);
}
@@ -323,7 +356,19 @@ public class SourceOperator<OUT, SplitT extends SourceSplit> extends AbstractStr
@Override
public CompletableFuture<?> getAvailableFuture() {
- return sourceReader.isAvailable();
+ switch (operatingMode) {
+ case OUTPUT_NOT_INITIALIZED:
+ case READING:
+ CompletableFuture<Void> sourceReaderAvailable = sourceReader.isAvailable();
+ return sourceReaderAvailable == AvailabilityProvider.AVAILABLE
+ ? sourceReaderAvailable
+ : CompletableFuture.anyOf(sourceReaderAvailable, forcedStop);
+ case SOURCE_STOPPED:
+ case DATA_FINISHED:
+ return AvailabilityProvider.AVAILABLE;
+ default:
+ throw new IllegalStateException("Unknown operating mode: " + operatingMode);
+ }
}
@Override
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamSource.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamSource.java
index 51adef5..833d2a4 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamSource.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamSource.java
@@ -114,21 +114,6 @@ public class StreamSource<OUT, SRC extends SourceFunction<OUT>>
try {
userFunction.run(ctx);
-
- // if we get here, then the user function either exited after being done (finite source)
- // or the function was canceled or stopped. For the finite source case, we should emit
- // a final watermark that indicates that we reached the end of event-time, and end
- // inputs
- // of the operator chain
- if (!isCanceledOrStopped()) {
- // in theory, the subclasses of StreamSource may implement the BoundedOneInput
- // interface,
- // so we still need the following call to end the input
- synchronized (lockingObject) {
- operatorChain.setIgnoreEndOfInput(false);
- operatorChain.endInput(1);
- }
- }
} finally {
if (latencyEmitter != null) {
latencyEmitter.close();
@@ -145,6 +130,10 @@ public class StreamSource<OUT, SRC extends SourceFunction<OUT>>
super.close();
}
+ public void stop() {
+ userFunction.cancel();
+ }
+
public void cancel() {
// important: marking the source as stopped has to happen before the function is stopped.
// the flag that tracks this status is volatile, so the memory model also guarantees
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamTaskSourceInput.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamTaskSourceInput.java
index c94aca7..5498860 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamTaskSourceInput.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamTaskSourceInput.java
@@ -153,4 +153,8 @@ public class StreamTaskSourceInput<T> implements StreamTaskInput<T>, Checkpointa
public OperatorID getOperatorID() {
return operator.getOperatorID();
}
+
+ public SourceOperator<T, ?> getOperator() {
+ return operator;
+ }
}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTask.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTask.java
index 179aedb..5f8fe43 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTask.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTask.java
@@ -42,6 +42,7 @@ import org.apache.flink.streaming.runtime.metrics.MinWatermarkGauge;
import org.apache.flink.streaming.runtime.metrics.WatermarkGauge;
import org.apache.flink.streaming.runtime.partitioner.StreamPartitioner;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.concurrent.FutureUtils;
import javax.annotation.Nullable;
@@ -51,8 +52,8 @@ import java.util.HashMap;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.Future;
import java.util.function.Function;
+import java.util.stream.Collectors;
/**
* A {@link StreamTask} for executing a {@link MultipleInputStreamOperator} and supporting the
@@ -181,7 +182,7 @@ public class MultipleInputStreamTask<OUT>
}
@Override
- public Future<Boolean> triggerCheckpointAsync(
+ public CompletableFuture<Boolean> triggerCheckpointAsync(
CheckpointMetaData metadata, CheckpointOptions options) {
if (operatorChain.getSourceTaskInputs().size() == 0) {
@@ -194,6 +195,24 @@ public class MultipleInputStreamTask<OUT>
// EndOfPartitionEvent, we would not complement barriers for the
// unfinished network inputs, and the checkpoint would be triggered
// after received all the EndOfPartitionEvent.
+ if (options.getCheckpointType().shouldDrain()) {
+ CompletableFuture<Void> sourcesStopped =
+ FutureUtils.waitForAll(
+ operatorChain.getSourceTaskInputs().stream()
+ .map(s -> s.getOperator().stop())
+ .collect(Collectors.toList()));
+
+ return assertTriggeringCheckpointExceptions(
+ sourcesStopped.thenCompose(
+ ignore -> triggerSourcesCheckpointAsync(metadata, options)),
+ metadata.getCheckpointId());
+ } else {
+ return triggerSourcesCheckpointAsync(metadata, options);
+ }
+ }
+
+ private CompletableFuture<Boolean> triggerSourcesCheckpointAsync(
+ CheckpointMetaData metadata, CheckpointOptions options) {
CompletableFuture<Boolean> resultFuture = new CompletableFuture<>();
mainMailboxExecutor.execute(
() -> {
@@ -207,7 +226,7 @@ public class MultipleInputStreamTask<OUT>
pendingCheckpointCompletedFutures.put(
metadata.getCheckpointId(), resultFuture);
checkPendingCheckpointCompletedFuturesSize();
- triggerSourcesCheckpoint(
+ emitBarrierForSources(
new CheckpointBarrier(
metadata.getCheckpointId(),
metadata.getTimestamp(),
@@ -241,7 +260,7 @@ public class MultipleInputStreamTask<OUT>
}
}
- private void triggerSourcesCheckpoint(CheckpointBarrier checkpointBarrier) throws IOException {
+ private void emitBarrierForSources(CheckpointBarrier checkpointBarrier) throws IOException {
for (StreamTaskSourceInput<?> sourceInput : operatorChain.getSourceTaskInputs()) {
for (InputChannelInfo channelInfo : sourceInput.getChannelInfos()) {
checkpointBarrierHandler.processBarrier(checkpointBarrier, channelInfo);
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/OperatorChain.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/OperatorChain.java
index 24f70d5..f913a786f 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/OperatorChain.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/OperatorChain.java
@@ -133,8 +133,6 @@ public class OperatorChain<OUT, OP extends StreamOperator<OUT>>
private final OperatorEventDispatcherImpl operatorEventDispatcher;
- private boolean ignoreEndOfInput;
-
private final Closer closer = Closer.create();
public OperatorChain(
@@ -425,7 +423,7 @@ public class OperatorChain<OUT, OP extends StreamOperator<OUT>>
*/
@Override
public void endInput(int inputId) throws Exception {
- if (mainOperatorWrapper != null && !ignoreEndOfInput) {
+ if (mainOperatorWrapper != null) {
mainOperatorWrapper.endOperatorInput(inputId);
}
}
@@ -459,7 +457,7 @@ public class OperatorChain<OUT, OP extends StreamOperator<OUT>>
}
if (firstOperatorWrapper != null) {
- firstOperatorWrapper.finish(actionExecutor, ignoreEndOfInput);
+ firstOperatorWrapper.finish(actionExecutor);
}
}
@@ -801,10 +799,6 @@ public class OperatorChain<OUT, OP extends StreamOperator<OUT>>
return (tailOperatorWrapper == null) ? null : tailOperatorWrapper.getStreamOperator();
}
- public void setIgnoreEndOfInput(boolean ignoreEndOfInput) {
- this.ignoreEndOfInput = ignoreEndOfInput;
- }
-
/** Wrapper class to access the chained sources and their's outputs. */
public static class ChainedSource {
private final WatermarkGaugeExposingOutput<StreamRecord<?>> chainedSourceOutput;
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceOperatorStreamTask.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceOperatorStreamTask.java
index 0ddf972..e049d03 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceOperatorStreamTask.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceOperatorStreamTask.java
@@ -45,7 +45,6 @@ import org.apache.flink.streaming.runtime.streamstatus.StreamStatus;
import javax.annotation.Nullable;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.Future;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -61,11 +60,6 @@ public class SourceOperatorStreamTask<T> extends StreamTask<T, SourceOperator<T,
}
@Override
- protected CompletableFuture<Void> getCompletionFuture() {
- return super.getCompletionFuture();
- }
-
- @Override
public void init() throws Exception {
final SourceOperator<T, ?> sourceOperator = this.mainOperator;
// reader initialization, which cannot happen in the constructor due to the
@@ -119,26 +113,35 @@ public class SourceOperatorStreamTask<T> extends StreamTask<T, SourceOperator<T,
}
@Override
- public Future<Boolean> triggerCheckpointAsync(
+ public CompletableFuture<Boolean> triggerCheckpointAsync(
CheckpointMetaData checkpointMetaData, CheckpointOptions checkpointOptions) {
if (!isExternallyInducedSource) {
- return super.triggerCheckpointAsync(checkpointMetaData, checkpointOptions);
+ if (checkpointOptions.getCheckpointType().shouldDrain()) {
+ return triggerStopWithSavepointWithDrainAsync(
+ checkpointMetaData, checkpointOptions);
+ } else {
+ return super.triggerCheckpointAsync(checkpointMetaData, checkpointOptions);
+ }
} else {
return CompletableFuture.completedFuture(isRunning());
}
}
- @Override
- protected void advanceToEndOfEventTime() {
- output.emitWatermark(Watermark.MAX_WATERMARK);
+ private CompletableFuture<Boolean> triggerStopWithSavepointWithDrainAsync(
+ CheckpointMetaData checkpointMetaData, CheckpointOptions checkpointOptions) {
+ return assertTriggeringCheckpointExceptions(
+ mainOperator
+ .stop()
+ .thenCompose(
+ (ignore) ->
+ super.triggerCheckpointAsync(
+ checkpointMetaData, checkpointOptions)),
+ checkpointMetaData.getCheckpointId());
}
@Override
- protected void afterInvoke() throws Exception {
- if (!isCanceled()) {
- advanceToEndOfEventTime();
- }
- super.afterInvoke();
+ protected void advanceToEndOfEventTime() {
+ output.emitWatermark(Watermark.MAX_WATERMARK);
}
// --------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTask.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTask.java
index 4568e42..650441a 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTask.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTask.java
@@ -37,7 +37,7 @@ import org.apache.flink.util.FlinkException;
import org.apache.flink.util.Preconditions;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.Future;
+import java.util.concurrent.ExecutionException;
/**
* {@link StreamTask} for executing a {@link StreamSource}.
@@ -63,11 +63,30 @@ public class SourceStreamTask<
private volatile boolean externallyInducedCheckpoints;
+ private enum FinishingReason {
+ END_OF_DATA(true),
+ STOP_WITH_SAVEPOINT_DRAIN(true),
+ STOP_WITH_SAVEPOINT_NO_DRAIN(false);
+
+ private final boolean shouldCallFinish;
+
+ FinishingReason(boolean shouldCallFinish) {
+ this.shouldCallFinish = shouldCallFinish;
+ }
+
+ boolean shouldCallFinish() {
+ return this.shouldCallFinish;
+ }
+ }
+
/**
- * Indicates whether this Task was purposefully finished (by finishTask()), in this case we want
- * to ignore exceptions thrown after finishing, to ensure shutdown works smoothly.
+ * Indicates whether this Task was purposefully finished, in this case we want to ignore
+ * exceptions thrown after finishing, to ensure shutdown works smoothly.
+ *
+ * <p>Moreover we differentiate drain and no drain cases to see if we need to call finish() on
+ * the operators.
*/
- private volatile boolean wasStoppedExternally = false;
+ private volatile FinishingReason finishingReason = FinishingReason.END_OF_DATA;
public SourceStreamTask(Environment env) throws Exception {
this(env, new Object());
@@ -184,12 +203,12 @@ public class SourceStreamTask<
@Override
protected void cancelTask() {
- cancelTask(true);
+ cancelOperator(true);
}
@Override
protected void finishTask() {
- wasStoppedExternally = true;
+ this.finishingReason = FinishingReason.STOP_WITH_SAVEPOINT_NO_DRAIN;
/**
* Currently stop with savepoint relies on the EndOfPartitionEvents propagation and performs
* clean shutdown after the stop with savepoint (which can produce some records to process
@@ -197,16 +216,16 @@ public class SourceStreamTask<
* network stack in an inconsistent state. So, if we want to relay on the clean shutdown, we
* can not interrupt the source thread.
*/
- cancelTask(false);
+ cancelOperator(false);
}
- private void cancelTask(boolean interrupt) {
+ private void cancelOperator(boolean interruptThread) {
try {
if (mainOperator != null) {
mainOperator.cancel();
}
} finally {
- interruptSourceThread(interrupt);
+ interruptSourceThread(interruptThread);
}
}
@@ -236,10 +255,15 @@ public class SourceStreamTask<
// ------------------------------------------------------------------------
@Override
- public Future<Boolean> triggerCheckpointAsync(
+ public CompletableFuture<Boolean> triggerCheckpointAsync(
CheckpointMetaData checkpointMetaData, CheckpointOptions checkpointOptions) {
if (!externallyInducedCheckpoints) {
- return super.triggerCheckpointAsync(checkpointMetaData, checkpointOptions);
+ if (checkpointOptions.getCheckpointType().shouldDrain()) {
+ return triggerStopWithSavepointWithDrainAsync(
+ checkpointMetaData, checkpointOptions);
+ } else {
+ return super.triggerCheckpointAsync(checkpointMetaData, checkpointOptions);
+ }
} else {
// we do not trigger checkpoints here, we simply state whether we can trigger them
synchronized (lock) {
@@ -248,6 +272,31 @@ public class SourceStreamTask<
}
}
+ private CompletableFuture<Boolean> triggerStopWithSavepointWithDrainAsync(
+ CheckpointMetaData checkpointMetaData, CheckpointOptions checkpointOptions) {
+ mainMailboxExecutor.execute(
+ () ->
+ stopOperatorForStopWithSavepointWithDrain(
+ checkpointMetaData.getCheckpointId()),
+ "stop legacy source for stop-with-savepoint --drain");
+ return assertTriggeringCheckpointExceptions(
+ sourceThread
+ .getCompletionFuture()
+ .thenCompose(
+ ignore ->
+ super.triggerCheckpointAsync(
+ checkpointMetaData, checkpointOptions)),
+ checkpointMetaData.getCheckpointId());
+ }
+
+ private void stopOperatorForStopWithSavepointWithDrain(long checkpointId) {
+ setSynchronousSavepoint(checkpointId, true);
+ finishingReason = FinishingReason.STOP_WITH_SAVEPOINT_DRAIN;
+ if (mainOperator != null) {
+ mainOperator.stop();
+ }
+ }
+
@Override
protected void declineCheckpoint(long checkpointId) {
if (!externallyInducedCheckpoints) {
@@ -273,36 +322,36 @@ public class SourceStreamTask<
getTaskNameWithSubtaskAndId());
mainOperator.run(lock, operatorChain);
}
- if (!wasStoppedExternally && !isCanceled() && !isFailing()) {
- synchronized (lock) {
- operatorChain.setIgnoreEndOfInput(false);
- }
- CompletableFuture<Void> endOfDataProcessed = new CompletableFuture<>();
- mainMailboxExecutor.execute(
- () -> {
- endData();
- endOfDataProcessed.complete(null);
- },
- "SourceStreamTask finished processing data.");
-
- // wait until all operators are finished
- endOfDataProcessed.get();
- }
+ completeProcessing();
completionFuture.complete(null);
} catch (Throwable t) {
// Note, t can be also an InterruptedException
- completionFuture.completeExceptionally(t);
if (isCanceled()
&& ExceptionUtils.findThrowable(t, InterruptedException.class)
.isPresent()) {
completionFuture.completeExceptionally(new CancelTaskException(t));
- } else if (wasStoppedExternally) {
- // swallow all exceptions if the source was stopped externally
+ } else if (finishingReason == FinishingReason.STOP_WITH_SAVEPOINT_NO_DRAIN) {
+ // swallow all exceptions if the source was stopped without drain
completionFuture.complete(null);
} else {
completionFuture.completeExceptionally(t);
}
}
+ }
+
+ private void completeProcessing() throws InterruptedException, ExecutionException {
+ if (finishingReason.shouldCallFinish() && !isCanceled() && !isFailing()) {
+ mainMailboxExecutor
+ .submit(
+ () -> {
+ // theoretically the StreamSource can implement BoundedOneInput,
+ // so we
+ // need to call it here
+ operatorChain.endInput(1);
+ endData();
+ },
+ "SourceStreamTask finished processing data.")
+ .get();
}
}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamOperatorWrapper.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamOperatorWrapper.java
index 677dd5f..e8a24ae 100755
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamOperatorWrapper.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamOperatorWrapper.java
@@ -39,8 +39,8 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
* This class handles the finish, endInput and other related logic of a {@link StreamOperator}. It
* also automatically propagates the finish operation to the next wrapper that the {@link #next}
* points to, so we can use {@link #next} to link all operator wrappers in the operator chain and
- * finish all operators only by calling the {@link #finish(StreamTaskActionExecutor, boolean)}
- * method of the header operator wrapper.
+ * finish all operators only by calling the {@link #finish(StreamTaskActionExecutor)} method of the
+ * header operator wrapper.
*/
@Internal
public class StreamOperatorWrapper<OUT, OP extends StreamOperator<OUT>> {
@@ -120,9 +120,8 @@ public class StreamOperatorWrapper<OUT, OP extends StreamOperator<OUT>> {
* MailboxExecutor#yield()} to take the mails of closing operator and running timers and run
* them.
*/
- public void finish(StreamTaskActionExecutor actionExecutor, boolean isStoppingBySyncSavepoint)
- throws Exception {
- if (!isHead && !isStoppingBySyncSavepoint) {
+ public void finish(StreamTaskActionExecutor actionExecutor) throws Exception {
+ if (!isHead) {
// NOTE: This only do for the case where the operator is one-input operator. At present,
// any non-head operator on the operator chain is one-input operator.
actionExecutor.runThrowing(() -> endOperatorInput(1));
@@ -132,7 +131,7 @@ public class StreamOperatorWrapper<OUT, OP extends StreamOperator<OUT>> {
// propagate the close operation to the next wrapper
if (next != null) {
- next.finish(actionExecutor, isStoppingBySyncSavepoint);
+ next.finish(actionExecutor);
}
}
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 4415833..849e6fc 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
@@ -92,10 +92,12 @@ import org.apache.flink.streaming.runtime.tasks.mailbox.TaskMailboxImpl;
import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.FatalExitExceptionHandler;
import org.apache.flink.util.FlinkException;
+import org.apache.flink.util.FlinkRuntimeException;
import org.apache.flink.util.InstantiationUtil;
import org.apache.flink.util.Preconditions;
import org.apache.flink.util.SerializedValue;
import org.apache.flink.util.TernaryBoolean;
+import org.apache.flink.util.WrappingRuntimeException;
import org.apache.flink.util.concurrent.ExecutorThreadFactory;
import org.apache.flink.util.concurrent.FutureUtils;
import org.apache.flink.util.function.RunnableWithException;
@@ -261,8 +263,10 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>> extends Ab
/** TODO it might be replaced by the global IO executor on TaskManager level future. */
private final ExecutorService channelIOExecutor;
- private Long syncSavepointId = null;
- private Long activeSyncSavepointId = null;
+ private Long syncSavepointWithoutDrain = null;
+
+ private Long syncSavepointWithDrain = null;
+ private CompletableFuture<Void> syncSavepointWithDrainCompletionFuture = null;
private long latestReportCheckpointId = -1;
@@ -487,7 +491,6 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>> extends Ab
throw new IllegalStateException("We should not receive this event here.");
case END_OF_DATA:
endData();
-
return;
case END_OF_INPUT:
// Suspend the mailbox processor, it would be resumed in afterInvoke and finished
@@ -517,6 +520,7 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>> extends Ab
}
protected void endData() throws Exception {
+ advanceToEndOfEventTime();
// finish all operators in a chain effect way
operatorChain.finishOperators(actionExecutor);
this.finishedOperators = true;
@@ -528,40 +532,48 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>> extends Ab
this.endOfDataReceived = true;
}
- private void resetSynchronousSavepointId(long id, boolean succeeded) {
- if (!succeeded && activeSyncSavepointId != null && activeSyncSavepointId == id) {
- // allow to process further EndOfPartition events
- activeSyncSavepointId = null;
- operatorChain.setIgnoreEndOfInput(false);
- }
- syncSavepointId = null;
- }
-
- private void setSynchronousSavepointId(long checkpointId, boolean ignoreEndOfInput) {
+ protected void setSynchronousSavepoint(long checkpointId, boolean isDrain) {
checkState(
- syncSavepointId == null,
+ syncSavepointWithoutDrain == null
+ && (syncSavepointWithDrain == null
+ || (isDrain && syncSavepointWithDrain == checkpointId)),
"at most one stop-with-savepoint checkpoint at a time is allowed");
- syncSavepointId = checkpointId;
- activeSyncSavepointId = checkpointId;
- operatorChain.setIgnoreEndOfInput(ignoreEndOfInput);
+ if (isDrain) {
+ if (syncSavepointWithDrain == null) {
+ syncSavepointWithDrain = checkpointId;
+ syncSavepointWithDrainCompletionFuture = new CompletableFuture<>();
+ }
+ } else {
+ syncSavepointWithoutDrain = checkpointId;
+ }
}
@VisibleForTesting
OptionalLong getSynchronousSavepointId() {
- return syncSavepointId != null ? OptionalLong.of(syncSavepointId) : OptionalLong.empty();
+ if (syncSavepointWithoutDrain != null) {
+ return OptionalLong.of(syncSavepointWithoutDrain);
+ } else if (syncSavepointWithDrain != null) {
+ return OptionalLong.of(syncSavepointWithDrain);
+ } else {
+ return OptionalLong.empty();
+ }
}
- private boolean isSynchronousSavepointId(long checkpointId) {
- return syncSavepointId != null && syncSavepointId == checkpointId;
+ private boolean isCurrentSavepointWithDrain(long checkpointId) {
+ return syncSavepointWithDrain != null && syncSavepointWithDrain == checkpointId;
+ }
+
+ private boolean isCurrentSavepointWithoutDrain(long checkpointId) {
+ return syncSavepointWithoutDrain != null && syncSavepointWithoutDrain == checkpointId;
}
private void runSynchronousSavepointMailboxLoop() throws Exception {
- assert syncSavepointId != null;
+ assert syncSavepointWithoutDrain != null;
MailboxExecutor mailboxExecutor =
mailboxProcessor.getMailboxExecutor(TaskMailbox.MAX_PRIORITY);
- while (!canceled && syncSavepointId != null) {
+ while (!canceled && syncSavepointWithoutDrain != null) {
mailboxExecutor.yield();
}
}
@@ -807,22 +819,24 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>> extends Ab
LOG.debug("Finished task {}", getName());
getCompletionFuture().exceptionally(unused -> null).join();
+ List<CompletableFuture<Void>> terminationConditions = new ArrayList<>();
// If checkpoints are enabled, waits for all the records get processed by the downstream
// tasks. During this process, this task could coordinate with its downstream tasks to
// continue perform checkpoints.
- CompletableFuture<Void> allRecordsProcessedFuture;
if (endOfDataReceived && areCheckpointsWithFinishedTasksEnabled()) {
LOG.debug("Waiting for all the records processed by the downstream tasks.");
- List<CompletableFuture<Void>> partitionRecordsProcessedFutures = new ArrayList<>();
for (ResultPartitionWriter partitionWriter : getEnvironment().getAllWriters()) {
- partitionRecordsProcessedFutures.add(partitionWriter.getAllDataProcessedFuture());
+ terminationConditions.add(partitionWriter.getAllDataProcessedFuture());
}
- allRecordsProcessedFuture = FutureUtils.waitForAll(partitionRecordsProcessedFutures);
- } else {
- allRecordsProcessedFuture = CompletableFuture.completedFuture(null);
}
- allRecordsProcessedFuture.thenRun(mailboxProcessor::allActionsCompleted);
+
+ if (syncSavepointWithDrainCompletionFuture != null) {
+ terminationConditions.add(syncSavepointWithDrainCompletionFuture);
+ }
+
+ FutureUtils.waitForAll(terminationConditions)
+ .thenRun(mailboxProcessor::allActionsCompleted);
// Resumes the mailbox processor. The mailbox processor would be completed
// after all records are processed by the downstream tasks.
@@ -847,12 +861,13 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>> extends Ab
// the queued checkpoint requirements could be triggered normally.
actionExecutor.runThrowing(
() -> {
- // only set the StreamTask to not running after all operators have been closed!
+ // only set the StreamTask to not running after all operators have been
+ // finished!
// See FLINK-7430
isRunning = false;
});
- LOG.debug("Closed operators for task {}", getName());
+ LOG.debug("Finished operators for task {}", getName());
// make sure all buffered data is flushed
operatorChain.flushOutputs();
@@ -1078,7 +1093,7 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>> extends Ab
// ------------------------------------------------------------------------
@Override
- public Future<Boolean> triggerCheckpointAsync(
+ public CompletableFuture<Boolean> triggerCheckpointAsync(
CheckpointMetaData checkpointMetaData, CheckpointOptions checkpointOptions) {
CompletableFuture<Boolean> result = new CompletableFuture<>();
@@ -1186,6 +1201,28 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>> extends Ab
return true;
}
+ protected final CompletableFuture<Boolean> assertTriggeringCheckpointExceptions(
+ CompletableFuture<Boolean> triggerFuture, long checkpointId) {
+ CompletableFuture<Boolean> checkpointTriggered =
+ triggerFuture.exceptionally(
+ error -> {
+ if (error instanceof RejectedExecutionException) {
+ // This may happen if the mailbox is closed. It means that
+ // the task is shutting
+ // down, so we just ignore it.
+ LOG.debug(
+ "Triggering checkpoint {} for {} was rejected by the mailbox",
+ checkpointId,
+ getTaskNameWithSubtaskAndId());
+ } else {
+ throw new WrappingRuntimeException(error);
+ }
+ return null;
+ });
+ FutureUtils.assertNoException(checkpointTriggered);
+ return checkpointTriggered;
+ }
+
/**
* Acquires the optional {@link CheckpointBarrierHandler} associated with this stream task. The
* {@code CheckpointBarrierHandler} should exist if the task has data inputs and requires to
@@ -1205,7 +1242,7 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>> extends Ab
FlinkSecurityManager.monitorUserSystemExitForCurrentThread();
try {
if (performCheckpoint(checkpointMetaData, checkpointOptions, checkpointMetrics)) {
- if (isSynchronousSavepointId(checkpointMetaData.getCheckpointId())) {
+ if (isCurrentSavepointWithoutDrain(checkpointMetaData.getCheckpointId())) {
runSynchronousSavepointMailboxLoop();
}
}
@@ -1231,7 +1268,11 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>> extends Ab
@Override
public void abortCheckpointOnBarrier(long checkpointId, CheckpointException cause)
throws IOException {
- resetSynchronousSavepointId(checkpointId, false);
+ if (isCurrentSavepointWithoutDrain(checkpointId)) {
+ syncSavepointWithoutDrain = null;
+ } else if (isCurrentSavepointWithDrain(checkpointId)) {
+ throw new FlinkRuntimeException("Stop-with-savepoint --drain failed.");
+ }
subtaskCheckpointCoordinator.abortCheckpointOnBarrier(checkpointId, cause, operatorChain);
}
@@ -1251,17 +1292,9 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>> extends Ab
actionExecutor.runThrowing(
() -> {
if (checkpointOptions.getCheckpointType().isSynchronous()) {
- setSynchronousSavepointId(
+ setSynchronousSavepoint(
checkpointMetaData.getCheckpointId(),
- checkpointOptions.getCheckpointType().shouldIgnoreEndOfInput());
-
- if (checkpointOptions.getCheckpointType().shouldAdvanceToEndOfTime()) {
- advanceToEndOfEventTime();
- }
- } else if (activeSyncSavepointId != null
- && activeSyncSavepointId < checkpointMetaData.getCheckpointId()) {
- activeSyncSavepointId = null;
- operatorChain.setIgnoreEndOfInput(false);
+ checkpointOptions.getCheckpointType().shouldDrain());
}
subtaskCheckpointCoordinator.checkpointState(
@@ -1322,7 +1355,11 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>> extends Ab
notifyCheckpointComplete(latestCompletedCheckpointId);
}
- resetSynchronousSavepointId(checkpointId, false);
+ if (isCurrentSavepointWithoutDrain(checkpointId)) {
+ syncSavepointWithoutDrain = null;
+ } else if (isCurrentSavepointWithDrain(checkpointId)) {
+ throw new FlinkRuntimeException("Stop-with-savepoint --drain failed.");
+ }
subtaskCheckpointCoordinator.notifyCheckpointAborted(
checkpointId, operatorChain, this::isRunning);
},
@@ -1357,10 +1394,14 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>> extends Ab
subtaskCheckpointCoordinator.notifyCheckpointComplete(
checkpointId, operatorChain, this::isRunning);
- if (isRunning && isSynchronousSavepointId(checkpointId)) {
- finishTask();
- // Reset to "notify" the internal synchronous savepoint mailbox loop.
- resetSynchronousSavepointId(checkpointId, true);
+ if (isRunning) {
+ if (isCurrentSavepointWithoutDrain(checkpointId)) {
+ finishTask();
+ // Reset to "notify" the internal synchronous savepoint mailbox loop.
+ syncSavepointWithoutDrain = null;
+ } else if (isCurrentSavepointWithDrain(checkpointId)) {
+ syncSavepointWithDrainCompletionFuture.complete(null);
+ }
}
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/SourceOperatorTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/SourceOperatorTest.java
index a518c33..37dccda 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/SourceOperatorTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/SourceOperatorTest.java
@@ -42,7 +42,9 @@ import org.apache.flink.runtime.state.StateInitializationContextImpl;
import org.apache.flink.runtime.state.StateSnapshotContextSynchronousImpl;
import org.apache.flink.runtime.state.TestTaskStateManager;
import org.apache.flink.runtime.state.memory.MemoryStateBackend;
+import org.apache.flink.streaming.api.operators.source.CollectingDataOutput;
import org.apache.flink.streaming.api.operators.source.TestingSourceOperator;
+import org.apache.flink.streaming.runtime.io.DataInputStatus;
import org.apache.flink.streaming.runtime.tasks.SourceOperatorStreamTask;
import org.apache.flink.streaming.runtime.tasks.StreamMockEnvironment;
import org.apache.flink.streaming.util.MockOutput;
@@ -57,8 +59,10 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
+import java.util.concurrent.CompletableFuture;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
@@ -128,6 +132,26 @@ public class SourceOperatorTest {
}
@Test
+ public void testStop() throws Exception {
+ // Initialize the operator.
+ operator.initializeState(getStateContext());
+ // Open the operator.
+ operator.open();
+ // The source reader should have been assigned a split.
+ assertEquals(Collections.singletonList(MOCK_SPLIT), mockSourceReader.getAssignedSplits());
+
+ CollectingDataOutput<Integer> dataOutput = new CollectingDataOutput<>();
+ assertEquals(DataInputStatus.NOTHING_AVAILABLE, operator.emitNext(dataOutput));
+ assertFalse(operator.isAvailable());
+
+ CompletableFuture<Void> sourceStopped = operator.stop();
+ assertTrue(operator.isAvailable());
+ assertFalse(sourceStopped.isDone());
+ assertEquals(DataInputStatus.END_OF_DATA, operator.emitNext(dataOutput));
+ assertTrue(sourceStopped.isDone());
+ }
+
+ @Test
public void testHandleAddSplitsEvent() throws Exception {
operator.initializeState(getStateContext());
operator.open();
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/source/CollectingDataOutput.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/source/CollectingDataOutput.java
index 939645a..a373983 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/source/CollectingDataOutput.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/source/CollectingDataOutput.java
@@ -31,7 +31,7 @@ import java.util.List;
* A test utility implementation of {@link PushingAsyncDataInput.DataOutput} that collects all
* events.
*/
-final class CollectingDataOutput<E> implements PushingAsyncDataInput.DataOutput<E> {
+public final class CollectingDataOutput<E> implements PushingAsyncDataInput.DataOutput<E> {
final List<Object> events = new ArrayList<>();
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTaskChainedSourcesCheckpointingTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTaskChainedSourcesCheckpointingTest.java
index 11fc176..0ddacce 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTaskChainedSourcesCheckpointingTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTaskChainedSourcesCheckpointingTest.java
@@ -238,6 +238,78 @@ public class MultipleInputStreamTaskChainedSourcesCheckpointingTest {
}
}
+ /**
+ * In this scenario:
+ *
+ * <ul>
+ * <li>Network inputs are processed until CheckpointBarriers for synchronous savepoint are
+ * processed.
+ * <li>RPC for stop-with-savepoint comes for sources
+ * <li>Sources keep being invoked until they return END_OF_DATA
+ * <li>Synchronous savepoint is triggered
+ * </ul>
+ */
+ @Test
+ public void testStopWithSavepointDrainWaitsForSourcesFinish() throws Exception {
+ try (StreamTaskMailboxTestHarness<String> testHarness =
+ new StreamTaskMailboxTestHarnessBuilder<>(
+ MultipleInputStreamTask::new, BasicTypeInfo.STRING_TYPE_INFO)
+ .modifyExecutionConfig(applyObjectReuse(objectReuse))
+ .modifyStreamConfig(config -> config.setCheckpointingEnabled(true))
+ .addInput(BasicTypeInfo.STRING_TYPE_INFO)
+ .addSourceInput(
+ new SourceOperatorFactory<>(
+ new MockSource(Boundedness.CONTINUOUS_UNBOUNDED, 1),
+ WatermarkStrategy.noWatermarks()),
+ BasicTypeInfo.INT_TYPE_INFO)
+ .addSourceInput(
+ new SourceOperatorFactory<>(
+ new MockSource(Boundedness.CONTINUOUS_UNBOUNDED, 1),
+ WatermarkStrategy.noWatermarks()),
+ BasicTypeInfo.INT_TYPE_INFO)
+ .addInput(BasicTypeInfo.DOUBLE_TYPE_INFO)
+ .setupOutputForSingletonOperatorChain(
+ new MapToStringMultipleInputOperatorFactory(4, true))
+ .build(); ) {
+ testHarness.setAutoProcess(false);
+ ArrayDeque<Object> expectedOutput = new ArrayDeque<>();
+ CheckpointBarrier barrier = createStopWithSavepointDrainBarrier();
+
+ testHarness.processElement(new StreamRecord<>("44", TimestampAssigner.NO_TIMESTAMP), 0);
+ testHarness.processEvent(EndOfData.INSTANCE, 0);
+ testHarness.processEvent(barrier, 0);
+ testHarness.processElement(new StreamRecord<>(47d, TimestampAssigner.NO_TIMESTAMP), 1);
+ testHarness.processEvent(EndOfData.INSTANCE, 1);
+ testHarness.processEvent(barrier, 1);
+
+ addSourceRecords(testHarness, 1, Boundedness.CONTINUOUS_UNBOUNDED, 1, 2);
+ addSourceRecords(testHarness, 2, Boundedness.CONTINUOUS_UNBOUNDED, 3, 4);
+
+ testHarness.processAll();
+
+ Future<Boolean> checkpointFuture =
+ testHarness
+ .getStreamTask()
+ .triggerCheckpointAsync(metaData, barrier.getCheckpointOptions());
+ processSingleStepUntil(testHarness, checkpointFuture::isDone);
+
+ expectedOutput.add(new StreamRecord<>("3", TimestampAssigner.NO_TIMESTAMP));
+ expectedOutput.add(new StreamRecord<>("47.0", TimestampAssigner.NO_TIMESTAMP));
+ expectedOutput.add(new StreamRecord<>("44", TimestampAssigner.NO_TIMESTAMP));
+ expectedOutput.add(new StreamRecord<>("1", TimestampAssigner.NO_TIMESTAMP));
+ expectedOutput.add(new StreamRecord<>("4", TimestampAssigner.NO_TIMESTAMP));
+ expectedOutput.add(new StreamRecord<>("2", TimestampAssigner.NO_TIMESTAMP));
+
+ ArrayList<Object> actualOutput = new ArrayList<>(testHarness.getOutput());
+ assertThat(
+ actualOutput.subList(0, expectedOutput.size()),
+ containsInAnyOrder(expectedOutput.toArray()));
+ assertThat(
+ actualOutput.subList(actualOutput.size() - 2, actualOutput.size()),
+ contains(new StreamRecord<>("FINISH"), barrier));
+ }
+ }
+
@Test
public void testOnlyOneSource() throws Exception {
try (StreamTaskMailboxTestHarness<String> testHarness =
@@ -445,6 +517,16 @@ public class MultipleInputStreamTaskChainedSourcesCheckpointingTest {
addBarriers(testHarness, checkpointBarrier);
}
+ private CheckpointBarrier createStopWithSavepointDrainBarrier() {
+ CheckpointOptions checkpointOptions =
+ CheckpointOptions.alignedNoTimeout(
+ CheckpointType.SAVEPOINT_TERMINATE,
+ CheckpointStorageLocationReference.getDefault());
+
+ return new CheckpointBarrier(
+ metaData.getCheckpointId(), metaData.getTimestamp(), checkpointOptions);
+ }
+
private CheckpointBarrier createBarrier(StreamTaskMailboxTestHarness<String> testHarness) {
StreamConfig config = testHarness.getStreamTask().getConfiguration();
CheckpointOptions checkpointOptions =
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTaskTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTaskTest.java
index 3a169b7..5f61120 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTaskTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTaskTest.java
@@ -114,10 +114,10 @@ import static org.apache.flink.streaming.runtime.tasks.StreamTaskFinalCheckpoint
import static org.apache.flink.streaming.runtime.tasks.StreamTaskFinalCheckpointsTest.triggerCheckpoint;
import static org.apache.flink.util.Preconditions.checkArgument;
import static org.apache.flink.util.Preconditions.checkState;
+import static org.hamcrest.CoreMatchers.not;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.contains;
import static org.hamcrest.Matchers.containsInAnyOrder;
-import static org.hamcrest.Matchers.not;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
@@ -1001,13 +1001,22 @@ public class MultipleInputStreamTaskTest {
private static final long serialVersionUID = 1L;
private final int numberOfInputs;
+ private final boolean emitOnFinish;
private boolean openCalled;
private boolean closeCalled;
public MapToStringMultipleInputOperator(
StreamOperatorParameters<String> parameters, int numberOfInputs) {
+ this(parameters, numberOfInputs, false);
+ }
+
+ public MapToStringMultipleInputOperator(
+ StreamOperatorParameters<String> parameters,
+ int numberOfInputs,
+ boolean emitOnFinish) {
super(parameters, numberOfInputs);
this.numberOfInputs = numberOfInputs;
+ this.emitOnFinish = emitOnFinish;
}
@Override
@@ -1020,6 +1029,13 @@ public class MultipleInputStreamTaskTest {
}
@Override
+ public void finish() throws Exception {
+ if (emitOnFinish) {
+ output.collect(new StreamRecord<>("FINISH"));
+ }
+ }
+
+ @Override
public void close() throws Exception {
super.close();
if (!openCalled) {
@@ -1069,15 +1085,22 @@ public class MultipleInputStreamTaskTest {
protected static class MapToStringMultipleInputOperatorFactory
extends AbstractStreamOperatorFactory<String> {
private final int numberOfInputs;
+ private final boolean emitOnFinish;
public MapToStringMultipleInputOperatorFactory(int numberOfInputs) {
+ this(numberOfInputs, false);
+ }
+
+ public MapToStringMultipleInputOperatorFactory(int numberOfInputs, boolean emitOnFinish) {
this.numberOfInputs = numberOfInputs;
+ this.emitOnFinish = emitOnFinish;
}
@Override
public <T extends StreamOperator<String>> T createStreamOperator(
StreamOperatorParameters<String> parameters) {
- return (T) new MapToStringMultipleInputOperator(parameters, numberOfInputs);
+ return (T)
+ new MapToStringMultipleInputOperator(parameters, numberOfInputs, emitOnFinish);
}
@Override
@@ -1124,10 +1147,23 @@ public class MultipleInputStreamTaskTest {
static void addSourceRecords(
StreamTaskMailboxTestHarness<String> testHarness, int sourceId, int... records)
throws Exception {
+ addSourceRecords(testHarness, sourceId, Boundedness.BOUNDED, records);
+ }
+
+ static void addSourceRecords(
+ StreamTaskMailboxTestHarness<String> testHarness,
+ int sourceId,
+ Boundedness boundedness,
+ int... records)
+ throws Exception {
OperatorID sourceOperatorID = getSourceOperatorID(testHarness, sourceId);
// Prepare the source split and assign it to the source reader.
- MockSourceSplit split = new MockSourceSplit(0, 0, records.length);
+ MockSourceSplit split =
+ new MockSourceSplit(
+ 0,
+ 0,
+ boundedness == Boundedness.BOUNDED ? records.length : Integer.MAX_VALUE);
for (int record : records) {
split.addRecord(record);
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SourceOperatorStreamTaskTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SourceOperatorStreamTaskTest.java
index 9db6f8b..2dfdcf3 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SourceOperatorStreamTaskTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SourceOperatorStreamTaskTest.java
@@ -394,10 +394,11 @@ public class SourceOperatorStreamTaskTest extends SourceStreamTaskTestBase {
@Override
public InputStatus pollNext(ReaderOutput<Integer> output) throws Exception {
- numEmittedEvents++;
- if (numEmittedEvents == numEventsBeforeCheckpoint) {
+ if (numEmittedEvents == numEventsBeforeCheckpoint - 1) {
+ numEmittedEvents++;
return InputStatus.NOTHING_AVAILABLE;
} else if (numEmittedEvents < totalNumEvents) {
+ numEmittedEvents++;
return InputStatus.MORE_AVAILABLE;
} else {
return InputStatus.END_OF_INPUT;
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SourceTaskTerminationTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SourceTaskTerminationTest.java
index 9aa0fcc..4baafbd0 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SourceTaskTerminationTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SourceTaskTerminationTest.java
@@ -119,7 +119,9 @@ public class SourceTaskTerminationTest extends TestLogger {
assertTrue(srcTask.getSynchronousSavepointId().isPresent());
srcTask.notifyCheckpointCompleteAsync(syncSavepointId).get();
- assertFalse(srcTask.getSynchronousSavepointId().isPresent());
+ if (!shouldTerminate) {
+ assertFalse(srcTask.getSynchronousSavepointId().isPresent());
+ }
executionThread.join();
}
@@ -192,8 +194,10 @@ public class SourceTaskTerminationTest extends TestLogger {
while (isRunning) {
runLoopStart.await();
- ctx.emitWatermark(new Watermark(element));
- ctx.collect(element++);
+ if (isRunning) {
+ ctx.emitWatermark(new Watermark(element));
+ ctx.collect(element++);
+ }
runLoopEnd.trigger();
}
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamOperatorWrapperTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamOperatorWrapperTest.java
index 3c5a1e6..7a02f60 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamOperatorWrapperTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamOperatorWrapperTest.java
@@ -133,7 +133,7 @@ public class StreamOperatorWrapperTest extends TestLogger {
@Test
public void testFinish() throws Exception {
output.clear();
- operatorWrappers.get(0).finish(containingTask.getActionExecutor(), false);
+ operatorWrappers.get(0).finish(containingTask.getActionExecutor());
List<Object> expected = new ArrayList<>();
for (int i = 0; i < operatorWrappers.size(); i++) {
@@ -172,7 +172,7 @@ public class StreamOperatorWrapperTest extends TestLogger {
true);
try {
- operatorWrapper.finish(containingTask.getActionExecutor(), false);
+ operatorWrapper.finish(containingTask.getActionExecutor());
fail("should throw an exception");
} catch (Throwable t) {
Optional<Throwable> optional =
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskMailboxTestHarness.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskMailboxTestHarness.java
index 55751b9..a86542b 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskMailboxTestHarness.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskMailboxTestHarness.java
@@ -133,14 +133,18 @@ public class StreamTaskMailboxTestHarness<OUT> implements AutoCloseable {
}
public void endInput() {
+ endInput(true);
+ }
+
+ public void endInput(boolean allDataProcessed) {
for (int i = 0; i < inputGates.length; i++) {
- endInput(i);
+ endInput(i, allDataProcessed);
}
}
- public void endInput(int inputIndex) {
+ public void endInput(int inputIndex, boolean emitEndOfData) {
if (!inputGateEnded[inputIndex]) {
- inputGates[inputIndex].endInput();
+ inputGates[inputIndex].endInput(emitEndOfData);
inputGateEnded[inputIndex] = true;
}
}
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 2fb69ae..0d8558e 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
@@ -126,6 +126,7 @@ import org.apache.flink.streaming.util.TestSequentialReadingStreamOperator;
import org.apache.flink.util.CloseableIterable;
import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.FatalExitExceptionHandler;
+import org.apache.flink.util.FlinkRuntimeException;
import org.apache.flink.util.Preconditions;
import org.apache.flink.util.TestLogger;
import org.apache.flink.util.clock.SystemClock;
@@ -137,7 +138,9 @@ import org.apache.flink.util.function.SupplierWithException;
import org.hamcrest.Matchers;
import org.junit.Assert;
+import org.junit.Rule;
import org.junit.Test;
+import org.junit.rules.ExpectedException;
import org.mockito.ArgumentCaptor;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
@@ -208,6 +211,8 @@ public class StreamTaskTest extends TestLogger {
private static OneShotLatch syncLatch;
+ @Rule public ExpectedException thrown = ExpectedException.none();
+
@Test
public void testCancellationWaitsForActiveTimers() throws Exception {
StreamTaskWithBlockingTimer.reset();
@@ -256,11 +261,13 @@ public class StreamTaskTest extends TestLogger {
new CheckpointException(
UNKNOWN_TASK_CHECKPOINT_NOTIFICATION_FAILURE)),
CheckpointType.SAVEPOINT_SUSPEND,
- true);
+ false);
}
@Test
public void testSavepointTerminateAborted() throws Exception {
+ thrown.expect(FlinkRuntimeException.class);
+ thrown.expectMessage("Stop-with-savepoint --drain failed.");
testSyncSavepointWithEndInput(
(task, id) ->
task.abortCheckpointOnBarrier(
@@ -277,11 +284,13 @@ public class StreamTaskTest extends TestLogger {
(streamTask, abortCheckpointId) ->
streamTask.notifyCheckpointAbortAsync(abortCheckpointId, 0),
CheckpointType.SAVEPOINT_SUSPEND,
- true);
+ false);
}
@Test
public void testSavepointTerminateAbortedAsync() throws Exception {
+ thrown.expect(FlinkRuntimeException.class);
+ thrown.expectMessage("Stop-with-savepoint --drain failed.");
testSyncSavepointWithEndInput(
(streamTask, abortCheckpointId) ->
streamTask.notifyCheckpointAbortAsync(abortCheckpointId, 0),
@@ -333,7 +342,7 @@ public class StreamTaskTest extends TestLogger {
() -> {
try {
savepointTriggeredLatch.await();
- harness.endInput();
+ harness.endInput(expectEndInput);
inputEndedLatch.countDown();
} catch (InterruptedException e) {
fail(e.getMessage());
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SynchronousCheckpointITCase.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SynchronousCheckpointITCase.java
index 912474c..658be98 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SynchronousCheckpointITCase.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SynchronousCheckpointITCase.java
@@ -153,11 +153,11 @@ public class SynchronousCheckpointITCase {
}
@Override
- public Future<Boolean> triggerCheckpointAsync(
+ public CompletableFuture<Boolean> triggerCheckpointAsync(
CheckpointMetaData checkpointMetaData, CheckpointOptions checkpointOptions) {
try {
eventQueue.put(Event.PRE_TRIGGER_CHECKPOINT);
- Future<Boolean> result =
+ CompletableFuture<Boolean> result =
super.triggerCheckpointAsync(checkpointMetaData, checkpointOptions);
eventQueue.put(Event.POST_TRIGGER_CHECKPOINT);
return result;
diff --git a/flink-tests/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterStopWithSavepointITCase.java b/flink-tests/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterStopWithSavepointITCase.java
index 23278c8..39bf318 100644
--- a/flink-tests/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterStopWithSavepointITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterStopWithSavepointITCase.java
@@ -105,12 +105,6 @@ public class JobMasterStopWithSavepointITCase extends AbstractTestBase {
stopWithSavepointNormalExecutionHelper(false);
}
- @Test
- public void terminateWithSavepointWithoutComplicationsShouldSucceedAndLeadJobToFinished()
- throws Exception {
- stopWithSavepointNormalExecutionHelper(true);
- }
-
private void stopWithSavepointNormalExecutionHelper(final boolean terminate) throws Exception {
setUpJobGraph(NoOpBlockingStreamTask.class, RestartStrategies.noRestart());
@@ -347,7 +341,7 @@ public class JobMasterStopWithSavepointITCase extends AbstractTestBase {
}
@Override
- public Future<Boolean> triggerCheckpointAsync(
+ public CompletableFuture<Boolean> triggerCheckpointAsync(
CheckpointMetaData checkpointMetaData, CheckpointOptions checkpointOptions) {
final long checkpointId = checkpointMetaData.getCheckpointId();
final CheckpointType checkpointType = checkpointOptions.getCheckpointType();
@@ -443,7 +437,7 @@ public class JobMasterStopWithSavepointITCase extends AbstractTestBase {
}
@Override
- public Future<Boolean> triggerCheckpointAsync(
+ public CompletableFuture<Boolean> triggerCheckpointAsync(
final CheckpointMetaData checkpointMetaData,
final CheckpointOptions checkpointOptions) {
final long taskIndex = getEnvironment().getTaskInfo().getIndexOfThisSubtask();
diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/SavepointITCase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/SavepointITCase.java
index e602ba2..2e5c960 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/SavepointITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/SavepointITCase.java
@@ -34,19 +34,30 @@ import org.apache.flink.client.program.ClusterClient;
import org.apache.flink.configuration.CheckpointingOptions;
import org.apache.flink.configuration.ClusterOptions;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.HighAvailabilityOptions;
import org.apache.flink.configuration.MemorySize;
import org.apache.flink.configuration.StateBackendOptions;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.configuration.UnmodifiableConfiguration;
import org.apache.flink.core.testutils.OneShotLatch;
import org.apache.flink.runtime.checkpoint.CheckpointException;
+import org.apache.flink.runtime.checkpoint.CheckpointRecoveryFactory;
+import org.apache.flink.runtime.checkpoint.CheckpointsCleaner;
+import org.apache.flink.runtime.checkpoint.CompletedCheckpoint;
+import org.apache.flink.runtime.checkpoint.PerJobCheckpointRecoveryFactory;
+import org.apache.flink.runtime.checkpoint.StandaloneCheckpointIDCounter;
+import org.apache.flink.runtime.checkpoint.StandaloneCompletedCheckpointStore;
import org.apache.flink.runtime.client.JobExecutionException;
import org.apache.flink.runtime.execution.ExecutionState;
+import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
+import org.apache.flink.runtime.highavailability.HighAvailabilityServicesFactory;
+import org.apache.flink.runtime.highavailability.nonha.embedded.EmbeddedHaServices;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobgraph.JobGraphTestUtils;
import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
import org.apache.flink.runtime.messages.FlinkJobNotFoundException;
+import org.apache.flink.runtime.operators.testutils.ExpectedTestException;
import org.apache.flink.runtime.rest.RestClient;
import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
import org.apache.flink.runtime.rest.messages.JobMessageParameters;
@@ -65,6 +76,7 @@ import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.IterativeStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.DiscardingSink;
+import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.graph.StreamGraph;
@@ -81,6 +93,7 @@ import org.apache.flink.util.FlinkException;
import org.apache.flink.util.TestLogger;
import org.apache.flink.util.concurrent.FutureUtils;
+import org.hamcrest.CoreMatchers;
import org.hamcrest.Description;
import org.hamcrest.Matcher;
import org.hamcrest.TypeSafeDiagnosingMatcher;
@@ -112,6 +125,7 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;
import java.util.stream.Collectors;
@@ -207,6 +221,64 @@ public class SavepointITCase extends TestLogger {
}
}
+ @Test
+ public void testStopWithSavepointWithDrainCallsFinishBeforeSnapshotState() throws Exception {
+ int sinkParallelism = 5;
+ MiniClusterWithClientResource cluster =
+ new MiniClusterWithClientResource(
+ new MiniClusterResourceConfiguration.Builder()
+ .setNumberSlotsPerTaskManager(sinkParallelism + 1)
+ .build());
+
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.getConfig().setRestartStrategy(RestartStrategies.noRestart());
+ env.addSource(new InfiniteTestSource())
+ .name("Infinite Source")
+ .addSink(new FinishingSink<>())
+ // different parallelism to break chaining and add some concurrent tasks
+ .setParallelism(sinkParallelism);
+
+ final JobGraph jobGraph = env.getStreamGraph().getJobGraph();
+
+ cluster.before();
+ try {
+ ClusterClient<?> client = cluster.getClusterClient();
+ client.submitJob(jobGraph).get();
+ waitUntilAllTasksAreRunning(cluster.getRestAddres(), jobGraph.getJobID());
+
+ client.stopWithSavepoint(jobGraph.getJobID(), true, savepointDir.getAbsolutePath())
+ .get();
+ // there should be no exceptions and the finish should've been called in the
+ // FinishingSink
+ } finally {
+ cluster.after();
+ }
+ }
+
+ private static class FinishingSink<T> implements SinkFunction<T>, CheckpointedFunction {
+ private boolean finishCalled;
+
+ @Override
+ public void invoke(T value) throws Exception {
+ // ignore
+ }
+
+ @Override
+ public void finish() throws Exception {
+ this.finishCalled = true;
+ }
+
+ @Override
+ public void snapshotState(FunctionSnapshotContext context) throws Exception {
+ if (!finishCalled) {
+ fail("Finish is expected to be called before taking the savepoint with drain");
+ }
+ }
+
+ @Override
+ public void initializeState(FunctionInitializationContext context) throws Exception {}
+ }
+
/**
* Triggers a savepoint for a job that uses the FsStateBackend. We expect that all checkpoint
* files are written to a new savepoint directory.
@@ -638,6 +710,100 @@ public class SavepointITCase extends TestLogger {
assertAfterSnapshotCreationFailure());
}
+ @Test
+ public void testStopWithSavepointWithDrainGlobalFailoverIfSavepointAborted() throws Exception {
+ Configuration configuration = new Configuration();
+ configuration.setString(
+ HighAvailabilityOptions.HA_MODE, FailingSyncSavepointHAFactory.class.getName());
+ MiniClusterWithClientResource cluster =
+ new MiniClusterWithClientResource(
+ new MiniClusterResourceConfiguration.Builder()
+ .setConfiguration(configuration)
+ .build());
+
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setParallelism(1);
+ env.getConfig().setRestartStrategy(RestartStrategies.noRestart());
+ env.addSource(new InfiniteTestSource())
+ .name("Infinite Source")
+ .addSink(new DiscardingSink<>());
+
+ final JobGraph jobGraph = env.getStreamGraph().getJobGraph();
+
+ cluster.before();
+ try {
+ ClusterClient<?> client = cluster.getClusterClient();
+ client.submitJob(jobGraph).get();
+ waitUntilAllTasksAreRunning(cluster.getRestAddres(), jobGraph.getJobID());
+
+ try {
+ client.stopWithSavepoint(jobGraph.getJobID(), true, savepointDir.getAbsolutePath())
+ .get();
+ fail("The future should fail exceptionally.");
+ } catch (ExecutionException e) {
+ assertThat(
+ e.getMessage(),
+ CoreMatchers.startsWith(
+ "org.apache.flink.util.FlinkException: Inconsistent execution state"
+ + " after stopping with savepoint. At least one execution"
+ + " is still in one of the following states: FAILED. "
+ + "A global fail-over is triggered to recover the job"));
+ }
+ } finally {
+ cluster.after();
+ }
+ }
+
+ private static class FailingSyncSavepointCompletedCheckpointStore
+ extends StandaloneCompletedCheckpointStore {
+ FailingSyncSavepointCompletedCheckpointStore() {
+ super(1);
+ }
+
+ @Override
+ public void addCheckpoint(
+ CompletedCheckpoint checkpoint,
+ CheckpointsCleaner checkpointsCleaner,
+ Runnable postCleanup)
+ throws Exception {
+ if (checkpoint.getProperties().isSynchronous()) {
+ throw new ExpectedTestException();
+ } else {
+ super.addCheckpoint(checkpoint, checkpointsCleaner, postCleanup);
+ }
+ }
+ }
+
+ private static class TestingHaServices extends EmbeddedHaServices {
+ private final CheckpointRecoveryFactory checkpointRecoveryFactory;
+
+ TestingHaServices(CheckpointRecoveryFactory checkpointRecoveryFactory, Executor executor) {
+ super(executor);
+ this.checkpointRecoveryFactory = checkpointRecoveryFactory;
+ }
+
+ @Override
+ public CheckpointRecoveryFactory getCheckpointRecoveryFactory() {
+ return checkpointRecoveryFactory;
+ }
+ }
+
+ /**
+ * A factory for HA services used to inject {@link
+ * FailingSyncSavepointCompletedCheckpointStore}.
+ */
+ public static class FailingSyncSavepointHAFactory implements HighAvailabilityServicesFactory {
+ @Override
+ public HighAvailabilityServices createHAServices(
+ Configuration configuration, Executor executor) throws Exception {
+ return new TestingHaServices(
+ PerJobCheckpointRecoveryFactory.useSameServicesForAllJobs(
+ new FailingSyncSavepointCompletedCheckpointStore(),
+ new StandaloneCheckpointIDCounter()),
+ executor);
+ }
+ }
+
private static BiConsumer<JobID, ExecutionException> assertAfterSnapshotCreationFailure() {
return (jobId, actualException) -> {
if (ClusterOptions.isAdaptiveSchedulerEnabled(new Configuration())) {