This is an automated email from the ASF dual-hosted git repository. 1996fanrui pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit bf14146ea155d46b1af17f699f5a1b7844319814 Author: Rui Fan <[email protected]> AuthorDate: Mon Jul 6 00:30:30 2026 +0200 [FLINK-39521][network] Propagate needsRecovery in PartitionRequest; start view reader with zero credit Propagate the consumer channel's needsRecovery flag to the producer so it withholds credit while the consumer's exclusive buffers are on loan to recovery: - NettyMessage.PartitionRequest gains a needsRecovery boolean (ctor arg, writeBoolean/readBoolean, +Byte.BYTES in the length calculation). This is a wire-format change (+1 byte per PartitionRequest); safe under the usual single-version-cluster assumption for the TM network protocol, stated in the PR description. - CreditBasedSequenceNumberingViewReader ctor gains needsRecovery; numCreditsAvailable = needsRecovery ? 0 : initialCredit. - PartitionRequestServerHandler passes request.needsRecovery; NettyPartitionRequestClient passes inputChannel.needsRecovery(). Mechanical test adaptations: NettyMessageServerSideSerializationTest (round-trips the new field), PartitionRequestServerHandlerTest, PartitionRequestQueueTest, PartitionRequestRegistrationTest, CreditBasedSequenceNumberingViewReaderTest, CancelPartitionRequestTest, ServerTransportErrorHandlingTest. --- .../CreditBasedSequenceNumberingViewReader.java | 9 ++++++-- .../runtime/io/network/netty/NettyMessage.java | 14 ++++++++++--- .../network/netty/NettyPartitionRequestClient.java | 3 ++- .../netty/PartitionRequestServerHandler.java | 5 ++++- .../network/netty/CancelPartitionRequestTest.java | 6 ++++-- ...CreditBasedSequenceNumberingViewReaderTest.java | 2 +- .../NettyMessageServerSideSerializationTest.java | 4 +++- .../network/netty/PartitionRequestQueueTest.java | 24 +++++++++++++--------- .../netty/PartitionRequestRegistrationTest.java | 9 +++++--- .../netty/PartitionRequestServerHandlerTest.java | 4 ++-- .../netty/ServerTransportErrorHandlingTest.java | 3 ++- 11 files changed, 56 insertions(+), 27 deletions(-) diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/CreditBasedSequenceNumberingViewReader.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/CreditBasedSequenceNumberingViewReader.java index ae57b39b08d..1ac2687bf39 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/CreditBasedSequenceNumberingViewReader.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/CreditBasedSequenceNumberingViewReader.java @@ -81,12 +81,17 @@ class CreditBasedSequenceNumberingViewReader private int numCreditsAvailable; CreditBasedSequenceNumberingViewReader( - InputChannelID receiverId, int initialCredit, PartitionRequestQueue requestQueue) { + InputChannelID receiverId, + int initialCredit, + boolean needsRecovery, + PartitionRequestQueue requestQueue) { checkArgument(initialCredit >= 0, "Must be non-negative."); this.receiverId = receiverId; this.initialCredit = initialCredit; - this.numCreditsAvailable = initialCredit; + // During spill recovery, exclusive buffers are on loan to the recovery drain; real credit + // is announced only after recovery completes. + this.numCreditsAvailable = needsRecovery ? 0 : initialCredit; this.requestQueue = requestQueue; this.subpartitionId = -1; } diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyMessage.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyMessage.java index ebe596e01f6..6747e139a03 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyMessage.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyMessage.java @@ -583,15 +583,19 @@ public abstract class NettyMessage { final int credit; + final boolean needsRecovery; + PartitionRequest( ResultPartitionID partitionId, ResultSubpartitionIndexSet queueIndexSet, InputChannelID receiverId, - int credit) { + int credit, + boolean needsRecovery) { this.partitionId = checkNotNull(partitionId); this.queueIndexSet = queueIndexSet; this.receiverId = checkNotNull(receiverId); this.credit = credit; + this.needsRecovery = needsRecovery; } @Override @@ -604,6 +608,7 @@ public abstract class NettyMessage { queueIndexSet.writeTo(bb); receiverId.writeTo(bb); bb.writeInt(credit); + bb.writeBoolean(needsRecovery); }; writeToChannel( @@ -616,7 +621,8 @@ public abstract class NettyMessage { + ExecutionAttemptID.getByteBufLength() + ResultSubpartitionIndexSet.getByteBufLength(queueIndexSet) + InputChannelID.getByteBufLength() - + Integer.BYTES); + + Integer.BYTES + + Byte.BYTES); } static PartitionRequest readFrom(ByteBuf buffer) { @@ -628,8 +634,10 @@ public abstract class NettyMessage { ResultSubpartitionIndexSet.fromByteBuf(buffer); InputChannelID receiverId = InputChannelID.fromByteBuf(buffer); int credit = buffer.readInt(); + boolean needsRecovery = buffer.readBoolean(); - return new PartitionRequest(partitionId, queueIndexSet, receiverId, credit); + return new PartitionRequest( + partitionId, queueIndexSet, receiverId, credit, needsRecovery); } @Override diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyPartitionRequestClient.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyPartitionRequestClient.java index 7cc52a234e0..737e3de7269 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyPartitionRequestClient.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyPartitionRequestClient.java @@ -129,7 +129,8 @@ public class NettyPartitionRequestClient implements PartitionRequestClient { partitionId, subpartitionIndexSet, inputChannel.getInputChannelId(), - inputChannel.getInitialCredit()); + inputChannel.getInitialCredit(), + inputChannel.needsRecovery()); final ChannelFutureListener listener = future -> { diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestServerHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestServerHandler.java index f93651b9f3e..6f2cd6b13ad 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestServerHandler.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestServerHandler.java @@ -85,7 +85,10 @@ class PartitionRequestServerHandler extends SimpleChannelInboundHandler<NettyMes NetworkSequenceViewReader reader; reader = new CreditBasedSequenceNumberingViewReader( - request.receiverId, request.credit, outboundQueue); + request.receiverId, + request.credit, + request.needsRecovery, + outboundQueue); reader.requestSubpartitionViewOrRegisterListener( partitionProvider, request.partitionId, request.queueIndexSet); diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/CancelPartitionRequestTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/CancelPartitionRequestTest.java index 2adbded7d9e..d165f6a50cb 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/CancelPartitionRequestTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/CancelPartitionRequestTest.java @@ -106,7 +106,8 @@ class CancelPartitionRequestTest { pid, new ResultSubpartitionIndexSet(0), new InputChannelID(), - Integer.MAX_VALUE)) + Integer.MAX_VALUE, + false)) .await(); // Wait for the notification @@ -170,7 +171,8 @@ class CancelPartitionRequestTest { pid, new ResultSubpartitionIndexSet(0), inputChannelId, - Integer.MAX_VALUE)) + Integer.MAX_VALUE, + false)) .await(); // Wait for the notification diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/CreditBasedSequenceNumberingViewReaderTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/CreditBasedSequenceNumberingViewReaderTest.java index cd4a3103240..ba686adce04 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/CreditBasedSequenceNumberingViewReaderTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/CreditBasedSequenceNumberingViewReaderTest.java @@ -88,7 +88,7 @@ class CreditBasedSequenceNumberingViewReaderTest { channel.close(); CreditBasedSequenceNumberingViewReader reader = new CreditBasedSequenceNumberingViewReader( - new InputChannelID(), initialCredit, queue); + new InputChannelID(), initialCredit, false, queue); reader.notifySubpartitionsCreated( TestingResultPartition.newBuilder() .setCreateSubpartitionViewFunction( diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/NettyMessageServerSideSerializationTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/NettyMessageServerSideSerializationTest.java index 22b5420b616..f2e0e8146c8 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/NettyMessageServerSideSerializationTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/NettyMessageServerSideSerializationTest.java @@ -67,7 +67,8 @@ class NettyMessageServerSideSerializationTest { new ResultPartitionID(), new ResultSubpartitionIndexSet(queueIndex), new InputChannelID(), - random.nextInt()); + random.nextInt(), + random.nextBoolean()); NettyMessage.PartitionRequest actual = encodeAndDecode(expected, channel); @@ -75,6 +76,7 @@ class NettyMessageServerSideSerializationTest { assertThat(actual.queueIndexSet).isEqualTo(expected.queueIndexSet); assertThat(actual.receiverId).isEqualTo(expected.receiverId); assertThat(actual.credit).isEqualTo(expected.credit); + assertThat(actual.needsRecovery).isEqualTo(expected.needsRecovery); } @Test diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueueTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueueTest.java index 7046a851809..a4e15e4cf90 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueueTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueueTest.java @@ -90,7 +90,8 @@ class PartitionRequestQueueTest { ResultPartitionManager resultPartitionManager = new ResultPartitionManager(); ResultPartitionID resultPartitionId = new ResultPartitionID(); CreditBasedSequenceNumberingViewReader reader = - new CreditBasedSequenceNumberingViewReader(new InputChannelID(0, 0), 10, queue); + new CreditBasedSequenceNumberingViewReader( + new InputChannelID(0, 0), 10, false, queue); reader.requestSubpartitionViewOrRegisterListener( resultPartitionManager, resultPartitionId, new ResultSubpartitionIndexSet(0)); @@ -132,9 +133,11 @@ class PartitionRequestQueueTest { EmbeddedChannel channel = new EmbeddedChannel(queue); CreditBasedSequenceNumberingViewReader reader1 = - new CreditBasedSequenceNumberingViewReader(new InputChannelID(0, 0), 10, queue); + new CreditBasedSequenceNumberingViewReader( + new InputChannelID(0, 0), 10, false, queue); CreditBasedSequenceNumberingViewReader reader2 = - new CreditBasedSequenceNumberingViewReader(new InputChannelID(1, 1), 10, queue); + new CreditBasedSequenceNumberingViewReader( + new InputChannelID(1, 1), 10, false, queue); ResultSubpartitionView view1 = new EmptyAlwaysAvailableResultSubpartitionView(); reader1.notifySubpartitionsCreated( @@ -192,7 +195,8 @@ class PartitionRequestQueueTest { final InputChannelID receiverId = new InputChannelID(); final PartitionRequestQueue queue = new PartitionRequestQueue(); final CreditBasedSequenceNumberingViewReader reader = - new CreditBasedSequenceNumberingViewReader(receiverId, Integer.MAX_VALUE, queue); + new CreditBasedSequenceNumberingViewReader( + receiverId, Integer.MAX_VALUE, false, queue); final EmbeddedChannel channel = new EmbeddedChannel(queue); reader.notifySubpartitionsCreated(partition, new ResultSubpartitionIndexSet(0)); @@ -287,7 +291,7 @@ class PartitionRequestQueueTest { final InputChannelID receiverId = new InputChannelID(); final PartitionRequestQueue queue = new PartitionRequestQueue(); final CreditBasedSequenceNumberingViewReader reader = - new CreditBasedSequenceNumberingViewReader(receiverId, 0, queue); + new CreditBasedSequenceNumberingViewReader(receiverId, 0, false, queue); final EmbeddedChannel channel = new EmbeddedChannel(queue); reader.notifySubpartitionsCreated(partition, new ResultSubpartitionIndexSet(0)); @@ -340,7 +344,7 @@ class PartitionRequestQueueTest { final InputChannelID receiverId = new InputChannelID(); final PartitionRequestQueue queue = new PartitionRequestQueue(); final CreditBasedSequenceNumberingViewReader reader = - new CreditBasedSequenceNumberingViewReader(receiverId, 2, queue); + new CreditBasedSequenceNumberingViewReader(receiverId, 2, false, queue); final EmbeddedChannel channel = new EmbeddedChannel(queue); reader.addCredit(-2); @@ -421,7 +425,7 @@ class PartitionRequestQueueTest { InputChannelID receiverId = new InputChannelID(); PartitionRequestQueue queue = new PartitionRequestQueue(); CreditBasedSequenceNumberingViewReader reader = - new CreditBasedSequenceNumberingViewReader(receiverId, 2, queue); + new CreditBasedSequenceNumberingViewReader(receiverId, 2, false, queue); EmbeddedChannel channel = new EmbeddedChannel(queue); reader.notifySubpartitionsCreated(partition, new ResultSubpartitionIndexSet(0)); @@ -461,7 +465,7 @@ class PartitionRequestQueueTest { PartitionRequestQueue queue = new PartitionRequestQueue(); InputChannelID receiverId = new InputChannelID(); CreditBasedSequenceNumberingViewReader reader = - new CreditBasedSequenceNumberingViewReader(receiverId, 0, queue); + new CreditBasedSequenceNumberingViewReader(receiverId, 0, false, queue); EmbeddedChannel channel = new EmbeddedChannel(queue); reader.notifySubpartitionsCreated(partition, new ResultSubpartitionIndexSet(0)); @@ -498,7 +502,7 @@ class PartitionRequestQueueTest { final InputChannelID receiverId = new InputChannelID(); final PartitionRequestQueue queue = new PartitionRequestQueue(); final CreditBasedSequenceNumberingViewReader reader = - new CreditBasedSequenceNumberingViewReader(receiverId, 2, queue); + new CreditBasedSequenceNumberingViewReader(receiverId, 2, false, queue); final EmbeddedChannel channel = new EmbeddedChannel(queue); reader.notifySubpartitionsCreated(partition, new ResultSubpartitionIndexSet(0)); @@ -546,7 +550,7 @@ class PartitionRequestQueueTest { InputChannelID receiverId = new InputChannelID(); PartitionRequestQueue queue = new PartitionRequestQueue(); CreditBasedSequenceNumberingViewReader reader = - new CreditBasedSequenceNumberingViewReader(receiverId, 2, queue); + new CreditBasedSequenceNumberingViewReader(receiverId, 2, false, queue); EmbeddedChannel channel = new EmbeddedChannel(queue); reader.notifySubpartitionsCreated(partition, new ResultSubpartitionIndexSet(0)); diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestRegistrationTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestRegistrationTest.java index 4080858b74b..86d9588094f 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestRegistrationTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestRegistrationTest.java @@ -95,7 +95,8 @@ class PartitionRequestRegistrationTest { resultPartition.getPartitionId(), new ResultSubpartitionIndexSet(0), new InputChannelID(), - Integer.MAX_VALUE)) + Integer.MAX_VALUE, + false)) .await(); // Wait for the notification @@ -143,7 +144,8 @@ class PartitionRequestRegistrationTest { resultPartition.getPartitionId(), new ResultSubpartitionIndexSet(0), new InputChannelID(), - Integer.MAX_VALUE)) + Integer.MAX_VALUE, + false)) .await(); // Register result partition after partition request @@ -212,7 +214,8 @@ class PartitionRequestRegistrationTest { pid, new ResultSubpartitionIndexSet(0), remoteInputChannel.getInputChannelId(), - Integer.MAX_VALUE)) + Integer.MAX_VALUE, + false)) .await(); // Wait for the notification diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestServerHandlerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestServerHandlerTest.java index 7f19b199582..1fe0c00e79a 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestServerHandlerTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestServerHandlerTest.java @@ -81,7 +81,7 @@ class PartitionRequestServerHandlerTest { // Creates and registers the view to netty. NetworkSequenceViewReader viewReader = new CreditBasedSequenceNumberingViewReader( - inputChannelID, 2, partitionRequestQueue); + inputChannelID, 2, false, partitionRequestQueue); viewReader.notifySubpartitionsCreated(resultPartition, new ResultSubpartitionIndexSet(0)); partitionRequestQueue.notifyReaderCreated(viewReader); @@ -149,7 +149,7 @@ class PartitionRequestServerHandlerTest { TestViewReader( InputChannelID receiverId, int initialCredit, PartitionRequestQueue requestQueue) { - super(receiverId, initialCredit, requestQueue); + super(receiverId, initialCredit, false, requestQueue); } @Override diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/ServerTransportErrorHandlingTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/ServerTransportErrorHandlingTest.java index a4f753fe1e7..bfc4e01a153 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/ServerTransportErrorHandlingTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/ServerTransportErrorHandlingTest.java @@ -103,7 +103,8 @@ class ServerTransportErrorHandlingTest { new ResultPartitionID(), new ResultSubpartitionIndexSet(0), new InputChannelID(), - Integer.MAX_VALUE)); + Integer.MAX_VALUE, + false)); // Wait for the notification assertThat(sync.await(TestingUtils.TESTING_DURATION.toMillis(), TimeUnit.MILLISECONDS))
