[FLINK-23457][network] The upstream sends the buffer of the right size for broadcast case #17024

lometheus · 2021-08-27T11:51:12Z

What is the purpose of the change

This PR fit NewBufferSize in broadcast case

Brief change log

Support create broadcast buffer consumers with desirable size

Verifying this change

Added unit tests for each classes which behaviour was changed

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): ( no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (no)
The serializers: (no)
The runtime per-record code paths (performance sensitive): (no)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (no)
The S3 file system connector: (no)

Documentation

Does this pull request introduce a new feature? (no)
If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)

lometheus · 2021-08-27T12:01:22Z

cc @pnowojski

flinkbot · 2021-08-27T12:05:37Z

Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.

Automated Checks

Last check on commit e2ca022 (Fri Aug 27 12:05:36 UTC 2021)

Warnings:

No documentation files were touched! Remember to keep the Flink docs up to date!
This pull request references an unassigned Jira ticket. According to the code contribution guide, tickets need to be assigned before starting with the implementation work.

_{Mention the bot in a comment to re-run the automated checks.}

Review Progress

❓ 1. The [description] looks good.
❓ 2. There is [consensus] that the contribution should go into to Flink.
❓ 3. Needs [attention] from.
❓ 4. The change fits into the overall [architecture].
❓ 5. Overall code [quality] is good.

Please see the Pull Request Review Guide for a full explanation of the review process.

The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot approve description to approve one or more aspects (aspects: description, consensus, architecture and quality)
@flinkbot approve all to approve all aspects
@flinkbot approve-until architecture to approve everything until architecture
@flinkbot attention @username1 [@username2 ..] to require somebody's attention
@flinkbot disapprove architecture to remove an approval you gave earlier

flinkbot · 2021-08-27T13:07:44Z

CI report:

524997b Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run travis re-run the last Travis build
@flinkbot run azure re-run the last Azure build

lometheus · 2021-08-29T13:09:46Z

@flinkbot run azure

lometheus · 2021-08-29T13:24:30Z

I've updated, please have a look again.thx @pnowojski

akalash

@lometheus, thanks for your changes. I left a couple of comments in the PR. Also, I remind you that the correct format for the commit comment is the same as you use for the PR name([FLINK-task][component] Comment). So don't forget to update it.

akalash · 2021-08-30T17:02:41Z

