Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[FLINK-33743][runtime] Support consuming multiple subpartitions on a single channel #23927

Merged
merged 14 commits into from
Jan 17, 2024
Merged

[FLINK-33743][runtime] Support consuming multiple subpartitions on a single channel #23927

merged 14 commits into from
Jan 17, 2024

Conversation

yunfengzhou-hub
Copy link
Contributor

What is the purpose of the change

This pull request adds support for consuming multiple subpartitions in a single input channel.

Brief change log

  • Separate the notion of subpartition and channel in terms of naming, signature and comments.
  • Union the output from multiple subpartitions with one ResultSubpartitionView and TierConsumerAgent
  • Control the partial record split logic when writing buffers into subpartition to avoid potential deadlocks

Verifying this change

This change added tests and can be verified as follows:

  • Extended HybridShuffleITCase to cover cases when the feature proposed in this PR comes into effect.
  • Added unit tests for newly introduced options and classes
  • Manually verified situations that are not covered by e2e tests, like when tiered hybrid shuffle uses sort buffer.

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: yes
  • The runtime per-record code paths (performance sensitive): yes
  • 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? yes
  • If yes, how is the feature documented? no need to document

@flinkbot
Copy link
Collaborator

flinkbot commented Dec 14, 2023 •
edited

CI report:

Bot commands The @flinkbot bot supports the following commands:
  • @flinkbot run azure re-run the last Azure build

@yunfengzhou-hub yunfengzhou-hub force-pushed the input-channel-memory branch 7 times, most recently from 35eb891 to b32f79c Compare December 19, 2023 01:37
@yunfengzhou-hub
Copy link
Contributor Author

Hi @TanYuxin-tyx could you please take a look at this PR?

TanYuxin-tyx
TanYuxin-tyx reviewed Dec 20, 2023
View reviewed changes
Copy link
Contributor

@TanYuxin-tyx TanYuxin-tyx left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@yunfengzhou-hub Thanks for the contribution. I have some comments on the change, PTAL.

.../src/main/java/org/apache/flink/runtime/io/network/partition/ResultSubpartitionIndexSet.java Outdated Show resolved Hide resolved
.../src/main/java/org/apache/flink/runtime/io/network/partition/ResultSubpartitionIndexSet.java Outdated Show resolved Hide resolved
...me/src/main/java/org/apache/flink/runtime/io/network/partition/SortMergeResultPartition.java Show resolved Hide resolved
...time/src/test/java/org/apache/flink/runtime/io/network/netty/CancelPartitionRequestTest.java Show resolved Hide resolved
...rc/main/java/org/apache/flink/runtime/io/network/partition/ResultSubpartitionIndexRange.java Outdated Show resolved Hide resolved
...test/java/org/apache/flink/runtime/io/network/partition/UnionResultSubpartitionViewTest.java Outdated Show resolved Hide resolved
...test/java/org/apache/flink/runtime/io/network/partition/UnionResultSubpartitionViewTest.java Outdated Show resolved Hide resolved
...test/java/org/apache/flink/runtime/io/network/partition/UnionResultSubpartitionViewTest.java Show resolved Hide resolved
...rc/main/java/org/apache/flink/runtime/io/network/api/writer/ChannelSelectorRecordWriter.java Show resolved Hide resolved
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/Buffer.java Outdated Show resolved Hide resolved
@yunfengzhou-hub yunfengzhou-hub force-pushed the input-channel-memory branch 2 times, most recently from 1722add to 2cef883 Compare December 21, 2023 09:19
@TanYuxin-tyx
Copy link
Contributor

@yunfengzhou-hub Thanks for the update. I have no more comments on the change.

xintongsong
xintongsong reviewed Dec 26, 2023
View reviewed changes
.../src/main/java/org/apache/flink/runtime/io/network/partition/ResultSubpartitionIndexSet.java Outdated Show resolved Hide resolved
...rc/main/java/org/apache/flink/runtime/io/network/partition/ResultSubpartitionIndexRange.java Outdated Show resolved Hide resolved
...me/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java Show resolved Hide resolved
...org/apache/flink/runtime/io/network/partition/hybrid/tiered/netty/NettyConnectionReader.java Outdated Show resolved Hide resolved
...he/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageConsumerClient.java Outdated Show resolved Hide resolved
...he/flink/runtime/io/network/partition/hybrid/tiered/tier/remote/RemoteTierConsumerAgent.java Outdated Show resolved Hide resolved
...he/flink/runtime/io/network/partition/hybrid/tiered/tier/memory/MemoryTierConsumerAgent.java Outdated Show resolved Hide resolved
@yunfengzhou-hub yunfengzhou-hub force-pushed the input-channel-memory branch 3 times, most recently from eaea6b4 to 5d18a44 Compare January 9, 2024 06:06
@yunfengzhou-hub
Copy link
Contributor Author

