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/11 18:55:18 UTC
[flink] branch master updated (f22b6dd -> 3e9e530)
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git.
from f22b6dd [FLINK-23693][tests] Add principal creation possibilities to SecureTestEnvironment
new e6996cd [hotfix][runtime] setThroughputMeter renamed to setThroughputCalculator according to name of the set object.
new 7478445 [refactor][runtime] Unification of obtaining reader in PartitionRequestQueue
new 77f7b7f [FLINK-23454][runtime] NewBufferSize migrated from long to int for bufferSize because the bufferSize is always int.
new 7560fe3 [FLINK-23454][runtime] Added the support of changing the max capacity to BufferBuilder
new 703662a [FLINK-23454][runtime] Subpartition is able to notify the desirable buffer size for input buffer consumer.
new ebbc860 [FLINK-23454][runtime] Notifying the subpartitions about the new received buffer size.
new 88c5091 [FLINK-23454][runtime] LocalInputChannel notifying the subpartition about the new buffer size directly.
new 545b612 [FLINK-23454][runtime] ResultPartition changes the BufferBuilder size according to the desirable buffer size of each subpartition.
new 3e9e530 [FLINK-23726][runtime] Read buffer debloat configuration from task manager configuration rather than task configuration.
The 9 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../state/api/runtime/SavepointEnvironment.java | 2 +-
.../flink/runtime/execution/Environment.java | 2 +-
.../io/network/NetworkSequenceViewReader.java | 2 +
.../runtime/io/network/buffer/BufferBuilder.java | 13 +++-
.../CreditBasedSequenceNumberingViewReader.java | 5 ++
.../runtime/io/network/netty/NettyMessage.java | 10 ++--
.../io/network/netty/PartitionRequestQueue.java | 29 +++++----
.../netty/PartitionRequestServerHandler.java | 1 +
.../partition/BoundedBlockingSubpartition.java | 11 +++-
...edBlockingSubpartitionDirectTransferReader.java | 5 ++
.../BoundedBlockingSubpartitionReader.java | 5 ++
.../partition/BufferWritingResultPartition.java | 18 +++++-
.../partition/NoOpResultSubpartitionView.java | 3 +
.../network/partition/PipelinedSubpartition.java | 24 ++++++--
.../partition/PipelinedSubpartitionView.java | 5 ++
.../io/network/partition/ResultSubpartition.java | 8 ++-
.../network/partition/ResultSubpartitionView.java | 2 +
.../partition/SortMergeSubpartitionReader.java | 3 +
.../partition/consumer/LocalInputChannel.java | 5 +-
.../runtime/taskmanager/RuntimeEnvironment.java | 2 +-
.../buffer/BufferBuilderAndConsumerTest.java | 42 +++++++++++++
.../network/netty/CancelPartitionRequestTest.java | 3 +
...editBasedPartitionRequestClientHandlerTest.java | 4 +-
.../NettyMessageServerSideSerializationTest.java | 2 +-
.../network/netty/PartitionRequestQueueTest.java | 50 ++++++++++++++++
.../netty/PartitionRequestServerHandlerTest.java | 28 +++++++++
.../partition/PipelinedSubpartitionTest.java | 40 +++++++++++++
.../io/network/partition/ResultPartitionTest.java | 70 ++++++++++++++++++++++
.../io/network/partition/SubpartitionTestBase.java | 4 +-
.../partition/consumer/LocalInputChannelTest.java | 52 +++++++++++++---
.../operators/testutils/DummyEnvironment.java | 2 +-
.../operators/testutils/MockEnvironment.java | 2 +-
.../testutils/MockEnvironmentBuilder.java | 3 +-
.../flink/streaming/runtime/tasks/StreamTask.java | 12 ++--
.../runtime/tasks/StreamMockEnvironment.java | 2 +-
.../tasks/StreamTaskMailboxTestHarnessBuilder.java | 13 +++-
.../streaming/runtime/tasks/StreamTaskTest.java | 24 ++++----
37 files changed, 440 insertions(+), 68 deletions(-)
[flink] 06/09: [FLINK-23454][runtime] Notifying the subpartitions
about the new received buffer size.
Posted by pn...@apache.org.
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 ebbc8608f09e8329e49f27ab9ae87a736dcac7ed
Author: Anton Kalashnikov <ka...@yandex.ru>
AuthorDate: Thu Aug 5 13:42:02 2021 +0200
[FLINK-23454][runtime] Notifying the subpartitions about the new received buffer size.
---
.../io/network/NetworkSequenceViewReader.java | 2 +
.../CreditBasedSequenceNumberingViewReader.java | 5 +++
.../io/network/netty/PartitionRequestQueue.java | 8 ++++
.../netty/PartitionRequestServerHandler.java | 1 +
...edBlockingSubpartitionDirectTransferReader.java | 5 +++
.../BoundedBlockingSubpartitionReader.java | 5 +++
.../partition/NoOpResultSubpartitionView.java | 3 ++
.../partition/PipelinedSubpartitionView.java | 5 +++
.../network/partition/ResultSubpartitionView.java | 2 +
.../partition/SortMergeSubpartitionReader.java | 3 ++
.../network/netty/CancelPartitionRequestTest.java | 3 ++
.../network/netty/PartitionRequestQueueTest.java | 50 ++++++++++++++++++++++
.../netty/PartitionRequestServerHandlerTest.java | 28 ++++++++++++
13 files changed, 120 insertions(+)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkSequenceViewReader.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkSequenceViewReader.java
index d5daaa1..978ab96 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkSequenceViewReader.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkSequenceViewReader.java
@@ -82,4 +82,6 @@ public interface NetworkSequenceViewReader {
Throwable getFailureCause();
InputChannelID getReceiverId();
+
+ void notifyNewBufferSize(int newBufferSize);
}
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 ee553c4..ba41919 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
@@ -172,6 +172,11 @@ class CreditBasedSequenceNumberingViewReader
return receiverId;
}
+ @Override
+ public void notifyNewBufferSize(int newBufferSize) {
+ subpartitionView.notifyNewBufferSize(newBufferSize);
+ }
+
@VisibleForTesting
int getNumCreditsAvailable() {
return numCreditsAvailable;
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java
index a656bea..7b3ec69 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java
@@ -180,6 +180,14 @@ class PartitionRequestQueue extends ChannelInboundHandlerAdapter {
obtainReader(receiverId).acknowledgeAllRecordsProcessed();
}
+ void notifyNewBufferSize(InputChannelID receiverId, int newBufferSize) {
+ if (fatalError) {
+ return;
+ }
+
+ obtainReader(receiverId).notifyNewBufferSize(newBufferSize);
+ }
+
NetworkSequenceViewReader obtainReader(InputChannelID receiverId) {
NetworkSequenceViewReader reader = allReaders.get(receiverId);
if (reader == null) {
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 1c8af58..14fef21 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
@@ -131,6 +131,7 @@ class PartitionRequestServerHandler extends SimpleChannelInboundHandler<NettyMes
} else if (msgClazz == NewBufferSize.class) {
NewBufferSize request = (NewBufferSize) msg;
+ outboundQueue.notifyNewBufferSize(request.receiverId, request.bufferSize);
} else {
LOG.warn("Received unexpected client request: {}", msg);
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/BoundedBlockingSubpartitionDirectTransferReader.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/BoundedBlockingSubpartitionDirectTransferReader.java
index 2d5d420..17a09e8 100755
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/BoundedBlockingSubpartitionDirectTransferReader.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/BoundedBlockingSubpartitionDirectTransferReader.java
@@ -159,6 +159,11 @@ public class BoundedBlockingSubpartitionDirectTransferReader implements ResultSu
}
@Override
+ public void notifyNewBufferSize(int newBufferSize) {
+ parent.bufferSize(newBufferSize);
+ }
+
+ @Override
public void notifyDataAvailable() {
throw new UnsupportedOperationException("Method should never be called.");
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/BoundedBlockingSubpartitionReader.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/BoundedBlockingSubpartitionReader.java
index 72360f3..3c4f779 100755
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/BoundedBlockingSubpartitionReader.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/BoundedBlockingSubpartitionReader.java
@@ -188,6 +188,11 @@ final class BoundedBlockingSubpartitionReader implements ResultSubpartitionView
}
@Override
+ public void notifyNewBufferSize(int newBufferSize) {
+ parent.bufferSize(newBufferSize);
+ }
+
+ @Override
public String toString() {
return String.format(
"Blocking Subpartition Reader: ID=%s, index=%d",
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/NoOpResultSubpartitionView.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/NoOpResultSubpartitionView.java
index d81b1de..6cb4879 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/NoOpResultSubpartitionView.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/NoOpResultSubpartitionView.java
@@ -64,4 +64,7 @@ public class NoOpResultSubpartitionView implements ResultSubpartitionView {
public int getNumberOfQueuedBuffers() {
return 0;
}
+
+ @Override
+ public void notifyNewBufferSize(int newBufferSize) {}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartitionView.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartitionView.java
index 2c67c01..fa3efd6 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartitionView.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartitionView.java
@@ -105,6 +105,11 @@ public class PipelinedSubpartitionView implements ResultSubpartitionView {
}
@Override
+ public void notifyNewBufferSize(int newBufferSize) {
+ parent.bufferSize(newBufferSize);
+ }
+
+ @Override
public String toString() {
return String.format(
"%s(index: %d) of ResultPartition %s",
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultSubpartitionView.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultSubpartitionView.java
index 564f3ee..4ee89bb 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultSubpartitionView.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultSubpartitionView.java
@@ -63,6 +63,8 @@ public interface ResultSubpartitionView {
int getNumberOfQueuedBuffers();
+ void notifyNewBufferSize(int newBufferSize);
+
/**
* Availability of the {@link ResultSubpartitionView} and the backlog in the corresponding
* {@link ResultSubpartition}.
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SortMergeSubpartitionReader.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SortMergeSubpartitionReader.java
index 9b62995..cd04d6f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SortMergeSubpartitionReader.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SortMergeSubpartitionReader.java
@@ -242,4 +242,7 @@ class SortMergeSubpartitionReader
return buffersRead.size();
}
}
+
+ @Override
+ public void notifyNewBufferSize(int newBufferSize) {}
}
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 5368a2d..e61d97d 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
@@ -247,6 +247,9 @@ public class CancelPartitionRequestTest {
}
@Override
+ public void notifyNewBufferSize(int newBufferSize) {}
+
+ @Override
public Throwable getFailureCause() {
return null;
}
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 a6d920c..5fc9ab9 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
@@ -21,6 +21,7 @@ package org.apache.flink.runtime.io.network.netty;
import org.apache.flink.runtime.execution.CancelTaskException;
import org.apache.flink.runtime.io.disk.FileChannelManager;
import org.apache.flink.runtime.io.disk.FileChannelManagerImpl;
+import org.apache.flink.runtime.io.disk.NoOpFileChannelManager;
import org.apache.flink.runtime.io.network.NettyShuffleEnvironment;
import org.apache.flink.runtime.io.network.NettyShuffleEnvironmentBuilder;
import org.apache.flink.runtime.io.network.NetworkSequenceViewReader;
@@ -528,6 +529,55 @@ public class PartitionRequestQueueTest {
channel.close();
}
+ @Test
+ public void testNotifyNewBufferSize() throws Exception {
+ // given: Result partition and the reader for subpartition 0.
+ ResultPartition parent = createResultPartition();
+
+ BufferAvailabilityListener bufferAvailabilityListener = new NoOpBufferAvailablityListener();
+ ResultSubpartitionView view = parent.createSubpartitionView(0, bufferAvailabilityListener);
+ ResultPartitionProvider partitionProvider =
+ (partitionId, index, availabilityListener) -> view;
+
+ InputChannelID receiverId = new InputChannelID();
+ PartitionRequestQueue queue = new PartitionRequestQueue();
+ CreditBasedSequenceNumberingViewReader reader =
+ new CreditBasedSequenceNumberingViewReader(receiverId, 2, queue);
+ EmbeddedChannel channel = new EmbeddedChannel(queue);
+
+ reader.requestSubpartitionView(partitionProvider, new ResultPartitionID(), 0);
+ queue.notifyReaderCreated(reader);
+
+ // when: New buffer size received.
+ queue.notifyNewBufferSize(receiverId, 65);
+
+ // and: New records emit.
+ parent.emitRecord(ByteBuffer.allocate(128), 0);
+ parent.emitRecord(ByteBuffer.allocate(10), 0);
+
+ reader.notifyDataAvailable();
+ channel.runPendingTasks();
+
+ // then: Buffers of received size will be in outbound channel.
+ Object data1 = channel.readOutbound();
+ assertEquals(65, ((NettyMessage.BufferResponse) data1).buffer.getSize());
+ Object data2 = channel.readOutbound();
+ assertEquals(65, ((NettyMessage.BufferResponse) data2).buffer.getSize());
+ }
+
+ private static ResultPartition createResultPartition() throws IOException {
+ NettyShuffleEnvironment network =
+ new NettyShuffleEnvironmentBuilder()
+ .setNumNetworkBuffers(10)
+ .setBufferSize(BUFFER_SIZE)
+ .build();
+ ResultPartition resultPartition =
+ createPartition(
+ network, NoOpFileChannelManager.INSTANCE, ResultPartitionType.PIPELINED, 2);
+ resultPartition.setup();
+ return resultPartition;
+ }
+
private static ResultPartition createFinishedPartitionWithFilledData(
ResultPartitionManager partitionManager) throws Exception {
NettyShuffleEnvironment environment =
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 6348345..4e6f4b8 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
@@ -42,6 +42,7 @@ import java.util.concurrent.CompletableFuture;
import static org.hamcrest.Matchers.instanceOf;
import static org.hamcrest.Matchers.is;
+import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
@@ -138,8 +139,30 @@ public class PartitionRequestServerHandlerTest extends TestLogger {
assertFalse(allRecordsProcessedFuture.isCompletedExceptionally());
}
+ @Test
+ public void testNewBufferSize() {
+ final InputChannelID inputChannelID = new InputChannelID();
+ final PartitionRequestQueue partitionRequestQueue = new PartitionRequestQueue();
+ final TestViewReader testViewReader =
+ new TestViewReader(inputChannelID, 2, partitionRequestQueue);
+ final PartitionRequestServerHandler serverHandler =
+ new PartitionRequestServerHandler(
+ new ResultPartitionManager(),
+ new TaskEventDispatcher(),
+ partitionRequestQueue);
+ final EmbeddedChannel channel = new EmbeddedChannel(serverHandler);
+ partitionRequestQueue.notifyReaderCreated(testViewReader);
+
+ // Write the message of new buffer size to server
+ channel.writeInbound(new NettyMessage.NewBufferSize(666, inputChannelID));
+ channel.runPendingTasks();
+
+ assertEquals(666, testViewReader.bufferSize);
+ }
+
private static class TestViewReader extends CreditBasedSequenceNumberingViewReader {
private boolean consumptionResumed = false;
+ private int bufferSize;
TestViewReader(
InputChannelID receiverId, int initialCredit, PartitionRequestQueue requestQueue) {
@@ -150,5 +173,10 @@ public class PartitionRequestServerHandlerTest extends TestLogger {
public void resumeConsumption() {
consumptionResumed = true;
}
+
+ @Override
+ public void notifyNewBufferSize(int newBufferSize) {
+ bufferSize = newBufferSize;
+ }
}
}
[flink] 01/09: [hotfix][runtime] setThroughputMeter renamed to
setThroughputCalculator according to name of the set object.
Posted by pn...@apache.org.
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 e6996cd0b1bb7059ab32f63f051ebdedf93bcc9d
Author: Anton Kalashnikov <ka...@yandex.ru>
AuthorDate: Thu Aug 5 13:16:50 2021 +0200
[hotfix][runtime] setThroughputMeter renamed to setThroughputCalculator according to name of the set object.
---
.../org/apache/flink/state/api/runtime/SavepointEnvironment.java | 2 +-
.../main/java/org/apache/flink/runtime/execution/Environment.java | 2 +-
.../org/apache/flink/runtime/taskmanager/RuntimeEnvironment.java | 2 +-
.../apache/flink/runtime/operators/testutils/DummyEnvironment.java | 2 +-
.../apache/flink/runtime/operators/testutils/MockEnvironment.java | 2 +-
.../flink/runtime/operators/testutils/MockEnvironmentBuilder.java | 3 ++-
.../java/org/apache/flink/streaming/runtime/tasks/StreamTask.java | 2 +-
.../apache/flink/streaming/runtime/tasks/StreamMockEnvironment.java | 2 +-
.../runtime/tasks/StreamTaskMailboxTestHarnessBuilder.java | 2 +-
.../org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java | 6 +++---
10 files changed, 13 insertions(+), 12 deletions(-)
diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/runtime/SavepointEnvironment.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/runtime/SavepointEnvironment.java
index 73a2756..62e0250 100644
--- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/runtime/SavepointEnvironment.java
+++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/runtime/SavepointEnvironment.java
@@ -288,7 +288,7 @@ public class SavepointEnvironment implements Environment {
}
@Override
- public ThroughputCalculator getThroughputMeter() {
+ public ThroughputCalculator getThroughputCalculator() {
// The throughput calculator doesn't make sense for savepoint but the not null value is
// preferable when StreamTask is instantiated.
return new ThroughputCalculator(SystemClock.getInstance(), 10);
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/execution/Environment.java b/flink-runtime/src/main/java/org/apache/flink/runtime/execution/Environment.java
index caf8e0f..c6ab128 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/execution/Environment.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/execution/Environment.java
@@ -241,7 +241,7 @@ public interface Environment {
*
* @return the throughput calculation service.
*/
- ThroughputCalculator getThroughputMeter();
+ ThroughputCalculator getThroughputCalculator();
// --------------------------------------------------------------------------------------------
// Fields set in the StreamTask to provide access to mailbox and other runtime resources
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/RuntimeEnvironment.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/RuntimeEnvironment.java
index d78b972..8ad2798 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/RuntimeEnvironment.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/RuntimeEnvironment.java
@@ -345,7 +345,7 @@ public class RuntimeEnvironment implements Environment {
}
@Override
- public ThroughputCalculator getThroughputMeter() {
+ public ThroughputCalculator getThroughputCalculator() {
return throughputCalculator;
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/DummyEnvironment.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/DummyEnvironment.java
index 55da6301..f5faead 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/DummyEnvironment.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/DummyEnvironment.java
@@ -252,7 +252,7 @@ public class DummyEnvironment implements Environment {
}
@Override
- public ThroughputCalculator getThroughputMeter() {
+ public ThroughputCalculator getThroughputCalculator() {
return new ThroughputCalculator(SystemClock.getInstance(), 10);
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/MockEnvironment.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/MockEnvironment.java
index 4dc92c5..0331d07 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/MockEnvironment.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/MockEnvironment.java
@@ -318,7 +318,7 @@ public class MockEnvironment implements Environment, AutoCloseable {
}
@Override
- public ThroughputCalculator getThroughputMeter() {
+ public ThroughputCalculator getThroughputCalculator() {
return throughputCalculator;
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/MockEnvironmentBuilder.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/MockEnvironmentBuilder.java
index 32fed2d..b126130 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/MockEnvironmentBuilder.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/MockEnvironmentBuilder.java
@@ -163,7 +163,8 @@ public class MockEnvironmentBuilder {
return this;
}
- public MockEnvironmentBuilder setThroughputMeter(ThroughputCalculator throughputCalculator) {
+ public MockEnvironmentBuilder setThroughputCalculator(
+ ThroughputCalculator throughputCalculator) {
this.throughputCalculator = throughputCalculator;
return this;
}
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 1132c34..47c9dff 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
@@ -405,7 +405,7 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>> extends Ab
injectChannelStateWriterIntoChannels();
environment.getMetricGroup().getIOMetricGroup().setEnableBusyTime(true);
- this.throughputCalculator = environment.getThroughputMeter();
+ this.throughputCalculator = environment.getThroughputCalculator();
this.bufferDebloatPeriod = getTaskConfiguration().get(BUFFER_DEBLOAT_PERIOD).toMillis();
if (getTaskConfiguration().get(TaskManagerOptions.BUFFER_DEBLOAT_ENABLED)) {
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamMockEnvironment.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamMockEnvironment.java
index 5cad561..ab1eff0 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamMockEnvironment.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamMockEnvironment.java
@@ -309,7 +309,7 @@ public class StreamMockEnvironment implements Environment {
}
@Override
- public ThroughputCalculator getThroughputMeter() {
+ public ThroughputCalculator getThroughputCalculator() {
return throughputCalculator;
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskMailboxTestHarnessBuilder.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskMailboxTestHarnessBuilder.java
index ded2d24..c92bb9c 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskMailboxTestHarnessBuilder.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskMailboxTestHarnessBuilder.java
@@ -126,7 +126,7 @@ public class StreamTaskMailboxTestHarnessBuilder<OUT> {
return this;
}
- public <T> StreamTaskMailboxTestHarnessBuilder<OUT> setThroughputMeter(
+ public <T> StreamTaskMailboxTestHarnessBuilder<OUT> setThroughputCalculator(
ThroughputCalculator throughputCalculator) {
this.throughputCalculator = throughputCalculator;
return this;
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 0d8558e..d5e149b 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
@@ -1510,7 +1510,7 @@ public class StreamTaskTest extends TestLogger {
.build();
TaskIOMetricGroup ioMetricGroup =
task.getEnvironment().getMetricGroup().getIOMetricGroup();
- ThroughputCalculator throughputCalculator = environment.getThroughputMeter();
+ ThroughputCalculator throughputCalculator = environment.getThroughputCalculator();
final MailboxExecutor executor = task.mailboxProcessor.getMainMailboxExecutor();
final RunnableWithException completeFutureTask =
@@ -1791,7 +1791,7 @@ public class StreamTaskTest extends TestLogger {
.addInput(STRING_TYPE_INFO)
.setupOutputForSingletonOperatorChain(
new TestBoundedOneInputStreamOperator())
- .setThroughputMeter(
+ .setThroughputCalculator(
new ThroughputCalculator(SystemClock.getInstance(), 10) {
@Override
public long calculateThroughput() {
@@ -1864,7 +1864,7 @@ public class StreamTaskTest extends TestLogger {
.addInput(STRING_TYPE_INFO, inputChannels)
.setupOutputForSingletonOperatorChain(
new TestBoundedOneInputStreamOperator())
- .setThroughputMeter(
+ .setThroughputCalculator(
new ThroughputCalculator(SystemClock.getInstance(), 10) {
@Override
public long calculateThroughput() {
[flink] 04/09: [FLINK-23454][runtime] Added the support of changing
the max capacity to BufferBuilder
Posted by pn...@apache.org.
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 7560fe3196680a1fc3ad2eb63a33b096df146574
Author: Anton Kalashnikov <ka...@yandex.ru>
AuthorDate: Thu Aug 5 13:24:23 2021 +0200
[FLINK-23454][runtime] Added the support of changing the max capacity to BufferBuilder
---
.../runtime/io/network/buffer/BufferBuilder.java | 13 ++++++-
.../buffer/BufferBuilderAndConsumerTest.java | 42 ++++++++++++++++++++++
2 files changed, 54 insertions(+), 1 deletion(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferBuilder.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferBuilder.java
index 36d21e6..e977d61 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferBuilder.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferBuilder.java
@@ -37,6 +37,7 @@ import static org.apache.flink.util.Preconditions.checkState;
public class BufferBuilder implements AutoCloseable {
private final Buffer buffer;
private final MemorySegment memorySegment;
+ private int maxCapacity;
private final SettablePositionMarker positionMarker = new SettablePositionMarker();
@@ -45,6 +46,7 @@ public class BufferBuilder implements AutoCloseable {
public BufferBuilder(MemorySegment memorySegment, BufferRecycler recycler) {
this.memorySegment = checkNotNull(memorySegment);
this.buffer = new NetworkBuffer(memorySegment, recycler);
+ this.maxCapacity = buffer.getMaxCapacity();
}
/**
@@ -142,7 +144,16 @@ public class BufferBuilder implements AutoCloseable {
}
public int getMaxCapacity() {
- return buffer.getMaxCapacity();
+ return maxCapacity;
+ }
+
+ /**
+ * The result capacity can not be greater than allocated memorySegment. It also can not be less
+ * than already written data.
+ */
+ public void trim(int newSize) {
+ maxCapacity =
+ Math.min(Math.max(newSize, positionMarker.getCached()), buffer.getMaxCapacity());
}
@Override
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/BufferBuilderAndConsumerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/BufferBuilderAndConsumerTest.java
index 029e2da..b927917 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/BufferBuilderAndConsumerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/BufferBuilderAndConsumerTest.java
@@ -264,6 +264,48 @@ public class BufferBuilderAndConsumerTest {
assertEquals(1, recycler.recycleInvocationCounter);
}
+ @Test
+ public void trimToAvailableSize() {
+ BufferBuilder bufferBuilder = createBufferBuilder();
+ assertEquals(BUFFER_SIZE, bufferBuilder.getMaxCapacity());
+
+ bufferBuilder.trim(BUFFER_SIZE / 2);
+ assertEquals(BUFFER_SIZE / 2, bufferBuilder.getMaxCapacity());
+
+ bufferBuilder.trim(0);
+ assertEquals(0, bufferBuilder.getMaxCapacity());
+ }
+
+ @Test
+ public void trimToNegativeSize() {
+ BufferBuilder bufferBuilder = createBufferBuilder();
+ assertEquals(BUFFER_SIZE, bufferBuilder.getMaxCapacity());
+
+ bufferBuilder.trim(-1);
+ assertEquals(0, bufferBuilder.getMaxCapacity());
+ }
+
+ @Test
+ public void trimToSizeLessThanWritten() {
+ BufferBuilder bufferBuilder = createBufferBuilder();
+ assertEquals(BUFFER_SIZE, bufferBuilder.getMaxCapacity());
+
+ bufferBuilder.append(toByteBuffer(1, 2, 3));
+
+ bufferBuilder.trim(4);
+ // Should be minimum possible size = 3 * int == 12.
+ assertEquals(12, bufferBuilder.getMaxCapacity());
+ }
+
+ @Test
+ public void trimToSizeGreaterThanMax() {
+ BufferBuilder bufferBuilder = createBufferBuilder();
+ assertEquals(BUFFER_SIZE, bufferBuilder.getMaxCapacity());
+
+ bufferBuilder.trim(BUFFER_SIZE + 1);
+ assertEquals(BUFFER_SIZE, bufferBuilder.getMaxCapacity());
+ }
+
private static void testIsFinished(int writes) {
BufferBuilder bufferBuilder = createBufferBuilder();
BufferConsumer bufferConsumer = bufferBuilder.createBufferConsumer();
[flink] 07/09: [FLINK-23454][runtime] LocalInputChannel notifying
the subpartition about the new buffer size directly.
Posted by pn...@apache.org.
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 88c5091a4d152918fb71844f5c1f15a2d20a1ee1
Author: Anton Kalashnikov <ka...@yandex.ru>
AuthorDate: Thu Aug 5 13:43:12 2021 +0200
[FLINK-23454][runtime] LocalInputChannel notifying the subpartition about the new buffer size directly.
---
.../partition/consumer/LocalInputChannel.java | 5 ++-
.../partition/consumer/LocalInputChannelTest.java | 52 +++++++++++++++++++---
2 files changed, 49 insertions(+), 8 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java
index e54a4c1..ba1c075 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java
@@ -338,7 +338,10 @@ public class LocalInputChannel extends InputChannel implements BufferAvailabilit
@Override
void announceBufferSize(int newBufferSize) {
- // Not supported.
+ checkState(!isReleased, "Channel released.");
+
+ ResultSubpartitionView subpartitionView = checkNotNull(this.subpartitionView);
+ subpartitionView.notifyNewBufferSize(newBufferSize);
}
@Override
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannelTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannelTest.java
index 0823b60..f3f33b8 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannelTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannelTest.java
@@ -27,7 +27,6 @@ import org.apache.flink.runtime.io.network.TaskEventDispatcher;
import org.apache.flink.runtime.io.network.api.CheckpointBarrier;
import org.apache.flink.runtime.io.network.api.serialization.EventSerializer;
import org.apache.flink.runtime.io.network.buffer.Buffer;
-import org.apache.flink.runtime.io.network.buffer.BufferBuilderTestUtils;
import org.apache.flink.runtime.io.network.buffer.BufferConsumer;
import org.apache.flink.runtime.io.network.buffer.BufferPool;
import org.apache.flink.runtime.io.network.buffer.BufferProvider;
@@ -73,6 +72,7 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import static org.apache.flink.runtime.io.network.buffer.BufferBuilderTestUtils.createFilledFinishedBufferConsumer;
import static org.apache.flink.runtime.io.network.partition.InputChannelTestUtils.createLocalInputChannel;
import static org.apache.flink.runtime.io.network.partition.InputChannelTestUtils.createSingleInputGate;
import static org.apache.flink.runtime.io.network.partition.InputGateFairnessTest.setupInputGate;
@@ -80,6 +80,7 @@ import static org.apache.flink.runtime.io.network.partition.consumer.SingleInput
import static org.apache.flink.runtime.state.CheckpointStorageLocationReference.getDefault;
import static org.apache.flink.util.Preconditions.checkArgument;
import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThat;
@@ -110,7 +111,7 @@ public class LocalInputChannelTest {
FreeingBufferRecycler.INSTANCE,
Buffer.DataType.EVENT_BUFFER),
memorySegment.size());
- BufferConsumer data = BufferBuilderTestUtils.createFilledFinishedBufferConsumer(1);
+ BufferConsumer data = createFilledFinishedBufferConsumer(1);
RecordingChannelStateWriter stateWriter = new RecordingChannelStateWriter();
LocalInputChannel channel =
@@ -546,7 +547,7 @@ public class LocalInputChannelTest {
assertTrue(channel.getNextBuffer().isPresent());
// Add more data
- subpartition.add(BufferBuilderTestUtils.createFilledFinishedBufferConsumer(4096));
+ subpartition.add(createFilledFinishedBufferConsumer(4096));
subpartition.flush();
// No buffer since the subpartition is blocked.
@@ -588,13 +589,13 @@ public class LocalInputChannelTest {
// add 1 buffer before barrier and 1 buffer afterwards. Only the first buffer should be
// written.
- subpartition.add(BufferBuilderTestUtils.createFilledFinishedBufferConsumer(1));
+ subpartition.add(createFilledFinishedBufferConsumer(1));
assertTrue(channel.getNextBuffer().isPresent());
subpartition.add(EventSerializer.toBufferConsumer(barrier, true));
assertTrue(channel.getNextBuffer().isPresent());
- subpartition.add(BufferBuilderTestUtils.createFilledFinishedBufferConsumer(2));
+ subpartition.add(createFilledFinishedBufferConsumer(2));
assertTrue(channel.getNextBuffer().isPresent());
assertArrayEquals(
@@ -604,13 +605,50 @@ public class LocalInputChannelTest {
new int[] {1});
}
+ @Test
+ public void testAnnounceNewBufferSize() throws IOException, InterruptedException {
+ // given: Configured LocalInputChannel and pipelined subpartition.
+ PipelinedResultPartition parent =
+ (PipelinedResultPartition)
+ new ResultPartitionBuilder()
+ .setResultPartitionType(ResultPartitionType.PIPELINED)
+ .setFileChannelManager(NoOpFileChannelManager.INSTANCE)
+ .setNumberOfSubpartitions(2)
+ .build();
+ ResultSubpartition subpartition0 = parent.getAllPartitions()[0];
+ ResultSubpartition subpartition1 = parent.getAllPartitions()[1];
+
+ LocalInputChannel channel0 =
+ createLocalInputChannel(
+ new SingleInputGateBuilder().build(),
+ new TestingResultPartitionManager(subpartition0.createReadView(() -> {})));
+ LocalInputChannel channel1 =
+ createLocalInputChannel(
+ new SingleInputGateBuilder().build(),
+ new TestingResultPartitionManager(subpartition1.createReadView(() -> {})));
+
+ channel0.requestSubpartition(0);
+ channel1.requestSubpartition(0);
+
+ // and: Preferable buffer size is default value.
+ assertEquals(Integer.MAX_VALUE, subpartition0.add(createFilledFinishedBufferConsumer(16)));
+ assertEquals(Integer.MAX_VALUE, subpartition1.add(createFilledFinishedBufferConsumer(16)));
+
+ // when: Announce the different buffer size for different channels via LocalInputChannel.
+ channel0.announceBufferSize(9);
+ channel1.announceBufferSize(20);
+
+ // then: The corresponded subpartitions have the new size.
+ assertEquals(9, subpartition0.add(createFilledFinishedBufferConsumer(16)));
+ assertEquals(20, subpartition1.add(createFilledFinishedBufferConsumer(16)));
+ }
+
// ---------------------------------------------------------------------------------------------
private static ResultSubpartitionView createResultSubpartitionView(boolean addBuffer)
throws IOException {
return addBuffer
- ? createResultSubpartitionView(
- BufferBuilderTestUtils.createFilledFinishedBufferConsumer(4096))
+ ? createResultSubpartitionView(createFilledFinishedBufferConsumer(4096))
: createResultSubpartitionView();
}
[flink] 05/09: [FLINK-23454][runtime] Subpartition is able to
notify the desirable buffer size for input buffer consumer.
Posted by pn...@apache.org.
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 703662aa61a8f82d4df6de7e0518a568822f71db
Author: Anton Kalashnikov <ka...@yandex.ru>
AuthorDate: Thu Aug 5 13:35:51 2021 +0200
[FLINK-23454][runtime] Subpartition is able to notify the desirable buffer size for input buffer consumer.
---
.../partition/BoundedBlockingSubpartition.java | 11 +++++--
.../network/partition/PipelinedSubpartition.java | 24 ++++++++++++---
.../io/network/partition/ResultSubpartition.java | 8 +++--
.../partition/PipelinedSubpartitionTest.java | 36 ++++++++++++++++++++++
.../io/network/partition/SubpartitionTestBase.java | 4 +--
5 files changed, 70 insertions(+), 13 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 275b563..b8bbd3c 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
@@ -120,15 +120,15 @@ final class BoundedBlockingSubpartition extends ResultSubpartition {
}
@Override
- public boolean add(BufferConsumer bufferConsumer, int partialRecordLength) throws IOException {
+ public int add(BufferConsumer bufferConsumer, int partialRecordLength) throws IOException {
if (isFinished()) {
bufferConsumer.close();
- return false;
+ return -1;
}
flushCurrentBuffer();
currentBuffer = bufferConsumer;
- return true;
+ return Integer.MAX_VALUE;
}
@Override
@@ -272,6 +272,11 @@ final class BoundedBlockingSubpartition extends ResultSubpartition {
}
@Override
+ public void bufferSize(int desirableNewBufferSize) {
+ // not supported.
+ }
+
+ @Override
protected long getTotalNumberOfBuffers() {
return numBuffersAndEventsWritten;
}
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 03315e5..8f9bfee 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
@@ -108,6 +108,8 @@ public class PipelinedSubpartition extends ResultSubpartition
/** Writes in-flight data. */
private ChannelStateWriter channelStateWriter;
+ private int bufferSize = Integer.MAX_VALUE;
+
/**
* Whether this subpartition is blocked (e.g. by exactly once checkpoint) and is waiting for
* resumption.
@@ -136,7 +138,7 @@ public class PipelinedSubpartition extends ResultSubpartition
}
@Override
- public boolean add(BufferConsumer bufferConsumer, int partialRecordLength) {
+ public int add(BufferConsumer bufferConsumer, int partialRecordLength) {
return add(bufferConsumer, partialRecordLength, false);
}
@@ -147,7 +149,7 @@ public class PipelinedSubpartition extends ResultSubpartition
bufferConsumer,
parent.getOwningTaskName(),
subpartitionInfo);
- if (!add(bufferConsumer, Integer.MIN_VALUE)) {
+ if (add(bufferConsumer, Integer.MIN_VALUE) == -1) {
throw new IOException("Buffer consumer couldn't be added to ResultSubpartition");
}
}
@@ -165,15 +167,16 @@ public class PipelinedSubpartition extends ResultSubpartition
LOG.debug("{}: Finished {}.", parent.getOwningTaskName(), this);
}
- private boolean add(BufferConsumer bufferConsumer, int partialRecordLength, boolean finish) {
+ private int add(BufferConsumer bufferConsumer, int partialRecordLength, boolean finish) {
checkNotNull(bufferConsumer);
final boolean notifyDataAvailable;
int prioritySequenceNumber = -1;
+ int newBufferSize;
synchronized (buffers) {
if (isFinished || isReleased) {
bufferConsumer.close();
- return false;
+ return -1;
}
// Add the bufferConsumer and update the stats
@@ -185,6 +188,7 @@ public class PipelinedSubpartition extends ResultSubpartition
notifyDataAvailable = finish || shouldNotifyDataAvailable();
isFinished |= finish;
+ newBufferSize = bufferSize;
}
if (prioritySequenceNumber != -1) {
@@ -194,7 +198,7 @@ public class PipelinedSubpartition extends ResultSubpartition
notifyDataAvailable();
}
- return true;
+ return newBufferSize;
}
private boolean addBuffer(BufferConsumer bufferConsumer, int partialRecordLength) {
@@ -450,6 +454,16 @@ public class PipelinedSubpartition extends ResultSubpartition
}
}
+ @Override
+ public void bufferSize(int desirableNewBufferSize) {
+ if (desirableNewBufferSize < 0) {
+ throw new IllegalArgumentException("New buffer size can not be less than zero");
+ }
+ synchronized (buffers) {
+ bufferSize = desirableNewBufferSize;
+ }
+ }
+
// ------------------------------------------------------------------------
@Override
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 37ef834..fd790c9 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
@@ -62,7 +62,7 @@ public abstract class ResultSubpartition {
}
@VisibleForTesting
- public final boolean add(BufferConsumer bufferConsumer) throws IOException {
+ public final int add(BufferConsumer bufferConsumer) throws IOException {
return add(bufferConsumer, 0);
}
@@ -80,10 +80,10 @@ 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 true if operation succeeded and bufferConsumer was enqueued for consumption.
+ * @return the preferable buffer size for this subpartition or -1 if the add operation fails.
* @throws IOException thrown in case of errors while adding the buffer
*/
- public abstract boolean add(BufferConsumer bufferConsumer, int partialRecordLength)
+ public abstract int add(BufferConsumer bufferConsumer, int partialRecordLength)
throws IOException;
public abstract void flush();
@@ -109,6 +109,8 @@ public abstract class ResultSubpartition {
/** Get the current size of the queue. */
public abstract int getNumberOfQueuedBuffers();
+ public abstract void bufferSize(int desirableNewBufferSize);
+
// ------------------------------------------------------------------------
/**
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartitionTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartitionTest.java
index 8ef0937..d662f59 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartitionTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartitionTest.java
@@ -300,6 +300,42 @@ public class PipelinedSubpartitionTest extends SubpartitionTestBase {
assertEquals(1, subpartition.getNumberOfQueuedBuffers());
}
+ @Test
+ public void testNewBufferSize() throws Exception {
+ // given: Buffer size equal to integer max value by default.
+ final PipelinedSubpartition subpartition = createSubpartition();
+ assertEquals(Integer.MAX_VALUE, subpartition.add(createFilledFinishedBufferConsumer(4)));
+
+ // when: Changing buffer size.
+ subpartition.bufferSize(42);
+
+ // then: Changes successfully applied.
+ assertEquals(42, subpartition.add(createFilledFinishedBufferConsumer(4)));
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testNegativeNewBufferSize() throws Exception {
+ // given: Buffer size equal to integer max value by default.
+ final PipelinedSubpartition subpartition = createSubpartition();
+ assertEquals(Integer.MAX_VALUE, subpartition.add(createFilledFinishedBufferConsumer(4)));
+
+ // when: Changing buffer size to the negative value.
+ subpartition.bufferSize(-1);
+ }
+
+ @Test
+ public void testNegativeBufferSizeAsSignOfAddingFail() throws Exception {
+ // given: Buffer size equal to integer max value by default.
+ final PipelinedSubpartition subpartition = createSubpartition();
+ assertEquals(Integer.MAX_VALUE, subpartition.add(createFilledFinishedBufferConsumer(4)));
+
+ // when: Finishing the subpartition which make following adding impossible.
+ subpartition.finish();
+
+ // then: -1 should be return because the add operation fails.
+ assertEquals(-1, subpartition.add(createFilledFinishedBufferConsumer(4)));
+ }
+
private void verifyViewReleasedAfterParentRelease(ResultSubpartition partition)
throws Exception {
// Add a bufferConsumer
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/SubpartitionTestBase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/SubpartitionTestBase.java
index 6a70670..a993935 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/SubpartitionTestBase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/SubpartitionTestBase.java
@@ -65,7 +65,7 @@ public abstract class SubpartitionTestBase extends TestLogger {
BufferConsumer bufferConsumer = createFilledFinishedBufferConsumer(4096);
- assertFalse(subpartition.add(bufferConsumer));
+ assertEquals(-1, subpartition.add(bufferConsumer));
assertTrue(bufferConsumer.isRecycled());
assertEquals(1, subpartition.getTotalNumberOfBuffers());
@@ -86,7 +86,7 @@ public abstract class SubpartitionTestBase extends TestLogger {
BufferConsumer bufferConsumer = createFilledFinishedBufferConsumer(4096);
- assertFalse(subpartition.add(bufferConsumer));
+ assertEquals(-1, subpartition.add(bufferConsumer));
assertTrue(bufferConsumer.isRecycled());
} finally {
[flink] 02/09: [refactor][runtime] Unification of obtaining reader
in PartitionRequestQueue
Posted by pn...@apache.org.
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 7478445aa30cd28b7b6402565ca304c916024475
Author: Anton Kalashnikov <ka...@yandex.ru>
AuthorDate: Thu Aug 5 13:27:36 2021 +0200
[refactor][runtime] Unification of obtaining reader in PartitionRequestQueue
---
.../io/network/netty/PartitionRequestQueue.java | 21 ++++++++++-----------
1 file changed, 10 insertions(+), 11 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java
index 72ede48..a656bea 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java
@@ -166,15 +166,10 @@ class PartitionRequestQueue extends ChannelInboundHandlerAdapter {
return;
}
- NetworkSequenceViewReader reader = allReaders.get(receiverId);
- if (reader != null) {
- operation.accept(reader);
+ NetworkSequenceViewReader reader = obtainReader(receiverId);
- enqueueAvailableReader(reader);
- } else {
- throw new IllegalStateException(
- "No reader for receiverId = " + receiverId + " exists.");
- }
+ operation.accept(reader);
+ enqueueAvailableReader(reader);
}
void acknowledgeAllRecordsProcessed(InputChannelID receiverId) {
@@ -182,13 +177,17 @@ class PartitionRequestQueue extends ChannelInboundHandlerAdapter {
return;
}
+ obtainReader(receiverId).acknowledgeAllRecordsProcessed();
+ }
+
+ NetworkSequenceViewReader obtainReader(InputChannelID receiverId) {
NetworkSequenceViewReader reader = allReaders.get(receiverId);
- if (reader != null) {
- reader.acknowledgeAllRecordsProcessed();
- } else {
+ if (reader == null) {
throw new IllegalStateException(
"No reader for receiverId = " + receiverId + " exists.");
}
+
+ return reader;
}
/**
[flink] 09/09: [FLINK-23726][runtime] Read buffer debloat
configuration from task manager configuration rather than task
configuration.
Posted by pn...@apache.org.
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 3e9e5309db2740da0415afa56025d1d177374bd6
Author: Anton Kalashnikov <ka...@yandex.ru>
AuthorDate: Mon Aug 9 20:21:24 2021 +0200
[FLINK-23726][runtime] Read buffer debloat configuration from task manager configuration rather than task configuration.
---
.../flink/streaming/runtime/tasks/StreamTask.java | 10 ++++++----
.../tasks/StreamTaskMailboxTestHarnessBuilder.java | 11 +++++++++++
.../flink/streaming/runtime/tasks/StreamTaskTest.java | 18 ++++++++++--------
3 files changed, 27 insertions(+), 12 deletions(-)
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 47c9dff..160722b 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
@@ -20,6 +20,7 @@ package org.apache.flink.streaming.runtime.tasks;
import org.apache.flink.annotation.Internal;
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.operators.MailboxExecutor;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.core.fs.CloseableRegistry;
import org.apache.flink.core.fs.Path;
@@ -406,12 +407,13 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>> extends Ab
environment.getMetricGroup().getIOMetricGroup().setEnableBusyTime(true);
this.throughputCalculator = environment.getThroughputCalculator();
- this.bufferDebloatPeriod = getTaskConfiguration().get(BUFFER_DEBLOAT_PERIOD).toMillis();
+ Configuration taskManagerConf = environment.getTaskManagerInfo().getConfiguration();
- if (getTaskConfiguration().get(TaskManagerOptions.BUFFER_DEBLOAT_ENABLED)) {
+ this.bufferDebloatPeriod = taskManagerConf.get(BUFFER_DEBLOAT_PERIOD).toMillis();
+
+ if (taskManagerConf.get(TaskManagerOptions.BUFFER_DEBLOAT_ENABLED)) {
this.bufferDebloater =
- new BufferDebloater(
- getTaskConfiguration(), getEnvironment().getAllInputGates());
+ new BufferDebloater(taskManagerConf, getEnvironment().getAllInputGates());
environment
.getMetricGroup()
.gauge(
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskMailboxTestHarnessBuilder.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskMailboxTestHarnessBuilder.java
index c92bb9c..721c776 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskMailboxTestHarnessBuilder.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskMailboxTestHarnessBuilder.java
@@ -38,8 +38,10 @@ import org.apache.flink.runtime.state.TestLocalRecoveryConfig;
import org.apache.flink.runtime.state.TestTaskStateManager;
import org.apache.flink.runtime.state.TestTaskStateManagerBuilder;
import org.apache.flink.runtime.taskmanager.CheckpointResponder;
+import org.apache.flink.runtime.taskmanager.TaskManagerRuntimeInfo;
import org.apache.flink.runtime.taskmanager.TestCheckpointResponder;
import org.apache.flink.runtime.throughput.ThroughputCalculator;
+import org.apache.flink.runtime.util.TestingTaskManagerRuntimeInfo;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.graph.StreamConfig;
import org.apache.flink.streaming.api.graph.StreamConfig.InputConfig;
@@ -100,6 +102,8 @@ public class StreamTaskMailboxTestHarnessBuilder<OUT> {
private ThroughputCalculator throughputCalculator =
new ThroughputCalculator(SystemClock.getInstance(), 10);
+ private TaskManagerRuntimeInfo taskManagerRuntimeInfo = new TestingTaskManagerRuntimeInfo();
+
public StreamTaskMailboxTestHarnessBuilder(
FunctionWithException<Environment, ? extends StreamTask<OUT, ?>, Exception> taskFactory,
TypeInformation<OUT> outputType) {
@@ -132,6 +136,12 @@ public class StreamTaskMailboxTestHarnessBuilder<OUT> {
return this;
}
+ public <T> StreamTaskMailboxTestHarnessBuilder<OUT> setTaskManagerRuntimeInfo(
+ TaskManagerRuntimeInfo taskManagerRuntimeInfo) {
+ this.taskManagerRuntimeInfo = taskManagerRuntimeInfo;
+ return this;
+ }
+
public StreamTaskMailboxTestHarnessBuilder<OUT> addInput(TypeInformation<?> inputType) {
return addInput(inputType, 1);
}
@@ -217,6 +227,7 @@ public class StreamTaskMailboxTestHarnessBuilder<OUT> {
streamMockEnvironment.setCheckpointResponder(taskStateManager.getCheckpointResponder());
initializeInputs(streamMockEnvironment);
+ streamMockEnvironment.setTaskManagerInfo(taskManagerRuntimeInfo);
checkState(inputGates != null, "InputGates hasn't been initialised");
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 d5e149b..a9731e4 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
@@ -97,6 +97,7 @@ import org.apache.flink.runtime.taskmanager.TaskManagerActions;
import org.apache.flink.runtime.taskmanager.TestTaskBuilder;
import org.apache.flink.runtime.throughput.ThroughputCalculator;
import org.apache.flink.runtime.util.NettyShuffleDescriptorBuilder;
+import org.apache.flink.runtime.util.TestingTaskManagerRuntimeInfo;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions;
@@ -1846,20 +1847,21 @@ public class StreamTaskTest extends TestLogger {
public void testBufferSizeRecalculationStartSuccessfully() throws Exception {
int expectedThroughput = 13333;
int inputChannels = 3;
- Consumer<StreamConfig> configuration =
- (config) -> {
- // debloat period doesn't matter, we will schedule debloating manually
- config.getConfiguration().set(BUFFER_DEBLOAT_PERIOD, Duration.ofHours(10));
- config.getConfiguration().set(BUFFER_DEBLOAT_TARGET, Duration.ofSeconds(1));
- config.getConfiguration().set(BUFFER_DEBLOAT_ENABLED, true);
- };
+
+ // debloat period doesn't matter, we will schedule debloating manually
+ Configuration config =
+ new Configuration()
+ .set(BUFFER_DEBLOAT_PERIOD, Duration.ofHours(10))
+ .set(BUFFER_DEBLOAT_TARGET, Duration.ofSeconds(1))
+ .set(BUFFER_DEBLOAT_ENABLED, true);
+
Map<String, Metric> metrics = new ConcurrentHashMap<>();
final TaskMetricGroup taskMetricGroup =
StreamTaskTestHarness.createTaskMetricGroup(metrics);
try (StreamTaskMailboxTestHarness<String> harness =
new StreamTaskMailboxTestHarnessBuilder<>(OneInputStreamTask::new, STRING_TYPE_INFO)
- .modifyStreamConfig(configuration)
+ .setTaskManagerRuntimeInfo(new TestingTaskManagerRuntimeInfo(config))
.setTaskMetricGroup(taskMetricGroup)
.addInput(STRING_TYPE_INFO, inputChannels)
.setupOutputForSingletonOperatorChain(
[flink] 03/09: [FLINK-23454][runtime] NewBufferSize migrated from
long to int for bufferSize because the bufferSize is always int.
Posted by pn...@apache.org.
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 77f7b7f0f90725fcc65144fa86572d16314f3d0a
Author: Anton Kalashnikov <ka...@yandex.ru>
AuthorDate: Thu Aug 5 13:19:10 2021 +0200
[FLINK-23454][runtime] NewBufferSize migrated from long to int for bufferSize because the bufferSize is always int.
---
.../apache/flink/runtime/io/network/netty/NettyMessage.java | 10 +++++-----
.../netty/CreditBasedPartitionRequestClientHandlerTest.java | 4 ++--
.../network/netty/NettyMessageServerSideSerializationTest.java | 2 +-
3 files changed, 8 insertions(+), 8 deletions(-)
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 3f62400..8198456 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
@@ -849,11 +849,11 @@ public abstract class NettyMessage {
private static final byte ID = 10;
- final long bufferSize;
+ final int bufferSize;
final InputChannelID receiverId;
- NewBufferSize(long bufferSize, InputChannelID receiverId) {
+ NewBufferSize(int bufferSize, InputChannelID receiverId) {
checkArgument(bufferSize > 0, "The new buffer size should be greater than 0");
this.bufferSize = bufferSize;
this.receiverId = receiverId;
@@ -867,8 +867,8 @@ public abstract class NettyMessage {
try {
result =
allocateBuffer(
- allocator, ID, Long.BYTES + InputChannelID.getByteBufLength());
- result.writeLong(bufferSize);
+ allocator, ID, Integer.BYTES + InputChannelID.getByteBufLength());
+ result.writeInt(bufferSize);
receiverId.writeTo(result);
out.write(result, promise);
@@ -878,7 +878,7 @@ public abstract class NettyMessage {
}
static NewBufferSize readFrom(ByteBuf buffer) {
- long bufferSize = buffer.readLong();
+ int bufferSize = buffer.readInt();
InputChannelID receiverId = InputChannelID.fromByteBuf(buffer);
return new NewBufferSize(bufferSize, receiverId);
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/CreditBasedPartitionRequestClientHandlerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/CreditBasedPartitionRequestClientHandlerTest.java
index fc49356..6e8cab9 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/CreditBasedPartitionRequestClientHandlerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/CreditBasedPartitionRequestClientHandlerTest.java
@@ -711,11 +711,11 @@ public class CreditBasedPartitionRequestClientHandlerTest {
NettyMessage.NewBufferSize readOutbound = channel.readOutbound();
assertThat(readOutbound, instanceOf(NettyMessage.NewBufferSize.class));
assertThat(readOutbound.receiverId, is(inputChannels[0].getInputChannelId()));
- assertThat(readOutbound.bufferSize, is(333L));
+ assertThat(readOutbound.bufferSize, is(333));
readOutbound = channel.readOutbound();
assertThat(readOutbound.receiverId, is(inputChannels[1].getInputChannelId()));
- assertThat(readOutbound.bufferSize, is(333L));
+ assertThat(readOutbound.bufferSize, is(333));
} finally {
releaseResource(inputGate, networkBufferPool);
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 66714de..3aedfa3 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
@@ -140,7 +140,7 @@ public class NettyMessageServerSideSerializationTest extends TestLogger {
public void testNewBufferSize() {
NettyMessage.NewBufferSize expected =
new NettyMessage.NewBufferSize(
- random.nextInt(Integer.MAX_VALUE) + 1L, new InputChannelID());
+ random.nextInt(Integer.MAX_VALUE), new InputChannelID());
NettyMessage.NewBufferSize actual = encodeAndDecode(expected, channel);
assertEquals(expected.bufferSize, actual.bufferSize);
[flink] 08/09: [FLINK-23454][runtime] ResultPartition changes the
BufferBuilder size according to the desirable buffer size of each
subpartition.
Posted by pn...@apache.org.
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 545b6126cc42c5c9bce8783931b04a3bec571aa6
Author: Anton Kalashnikov <ka...@yandex.ru>
AuthorDate: Thu Aug 5 13:47:50 2021 +0200
[FLINK-23454][runtime] ResultPartition changes the BufferBuilder size according to the desirable buffer size of each subpartition.
---
.../partition/BufferWritingResultPartition.java | 18 +++++-
.../partition/PipelinedSubpartitionTest.java | 4 ++
.../io/network/partition/ResultPartitionTest.java | 70 ++++++++++++++++++++++
3 files changed, 89 insertions(+), 3 deletions(-)
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 abb7e1d..b4f2a1e 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
@@ -244,7 +244,7 @@ public abstract class BufferWritingResultPartition extends ResultPartition {
if (buffer == null) {
buffer = requestNewUnicastBufferBuilder(targetSubpartition);
- subpartitions[targetSubpartition].add(buffer.createBufferConsumerFromBeginning(), 0);
+ addToSubpartition(buffer, targetSubpartition, 0);
}
buffer.appendAndCommit(record);
@@ -252,6 +252,19 @@ public abstract class BufferWritingResultPartition extends ResultPartition {
return buffer;
}
+ private void addToSubpartition(BufferBuilder buffer, int targetSubpartition, int i)
+ throws IOException {
+ int desirableBufferSize =
+ subpartitions[targetSubpartition].add(
+ buffer.createBufferConsumerFromBeginning(), i);
+
+ if (desirableBufferSize > 0) {
+ // !! If some of partial data has written already to this buffer, the result size can
+ // not be less than written value.
+ buffer.trim(desirableBufferSize);
+ }
+ }
+
private BufferBuilder appendUnicastDataForRecordContinuation(
final ByteBuffer remainingRecordBytes, final int targetSubpartition)
throws IOException {
@@ -263,8 +276,7 @@ public abstract class BufferWritingResultPartition extends ResultPartition {
// with a complete record.
// !! The next two lines can not change order.
final int partialRecordBytes = buffer.appendAndCommit(remainingRecordBytes);
- subpartitions[targetSubpartition].add(
- buffer.createBufferConsumerFromBeginning(), partialRecordBytes);
+ addToSubpartition(buffer, targetSubpartition, partialRecordBytes);
return buffer;
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartitionTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartitionTest.java
index d662f59..59cd1a6 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartitionTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartitionTest.java
@@ -365,4 +365,8 @@ public class PipelinedSubpartitionTest extends SubpartitionTestBase {
return new PipelinedSubpartition(0, 2, parent);
}
+
+ public static PipelinedSubpartition createPipelinedSubpartition(ResultPartition parent) {
+ return new PipelinedSubpartition(0, 2, parent);
+ }
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionTest.java
index 6905f62..b299edc 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionTest.java
@@ -664,6 +664,76 @@ public class ResultPartitionTest {
}
}
+ @Test
+ public void testDifferentBufferSizeForSubpartitions() throws IOException {
+ // given: Configured pipelined result with 2 subpartitions.
+ BufferWritingResultPartition bufferWritingResultPartition =
+ createResultPartition(ResultPartitionType.PIPELINED_BOUNDED);
+
+ ResultSubpartition[] subpartitions = bufferWritingResultPartition.subpartitions;
+ assertEquals(2, subpartitions.length);
+
+ PipelinedSubpartition subpartition0 = (PipelinedSubpartition) subpartitions[0];
+ PipelinedSubpartition subpartition1 = (PipelinedSubpartition) subpartitions[1];
+
+ // when: Set the different buffers size.
+ subpartition0.bufferSize(10);
+ subpartition1.bufferSize(6);
+
+ // and: Add the buffer.
+ bufferWritingResultPartition.emitRecord(ByteBuffer.allocate(12), 0);
+ bufferWritingResultPartition.emitRecord(ByteBuffer.allocate(12), 1);
+
+ // then: The buffer less or equal to configured.
+ assertEquals(10, subpartition0.pollBuffer().buffer().getSize());
+ assertEquals(2, subpartition0.pollBuffer().buffer().getSize());
+ assertEquals(6, subpartition1.pollBuffer().buffer().getSize());
+ assertEquals(6, subpartition1.pollBuffer().buffer().getSize());
+
+ // when: Reset the buffer size.
+ subpartition0.bufferSize(13);
+ subpartition1.bufferSize(5);
+
+ // and: Add the buffer.
+ bufferWritingResultPartition.emitRecord(ByteBuffer.allocate(20), 0);
+ bufferWritingResultPartition.emitRecord(ByteBuffer.allocate(9), 1);
+
+ // then: The buffer less or equal to configured.
+ // 8 bytes which fitted to the previous unfinished buffer(10 - 2).
+ assertEquals(8, subpartition0.pollBuffer().buffer().getSize());
+ // 12 rest bytes which fitted to a new buffer which has 13 bytes.
+ assertEquals(12, subpartition0.pollBuffer().buffer().getSize());
+ assertEquals(5, subpartition1.pollBuffer().buffer().getSize());
+ assertEquals(4, subpartition1.pollBuffer().buffer().getSize());
+ }
+
+ @Test
+ public void testBufferSizeNotChanged() throws IOException {
+ // given: Configured pipelined result with 2 subpartitions.
+ BufferWritingResultPartition bufferWritingResultPartition =
+ createResultPartition(ResultPartitionType.PIPELINED_BOUNDED);
+
+ ResultSubpartition[] subpartitions = bufferWritingResultPartition.subpartitions;
+ assertEquals(2, subpartitions.length);
+
+ PipelinedSubpartition subpartition0 = (PipelinedSubpartition) subpartitions[0];
+ PipelinedSubpartition subpartition1 = (PipelinedSubpartition) subpartitions[1];
+
+ // when: Set the different buffers size.
+ subpartition0.bufferSize(bufferSize + 1);
+ subpartition1.bufferSize(Integer.MAX_VALUE);
+
+ // and: Add the buffer.
+ bufferWritingResultPartition.emitRecord(ByteBuffer.allocate(bufferSize), 0);
+ bufferWritingResultPartition.emitRecord(ByteBuffer.allocate(bufferSize), 1);
+
+ // then: The buffer has initial size because new buffer was greater than max.
+ assertEquals(bufferSize, subpartition0.pollBuffer().buffer().getSize());
+
+ // and: The buffer has initial size because new buffer was less than 0.
+ assertEquals(bufferSize, subpartition1.pollBuffer().buffer().getSize());
+ }
+
private static class TestResultPartitionConsumableNotifier
implements ResultPartitionConsumableNotifier {
private JobID jobID;