...rc/main/java/org/apache/flink/runtime/io/network/partition/BufferWritingResultPartition.java

            for (ResultSubpartition subpartition : subpartitions) {
-                subpartition.add(consumer.copy(), partialRecordBytes);
+                int subPartitionBufferSize = subpartition.add(consumer.copy(), partialRecordBytes);


If subpartition#add fails it returns the negative value(-1), so we should take into account this case and for example ignore such value.

Thank you very much for review. I add special treatment for negative value

I believe more correctly is to ignore the value for calculation of desirable buffer size rather than ignore the total value. I mean:

if(subPartitionBufferSize > 0) { desirableBufferSize = Math.min(desirableBufferSize, subPartitionBufferSize); }

I think that it is better because if one of the subpartitions fails we are still able to send it to another ones. As I understand eventually they all will be closed if at least one was closed so maybe it is not so important but anyway it is better to follow the current semantic which doesn't forbid sending data to subpartitions even if one of them is closed.

akalash · 2021-08-30T17:11:46Z

...runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionTest.java

+
+        // then: The buffer less or equal to configured.
+        assertEquals(5, subpartition0.pollBuffer().buffer().getSize());
+        assertEquals(5, subpartition1.pollBuffer().buffer().getSize());


Please, take a look more carefully at this test(testDifferentBufferSizeForSubpartitions), there are several scenarios that were tested here(send buffer less/greater/equal than the buffer size, change the buffer size and send the again). But you have only one test case which is not enough. So please, increase your test coverage.
Also I suggest extracting your test in the separated method like testDynamicBufferSizeForBroadcast or something similar.

I add testDynamicBufferSizeForBroadcast for less/greater/equal case

akalash

@lometheus , I left one more comment, you can check it and then you can squash your commits to one for the final review and test.

akalash · 2021-08-31T17:14:55Z

...rc/main/java/org/apache/flink/runtime/io/network/partition/BufferWritingResultPartition.java

            for (ResultSubpartition subpartition : subpartitions) {
-                subpartition.add(consumer.copy(), partialRecordBytes);
+                int subPartitionBufferSize = subpartition.add(consumer.copy(), partialRecordBytes);


I believe more correctly is to ignore the value for calculation of desirable buffer size rather than ignore the total value. I mean:

if(subPartitionBufferSize > 0) { desirableBufferSize = Math.min(desirableBufferSize, subPartitionBufferSize); }

I think that it is better because if one of the subpartitions fails we are still able to send it to another ones. As I understand eventually they all will be closed if at least one was closed so maybe it is not so important but anyway it is better to follow the current semantic which doesn't forbid sending data to subpartitions even if one of them is closed.

lometheus · 2021-09-01T01:43:47Z

@lometheus , I left one more comment, you can check it and then you can squash your commits to one for the final review and test.

Wonderful idea ,I just fixed it with select a desirable buffer size.
exp: A question when I use
PipelinedSubpartition.bufferSize($size)
reset subpartition buffers size ,It seems only first buffer affected by $size , is this normal?

akalash · 2021-09-01T09:06:31Z

A question when I use PipelinedSubpartition.bufferSize($size) reset subpartition buffers size ,It seems only first buffer affected by $size , is this normal?

If I understood your question right then if one record allocates more than one buffer then the first buffer would be trimmed but the second one will be equal to the size of the rest of the record. In an ideal world, we should trim the second buffer too but in reality, it requires serious changes in the code which actually doesn't make any sense because it is a big mistake to configure buffer size less than one record.
So the answer is yes, it trims only the first buffer for one record but it will trim for each next record too. You can take a look at the comment inside of BufferWritingResultPartition#addToSubpartition.

P.S. Please, squash your commits into one(and you can rebase onto the fresh master).

akalash · 2021-09-01T09:09:57Z

...rc/main/java/org/apache/flink/runtime/io/network/partition/BufferWritingResultPartition.java

+                                ? Math.min(desirableBufferSize, subPartitionBufferSize)
+                                : desirableBufferSize;
+            }
+            if (desirableBufferSize != Integer.MAX_VALUE) {


Just a minor comment: it is up to you but this condition is not really needed here because buffer.trim(MAX_VALUE) is not a problem it just set the buffer size to the maximum possible value which is ok. But it is not a mistake if you left this condition here.

You are correct, the code should be as concise as possible,I drop this condition.

lometheus · 2021-09-02T00:53:37Z

I squash all my commits into one , thinks for your review and patiently answer again

P.S. Please, squash your commits into one(and you can rebase onto the fresh master).

akalash

LGTM. @pnowojski, can you help with the merge, please?

pnowojski

LGTM, there is an unrelated test failure that has been fixed FLINK-24036.

pnowojski · 2021-09-02T09:45:52Z

@flinkbot run azure

pnowojski · 2021-09-02T09:48:46Z

@lometheus , could you rebase this PR on top of the latest master to pull in the fix for this failure so that we could have a green build?

lometheus · 2021-09-02T11:56:32Z

restored dev_20210827

lometheus · 2021-09-02T11:57:50Z

@flinkbot run azure

lometheus · 2021-09-03T00:48:06Z

@flinkbot run azure

lometheus · 2021-09-03T01:15:01Z

@flinkbot run azure

lometheus · 2021-09-03T01:48:22Z

@pnowojski flinknot seems doesn't work anymore, how can I deal with this condition

pnowojski · 2021-09-03T07:03:53Z

It takes some time for the flinkbot to pickup PR. It looks like it did just that and the most recent version of this PR seems to have some failures?

bfeab81 Azure: FAILURE

pnowojski · 2021-09-03T07:08:54Z

...rc/main/java/org/apache/flink/runtime/io/network/partition/BufferWritingResultPartition.java

@@ -330,9 +330,16 @@ private BufferBuilder appendBroadcastDataForRecordContinuation(
    private void createBroadcastBufferConsumers(BufferBuilder buffer, int partialRecordBytes)
            throws IOException {
        try (final BufferConsumer consumer = buffer.createBufferConsumerFromBeginning()) {
+            int desirableBufferSize = Integer.MAX_VALUE;
            for (ResultSubpartition subpartition : subpartitions) {
                subpartition.add(consumer.copy(), partialRecordBytes);


You have duplicated:

subpartition.add(consumer.copy(), partialRecordBytes);

which is causing Azure failures.

(edit: I have already fixed it)

...rc/main/java/org/apache/flink/runtime/io/network/partition/BufferWritingResultPartition.java

…k/partition/BufferWritingResultPartition.java

pnowojski · 2021-09-07T08:25:43Z

Azure was green, merged to master manually after squashing the commits. Thanks @lometheus for your contribution :)

lometheus changed the title ~~FLINK-23973 The upstream sends the buffer of the right size for broadcast case~~ [FLINK-23973] The upstream sends the buffer of the right size for broadcast case Aug 27, 2021

rmetzger added the review=description? label Aug 27, 2021

lometheus changed the title ~~[FLINK-23973] The upstream sends the buffer of the right size for broadcast case~~ [FLINK-23973](https://issues.apache.org/jira/browse/FLINK-23457). The upstream sends the buffer of the right size for broadcast case Aug 27, 2021

lometheus changed the title ~~[FLINK-23973](https://issues.apache.org/jira/browse/FLINK-23457) The upstream sends the buffer of the right size for broadcast case~~ [FLINK-23973] The upstream sends the buffer of the right size for broadcast case Aug 27, 2021

pnowojski changed the title ~~[FLINK-23973] The upstream sends the buffer of the right size for broadcast case~~ [FLINK-23457] The upstream sends the buffer of the right size for broadcast case Aug 27, 2021

pnowojski changed the title ~~[FLINK-23457] The upstream sends the buffer of the right size for broadcast case~~ [FLINK-23457][network] The upstream sends the buffer of the right size for broadcast case Aug 27, 2021

akalash reviewed Aug 30, 2021

View reviewed changes

lometheus force-pushed the dev_20210827 branch from d94a61c to e83ed6a Compare August 31, 2021 13:31

akalash reviewed Aug 31, 2021

View reviewed changes

akalash reviewed Sep 1, 2021

View reviewed changes

lometheus force-pushed the dev_20210827 branch from fcfa7e6 to a941706 Compare September 2, 2021 00:46

akalash approved these changes Sep 2, 2021

View reviewed changes

pnowojski approved these changes Sep 2, 2021

View reviewed changes

lometheus closed this Sep 2, 2021

lometheus force-pushed the dev_20210827 branch from bbdb8d5 to 71836bf Compare September 2, 2021 11:38

lometheus deleted the dev_20210827 branch September 2, 2021 11:38

lometheus restored the dev_20210827 branch September 2, 2021 11:38

lometheus deleted the dev_20210827 branch September 2, 2021 11:39

lometheus restored the dev_20210827 branch September 2, 2021 11:40

lometheus reopened this Sep 2, 2021

[FLINK-23457][network] Select a desirable buffer size

bfeab81

lometheus force-pushed the dev_20210827 branch from d7e21cf to bfeab81 Compare September 3, 2021 01:14

pnowojski reviewed Sep 3, 2021

View reviewed changes

...rc/main/java/org/apache/flink/runtime/io/network/partition/BufferWritingResultPartition.java Outdated Show resolved Hide resolved

Update flink-runtime/src/main/java/org/apache/flink/runtime/io/networ…

524997b

…k/partition/BufferWritingResultPartition.java

pnowojski closed this Sep 7, 2021

flinkbot added the component=<none> label Mar 17, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-23457][network] The upstream sends the buffer of the right size for broadcast case #17024

[FLINK-23457][network] The upstream sends the buffer of the right size for broadcast case #17024

lometheus commented Aug 27, 2021

lometheus commented Aug 27, 2021

flinkbot commented Aug 27, 2021

flinkbot commented Aug 27, 2021 •

edited

lometheus commented Aug 29, 2021

lometheus commented Aug 29, 2021

akalash left a comment

akalash Aug 30, 2021

lometheus Aug 31, 2021

akalash Aug 31, 2021

akalash Aug 30, 2021

lometheus Aug 31, 2021

akalash left a comment

akalash Aug 31, 2021

lometheus commented Sep 1, 2021

akalash commented Sep 1, 2021

akalash Sep 1, 2021

lometheus Sep 2, 2021

lometheus commented Sep 2, 2021

akalash left a comment

pnowojski left a comment

pnowojski commented Sep 2, 2021

pnowojski commented Sep 2, 2021

lometheus commented Sep 2, 2021

lometheus commented Sep 2, 2021

lometheus commented Sep 3, 2021

lometheus commented Sep 3, 2021

lometheus commented Sep 3, 2021

pnowojski commented Sep 3, 2021

pnowojski Sep 3, 2021 •

edited

pnowojski commented Sep 7, 2021

[FLINK-23457][network] The upstream sends the buffer of the right size for broadcast case #17024

[FLINK-23457][network] The upstream sends the buffer of the right size for broadcast case #17024

Conversation

lometheus commented Aug 27, 2021

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

lometheus commented Aug 27, 2021

flinkbot commented Aug 27, 2021

Automated Checks

Review Progress

flinkbot commented Aug 27, 2021 • edited

CI report:

lometheus commented Aug 29, 2021

lometheus commented Aug 29, 2021

akalash left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

akalash left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lometheus commented Sep 1, 2021

akalash commented Sep 1, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lometheus commented Sep 2, 2021

akalash left a comment

Choose a reason for hiding this comment

pnowojski left a comment

Choose a reason for hiding this comment

pnowojski commented Sep 2, 2021

pnowojski commented Sep 2, 2021

lometheus commented Sep 2, 2021

lometheus commented Sep 2, 2021

lometheus commented Sep 3, 2021

lometheus commented Sep 3, 2021

lometheus commented Sep 3, 2021

pnowojski commented Sep 3, 2021

pnowojski Sep 3, 2021 • edited

Choose a reason for hiding this comment

pnowojski commented Sep 7, 2021

flinkbot commented Aug 27, 2021 •

edited

pnowojski Sep 3, 2021 •

edited