Hi @reswqa Could you please take a look at this PR?

@yunfengzhou-hub yunfengzhou-hub force-pushed the input-channel-memory branch 3 times, most recently from 7758b0a to 54200b3 Compare January 15, 2024 03:54
@yunfengzhou-hub yunfengzhou-hub force-pushed the input-channel-memory branch 2 times, most recently from 7b58307 to 841bba0 Compare January 15, 2024 08:53
@yunfengzhou-hub
[FLINK-33743][runtime] Distinguish channel and subpartition
ccf36f0
@yunfengzhou-hub
[FLINK-33743][runtime] Replace consumer-side SubpartitionId with Inpu…
48ef8a4 
…tChannelId
@yunfengzhou-hub
[FLINK-33743][runtime] Modify parameter subpartitionIndex to subparti…
121c5bb 
…tionIndexSet
@yunfengzhou-hub
[FLINK-33743][runtime] Identify subpartitionId in notifyRequiredSegment
097c2fc
@yunfengzhou-hub
[FLINK-33743][runtime] Disable tier shuffle during recovery
3e5eba7
@yunfengzhou-hub
[FLINK-33743][runtime] Acquire subpartition id for input channel dyna…
b6c7896 
…mically in tiered shuffle
@yunfengzhou-hub
[FLINK-33743][runtime] Change numCreditsAvailable to isCreditAvailabl…
e94e218 
…e in ResultSubpartitionView#getAvailabilityAndBacklog
@yunfengzhou-hub
[FLINK-33743][runtime] Add ResultSubpartitionView parameter to notify…
57d5fea 
…DataAvailable() method
@yunfengzhou-hub
[FLINK-33743][runtime] Support consuming multiple subpartition in one…
4b51caa 
… inputchannel
@yunfengzhou-hub
[FLINK-33743][runtime] Rename InputChannelStatus to SubpartitionStatus
25cae9a
@yunfengzhou-hub
[FLINK-33743][runtime] Align watermark at subpartition granularity
cfc1869
@yunfengzhou-hub
[FLINK-33743][runtime] Ignore RecordAttributes when adaptive parallel…
d1609ef 
…ism is used
@yunfengzhou-hub
[FLINK-33743][runtime] Flush one accumulated buffer at a time
71a9689
@yunfengzhou-hub
[FLINK-33743][runtime] Optimize partial record split logic
c1ad90d
@reswqa reswqa merged commit a886339 into apache:master Jan 17, 2024
reswqa
reswqa reviewed Jan 17, 2024
View reviewed changes
Copy link
Member

@reswqa reswqa left a comment •
edited

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks @yunfengzhou-hub, merged(I somehow forgot to approval :)).

@snuyanzin snuyanzin mentioned this pull request Jan 18, 2024
Closed
@yunfengzhou-hub yunfengzhou-hub mentioned this pull request Jan 24, 2024
Merged
@SteNicholas SteNicholas mentioned this pull request Mar 18, 2024
Closed
RexXiong pushed a commit to apache/celeborn that referenced this pull request Mar 20, 2024
@SteNicholas @RexXiong
[CELEBORN-1310][FLINK] Support Flink 1.19
adaa96f 
### What changes were proposed in this pull request?

Support Flink 1.19.

### Why are the changes needed?

