rkhachatryan commented on code in PR #28659:
URL: https://github.com/apache/flink/pull/28659#discussion_r3598709891
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java:
##########
@@ -881,58 +918,120 @@ private CompletableFuture<Void> restoreStateAndGates(
INITIALIZE_STATE_DURATION, initializeStateEndTs -
readOutputDataTs);
IndexedInputGate[] inputGates = getEnvironment().getAllInputGates();
- boolean checkpointingDuringRecoveryEnabled =
-
CheckpointingOptions.isCheckpointingDuringRecoveryEnabled(getJobConfiguration());
+ recoveryCompletionFuture =
+
CheckpointingOptions.isCheckpointingDuringRecoveryEnabled(getJobConfiguration())
+ ? recoverChannelsWithCheckpointing(reader, inputGates)
+ : recoverChannelsWithoutCheckpointing(reader,
inputGates);
+
+ recoveryCompletionFuture.whenComplete((ign, throwable) ->
mailboxProcessor.suspend());
+ return recoveryCompletionFuture;
+ }
+
+ private CompletableFuture<Void> recoverChannelsWithCheckpointing(
+ SequentialChannelStateReader reader, IndexedInputGate[]
inputGates) {
+ if (inputGates.length == 0) {
+ // No input channels to recover (e.g. a source task). Complete
synchronously, exactly
+ // like recoverChannelsWithoutCheckpointing, so that the
recovery-completion suspend()
+ // in restoreStateAndGates is enqueued before the restore mailbox
loop runs the default
+ // action. The asynchronous chain below relies on mailbox mails
that only run while the
+ // restore loop is pumping; with no channel work to do it would
merely defer completion,
+ // letting a fast source finish and suspend the restore loop
before recovery completes
+ // -- tripping "Mailbox loop interrupted before recovery was
finished" in
+ // restoreInternal.
+ return FutureUtils.completedVoidFuture();
Review Comment:
NIT: drop the comment?
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java:
##########
@@ -881,58 +918,120 @@ private CompletableFuture<Void> restoreStateAndGates(
INITIALIZE_STATE_DURATION, initializeStateEndTs -
readOutputDataTs);
IndexedInputGate[] inputGates = getEnvironment().getAllInputGates();
- boolean checkpointingDuringRecoveryEnabled =
-
CheckpointingOptions.isCheckpointingDuringRecoveryEnabled(getJobConfiguration());
+ recoveryCompletionFuture =
+
CheckpointingOptions.isCheckpointingDuringRecoveryEnabled(getJobConfiguration())
+ ? recoverChannelsWithCheckpointing(reader, inputGates)
+ : recoverChannelsWithoutCheckpointing(reader,
inputGates);
+
+ recoveryCompletionFuture.whenComplete((ign, throwable) ->
mailboxProcessor.suspend());
+ return recoveryCompletionFuture;
+ }
+
+ private CompletableFuture<Void> recoverChannelsWithCheckpointing(
+ SequentialChannelStateReader reader, IndexedInputGate[]
inputGates) {
+ if (inputGates.length == 0) {
+ // No input channels to recover (e.g. a source task). Complete
synchronously, exactly
+ // like recoverChannelsWithoutCheckpointing, so that the
recovery-completion suspend()
+ // in restoreStateAndGates is enqueued before the restore mailbox
loop runs the default
+ // action. The asynchronous chain below relies on mailbox mails
that only run while the
+ // restore loop is pumping; with no channel work to do it would
merely defer completion,
+ // letting a fast source finish and suspend the restore loop
before recovery completes
+ // -- tripping "Mailbox loop interrupted before recovery was
finished" in
+ // restoreInternal.
+ return FutureUtils.completedVoidFuture();
+ }
+ // FLINK-38544 transitional: removed when the spilling backend lands.
In-memory backend:
+ // readInputChannelState first filters the recovered state completely
(the filter output
+ // accumulates in the recovered channels' in-memory queues), then
conversion pushes it into
+ // the physical channels through the push interface
(requestPartitions(true)). The final
+ // form instead fetches the filter output to disk and drains it
incrementally from here.
+ return CompletableFuture.runAsync(
+ () -> readInputChannelState(reader, inputGates),
channelIOExecutor)
+ .thenCompose(ign -> requestPartitions(inputGates, true))
+ .thenCompose(
+ ign ->
+ completeAll(
+ Arrays.stream(inputGates)
+
.map(InputGate::getStateConsumedFuture)
+
.collect(Collectors.toList())));
+ }
+
+ private CompletableFuture<Void> recoverChannelsWithoutCheckpointing(
+ SequentialChannelStateReader reader, IndexedInputGate[]
inputGates) {
+ // Feed recovered channel state on the IO thread. This is
intentionally NOT part of the
+ // completion gate below: a gate's stateConsumedFuture only completes
once the consumer (the
+ // default action, running in the restore mailbox loop) has drained
the end-of-state
+ // sentinel that readInputChannelState pushes, so gating on
stateConsumedFuture already
+ // implies the read has finished. Including the read future in the
gate would defer
+ // suspend() past the restore loop's first default action, letting
input be processed before
+ // recovery completes -- causing record loss and "Mailbox loop
interrupted before recovery
+ // was finished". Read failures are surfaced via asyncExceptionHandler
inside
+ // readInputChannelState.
+ channelIOExecutor.execute(() -> readInputChannelState(reader,
inputGates));
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+ for (InputGate inputGate : inputGates) {
+ CompletableFuture<Void> stateConsumed =
inputGate.getStateConsumedFuture();
+ futures.add(stateConsumed);
+ // Convert and request partitions for each gate as soon as ITS OWN
recovered state is
+ // drained, independent of the other gates. Deferring conversion
until every gate has
+ // drained deadlocks selective-reading multi-input operators: such
an operator only
+ // drains the selected input's end-of-state sentinel, so an
unselected gate would never
+ // drain (it is read only after conversion) while conversion would
wait for it to drain
+ // first. suspend() is still gated on completeAll(futures) below.
+ stateConsumed.thenRun(
+ () ->
+ mainMailboxExecutor.execute(
+ () -> inputGate.requestPartitions(false),
+ "Input gate request partitions"));
+ }
+ return completeAll(futures);
+ }
- // Must set the flag on input gates BEFORE starting the async read
task, because
- // finishReadRecoveredState() checks this flag to complete
bufferFilteringCompleteFuture.
- for (IndexedInputGate inputGate : inputGates) {
-
inputGate.setCheckpointingDuringRecoveryEnabled(checkpointingDuringRecoveryEnabled);
+ private void readInputChannelState(
+ SequentialChannelStateReader reader, IndexedInputGate[]
inputGates) {
+ try {
+ reader.readInputData(inputGates, createRecordFilterContext());
+
+ for (IndexedInputGate gate : inputGates) {
+ gate.finishReadRecoveredState(); // this is called from IO
thread - is that fine?
Review Comment:
I think so :)
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java:
##########
@@ -881,58 +918,120 @@ private CompletableFuture<Void> restoreStateAndGates(
INITIALIZE_STATE_DURATION, initializeStateEndTs -
readOutputDataTs);
IndexedInputGate[] inputGates = getEnvironment().getAllInputGates();
- boolean checkpointingDuringRecoveryEnabled =
-
CheckpointingOptions.isCheckpointingDuringRecoveryEnabled(getJobConfiguration());
+ recoveryCompletionFuture =
+
CheckpointingOptions.isCheckpointingDuringRecoveryEnabled(getJobConfiguration())
+ ? recoverChannelsWithCheckpointing(reader, inputGates)
+ : recoverChannelsWithoutCheckpointing(reader,
inputGates);
+
+ recoveryCompletionFuture.whenComplete((ign, throwable) ->
mailboxProcessor.suspend());
Review Comment:
I don't remember if this change from `thenRun` to `whenComplete` was
strictly necessary.
However, I don't see any downsides of using `whenComplete` here either.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
