1996fanrui commented on code in PR #28652:
URL: https://github.com/apache/flink/pull/28652#discussion_r3555147851
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RecoveredInputChannel.java:
##########
@@ -115,23 +115,49 @@ public void setChannelStateWriter(ChannelStateWriter
channelStateWriter) {
this.channelStateWriter = checkNotNull(channelStateWriter);
}
- public final InputChannel toInputChannel() throws IOException {
- Preconditions.checkState(
- bufferFilteringCompleteFuture.isDone(), "buffer filtering is
not complete");
- if (!inputGate.isCheckpointingDuringRecoveryEnabled()) {
- Preconditions.checkState(
- stateConsumedFuture.isDone(), "recovered state is not
fully consumed");
+ public final InputChannel toInputChannel(boolean needsRecovery) throws
IOException {
+ if (needsRecovery) {
+ return toInputChannelInRecovery();
+ }
+ synchronized (receivedBuffers) {
+ Preconditions.checkState(receivedBuffers.isEmpty(), "Received
buffer should be empty.");
}
- // Extract remaining buffers before conversion.
- // These buffers have been filtered but not yet consumed by the Task.
+ final InputChannel inputChannel =
toInputChannelInternal(needsRecovery);
+ inputChannel.setup();
+ inputChannel.checkpointStopped(lastStoppedCheckpointId);
+ return inputChannel;
+ }
+
+ /**
+ * FLINK-38544 transitional: removed when the spilling backend lands.
Creates the physical
+ * channel in recovery state and synchronously hands every queued
recovered buffer over through
+ * the push interface. The legacy {@link EndOfInputChannelStateEvent} in
the queue is dropped in
+ * translation; the {@link EndOfFetchedChannelStateEvent} sentinel takes
its place. The sentinel
+ * is appended directly instead of via {@link
+ * RecoverableInputChannel#finishRecoveredBufferDelivery()} because that
method waits for
+ * upstream readiness, which cannot happen while the mailbox thread is
still converting channels
+ * (partitions are requested only after conversion).
+ */
+ private InputChannel toInputChannelInRecovery() throws IOException {
final ArrayDeque<Buffer> remainingBuffers;
synchronized (receivedBuffers) {
remainingBuffers = new ArrayDeque<>(receivedBuffers);
receivedBuffers.clear();
}
- final InputChannel inputChannel =
toInputChannelInternal(remainingBuffers);
+ final InputChannel inputChannel = toInputChannelInternal(true);
+ inputChannel.setup();
+ final RecoverableInputChannel recoverableChannel =
(RecoverableInputChannel) inputChannel;
+ for (Buffer buffer : remainingBuffers) {
+ if (isEndOfInputChannelStateEvent(buffer)) {
+ buffer.recycleBuffer();
Review Comment:
done via 2147e6dd19fa70341788939d8c4a00df8b4a9d44
--
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]