Flink 1.19.0 is announced to release: [Announcing the Release of Apache Flink 1.19] (https://flink.apache.org/2024/03/18/announcing-the-release-of-apache-flink-1.19).

The main changes includes:

- `org.apache.flink.runtime.io.network.partition.consumer.RemoteInputChannel` constructor change parameters:
   - `consumedSubpartitionIndex` changes to `consumedSubpartitionIndexSet`: [[FLINK-33743][runtime] Support consuming multiple subpartitions on a single channel](apache/flink#23927).
   - adds `partitionRequestListenerTimeout`: [[FLINK-25055][network] Support listen and notify mechanism for partition request](apache/flink#23565).
- `org.apache.flink.runtime.io.network.partition.consumer.SingleInputGate` constructor removes parameters `subpartitionIndexRange`, `tieredStorageConsumerClient`, `nettyService` and `tieredStorageConsumerSpecs`: [[FLINK-33743][runtime] Support consuming multiple subpartitions on a single channel](apache/flink#23927).
- Change the default config file to `config.yaml` in `flink-dist`: [[FLINK-33577][dist] Change the default config file to config.yaml in flink-dist](apache/flink#24177).
- `org.apache.flink.configuration.RestartStrategyOptions` uses `org.apache.commons.compress.utils.Sets` of `commons-compress` dependency: [[FLINK-33865][runtime] Adding an ITCase to ensure exponential-delay.attempts-before-reset-backoff works well](apache/flink#23942).

### Does this PR introduce _any_ user-facing change?

No.

### How was this patch tested?

Local test:

- Flink batch job submission

```
$ ./bin/flink run examples/streaming/WordCount.jar --execution-mode BATCH
Executing example with default input data.
Use --input to specify file input.
Printing result to stdout. Use --output to specify output path.
Job has been submitted with JobID 2e9fb659991a9c29d376151783bdf6de
Program execution finished
Job with JobID 2e9fb659991a9c29d376151783bdf6de has finished.
Job Runtime: 1912 ms
```

- Flink batch job execution

![image](https://github.com/apache/incubator-celeborn/assets/10048174/18b60861-cafc-4df3-b94d-93307e728be2)

- Celeborn master log
```

24/03/18 20:52:47,513 INFO [celeborn-dispatcher-42] Master: Offer slots successfully for 1 reducers of 1710766312631-2e9fb659991a9c29d376151783bdf6de-0 on 1 workers.
```

- Celeborn worker log
```
24/03/18 20:52:47,704 INFO [celeborn-dispatcher-1] StorageManager: created file at /Users/nicholas/Software/Celeborn/apache-celeborn-0.5.0-SNAPSHOT/shuffle/celeborn-worker/shuffle_data/1710766312631-2e9fb659991a9c29d376151783bdf6de/0/0-0-0
24/03/18 20:52:47,707 INFO [celeborn-dispatcher-1] Controller: Reserved 1 primary location and 0 replica location for 1710766312631-2e9fb659991a9c29d376151783bdf6de-0
24/03/18 20:52:47,874 INFO [celeborn-dispatcher-2] Controller: Start commitFiles for 1710766312631-2e9fb659991a9c29d376151783bdf6de-0
24/03/18 20:52:47,890 INFO [worker-rpc-async-replier] Controller: CommitFiles for 1710766312631-2e9fb659991a9c29d376151783bdf6de-0 success with 1 committed primary partitions, 0 empty primary partitions, 0 failed primary partitions, 0 committed replica partitions, 0 empty replica partitions, 0 failed replica partitions.
```

Closes #2399 from SteNicholas/CELEBORN-1310.

Authored-by: SteNicholas <programgeek@163.com>
Signed-off-by: Shuang <lvshuang.xjs@alibaba-inc.com>
@SteNicholas SteNicholas mentioned this pull request Apr 7, 2024
Closed
RexXiong pushed a commit to apache/celeborn that referenced this pull request Apr 9, 2024
@SteNicholas @RexXiong
[CELEBORN-1374] Refactor SortBuffer and PartitionSortedBuffer
7fa1d32 
### What changes were proposed in this pull request?

Refactor `SortBuffer` and `PartitionSortedBuffer` with introduction of `DataBuffer` and `SortBasedDataBuffer`.

### Why are the changes needed?

`SortBuffer` and `PartitionSortedBuffer` is refactored in apache/flink#18505. Celeborn Flink should also refactor `SortBuffer` and `PartitionSortedBuffer` to sync the interface changes in Flink. Meanwhile, `SortBuffer` and `PartitionSortedBuffer` should distinguish channel and subpartition for apache/flink#23927.

### Does this PR introduce _any_ user-facing change?

- `SortBuffer` renames to `DataBuffer`.
- `PartitionSortedBuffer` renames to `SortBasedDataBuffer`.
- `SortBuffer.BufferWithChannel` renames to `BufferWithSubpartition`

### How was this patch tested?

UT and IT.

Closes #2448 from SteNicholas/CELEBORN-1374.

Authored-by: SteNicholas <programgeek@163.com>
Signed-off-by: Shuang <lvshuang.xjs@alibaba-inc.com>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Reviewers

@xintongsong xintongsong xintongsong left review comments

@reswqa reswqa reswqa left review comments

@TanYuxin-tyx TanYuxin-tyx TanYuxin-tyx left review comments

Assignees
No one assigned
Labels
component=Runtime/Network
Projects
None yet
Milestone
No milestone
5 participants
@yunfengzhou-hub @flinkbot @TanYuxin-tyx @xintongsong @reswqa