You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by fa...@apache.org on 2023/07/14 13:20:10 UTC
[flink] 01/03: [FLINK-32298][refactor] Using the static const instead of magic code in ResultSubpartition
This is an automated email from the ASF dual-hosted git repository.
fanrui pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 72a30e5d6453b7289330046d1907de62ac259d04
Author: 1996fanrui <19...@gmail.com>
AuthorDate: Thu Jun 8 16:12:50 2023 +0800
[FLINK-32298][refactor] Using the static const instead of magic code in ResultSubpartition
---
.../runtime/io/network/partition/BoundedBlockingSubpartition.java | 2 +-
.../runtime/io/network/partition/BufferWritingResultPartition.java | 7 +++----
.../flink/runtime/io/network/partition/PipelinedSubpartition.java | 4 ++--
.../flink/runtime/io/network/partition/ResultSubpartition.java | 6 +++++-
4 files changed, 11 insertions(+), 8 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/BoundedBlockingSubpartition.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/BoundedBlockingSubpartition.java
index c09975cc606..b9d8d5105a2 100755
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/BoundedBlockingSubpartition.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/BoundedBlockingSubpartition.java
@@ -125,7 +125,7 @@ final class BoundedBlockingSubpartition extends ResultSubpartition {
public int add(BufferConsumer bufferConsumer, int partialRecordLength) throws IOException {
if (isFinished()) {
bufferConsumer.close();
- return -1;
+ return ADD_BUFFER_ERROR_CODE;
}
flushCurrentBuffer();
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/BufferWritingResultPartition.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/BufferWritingResultPartition.java
index 2107a24c747..85a6163bb70 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/BufferWritingResultPartition.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/BufferWritingResultPartition.java
@@ -374,10 +374,9 @@ public abstract class BufferWritingResultPartition extends ResultPartition {
int desirableBufferSize = Integer.MAX_VALUE;
for (ResultSubpartition subpartition : subpartitions) {
int subPartitionBufferSize = subpartition.add(consumer.copy(), partialRecordBytes);
- desirableBufferSize =
- subPartitionBufferSize > 0
- ? Math.min(desirableBufferSize, subPartitionBufferSize)
- : desirableBufferSize;
+ if (subPartitionBufferSize != ResultSubpartition.ADD_BUFFER_ERROR_CODE) {
+ desirableBufferSize = Math.min(desirableBufferSize, subPartitionBufferSize);
+ }
}
resizeBuffer(buffer, desirableBufferSize, minDesirableBufferSize);
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java
index 08d8ac34de9..8050853aa5c 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java
@@ -165,7 +165,7 @@ public class PipelinedSubpartition extends ResultSubpartition
bufferConsumer,
parent.getOwningTaskName(),
subpartitionInfo);
- if (add(bufferConsumer, Integer.MIN_VALUE) == -1) {
+ if (add(bufferConsumer, Integer.MIN_VALUE) == ADD_BUFFER_ERROR_CODE) {
throw new IOException("Buffer consumer couldn't be added to ResultSubpartition");
}
}
@@ -192,7 +192,7 @@ public class PipelinedSubpartition extends ResultSubpartition
synchronized (buffers) {
if (isFinished || isReleased) {
bufferConsumer.close();
- return -1;
+ return ADD_BUFFER_ERROR_CODE;
}
// Add the bufferConsumer and update the stats
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultSubpartition.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultSubpartition.java
index 300d5a41061..03fee8d303a 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultSubpartition.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultSubpartition.java
@@ -31,6 +31,9 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
/** A single subpartition of a {@link ResultPartition} instance. */
public abstract class ResultSubpartition {
+ // The error code when adding a buffer fails.
+ public static final int ADD_BUFFER_ERROR_CODE = -1;
+
/** The info of the subpartition to identify it globally within a task. */
protected final ResultSubpartitionInfo subpartitionInfo;
@@ -85,7 +88,8 @@ public abstract class ResultSubpartition {
* @param bufferConsumer the buffer to add (transferring ownership to this writer)
* @param partialRecordLength the length of bytes to skip in order to start with a complete
* record, from position index 0 of the underlying {@cite MemorySegment}.
- * @return the preferable buffer size for this subpartition or -1 if the add operation fails.
+ * @return the preferable buffer size for this subpartition or {@link #ADD_BUFFER_ERROR_CODE} if
+ * the add operation fails.
* @throws IOException thrown in case of errors while adding the buffer
*/
public abstract int add(BufferConsumer bufferConsumer, int partialRecordLength)