This is an automated email from the ASF dual-hosted git repository.
SteNicholas pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/celeborn.git
The following commit(s) were added to refs/heads/main by this push:
new 63e28aedd7 [CELEBORN-2373] Support Flink 2.3
63e28aedd7 is described below
commit 63e28aedd7f0703ffeba438d49ea6f3809160892
Author: Nicholas Jiang <[email protected]>
AuthorDate: Tue Jun 30 10:35:14 2026 +0800
[CELEBORN-2373] Support Flink 2.3
### What changes were proposed in this pull request?
This PR adds an Apache Flink 2.3 client for Celeborn:
- Add the `client-flink/flink-2.3` and `client-flink/flink-2.3-shaded`
modules.
- Introduce `CelebornFlinkShimV23` and `tiered/CelebornTierFactory` for
Flink 2.3, registered via the
`org.apache.celeborn.plugin.flink.CelebornFlinkShim` service file.
- Adapt the common `RemoteShuffleInputGate` to the Flink 2.3
`IndexedInputGate`/`InputGate` API surface: add
`getBufferFilteringCompleteFuture()` and the
`setCheckpointingDuringRecoveryEnabled`/`isCheckpointingDuringRecoveryEnabled`
hooks, and pass the extra recovered-state argument (`new ArrayDeque<>()`)
introduced by the recovery-checkpointing work in Flink 2.3.
- Port the existing Flink client UT suites to the new module
(`RemoteShuffleMasterSuiteJV23`, `RemoteShuffleResultPartitionSuiteJV23`,
`RemoteShuffleResultPartitionFactorySuiteJV23`,
`RemoteShuffleServiceFactorySuiteJV23`, `ShuffleResourceTrackerSuiteJV23`,
`tiered/CelebornTierMasterAgentSuiteJV23`).
- Wire the new module into the build/release/CI: `pom.xml`,
`project/CelebornBuild.scala`, `build/make-distribution.sh`,
`build/release/release.sh`, `dev/dependencies.sh`, `dev/reformat`, the
`deps`/`license`/`style` workflows, and add the
`dev/deps/dependencies-client-flink-2.3` dependency list plus the shaded
`LICENSE`/`NOTICE` files.
- Update docs (`README.md`, `docs/README.md`, `docs/deploy.md`,
`docs/developers/overview.md`, `docs/developers/sbt.md`) to include Flink 2.3.
### Why are the changes needed?
[Apache Flink
2.3.0](https://flink.apache.org/2026/06/25/apache-flink-2.3.0-release-announcement/)
was released on June 25, 2026. Among other things it adds support for
checkpointing during unaligned-checkpoint recovery, which changed the
`InputGate` API that the Celeborn Flink client integrates against:
- [FLINK-38543]
[apache/flink#27862](https://github.com/apache/flink/pull/27862) — change the
overall UC restore process, JM and task initialization (introduces
`bufferFilteringCompleteFuture` and the checkpointing-during-recovery
transition).
- [FLINK-38930]
[apache/flink#27783](https://github.com/apache/flink/pull/27783) — filter
records before processing without a spilling strategy during recovery.
- [FLINK-39018]
[apache/flink#27861](https://github.com/apache/flink/pull/27861) — support
checkpoint for `LocalInputChannel`.
Celeborn needs a dedicated Flink 2.3 client so users running on Flink 2.3
can use remote shuffle.
### Does this PR resolve a correctness bug?
- [ ] Yes
No.
### Does this PR introduce _any_ user-facing change?
- [x] Yes
A new Flink 2.3 client is available via the `-Pflink-2.3` profile, the
release binary now ships a `celeborn-client-flink-2.3-shaded` jar, and the docs
list Flink 2.3 as a supported version.
### How was this patch tested?
The existing Flink client UT suites are ported to the new `flink-2.3`
module and run under the `-Pflink-2.3` profile. The `deps`, `license`, and
`style` CI workflows are extended to cover `flink-2.3`.
Closes #3751 from SteNicholas/CELEBORN-2373.
Authored-by: Nicholas Jiang <[email protected]>
Signed-off-by: Nicholas Jiang <[email protected]>
---
.github/workflows/deps.yml | 1 +
.github/workflows/license.yml | 1 +
.github/workflows/maven.yml | 1 +
.github/workflows/sbt.yml | 1 +
.github/workflows/style.yml | 1 +
README.md | 5 +-
build/make-distribution.sh | 2 +
build/release/release.sh | 3 +
.../plugin/flink/RemoteShuffleInputGate.java | 14 ++
client-flink/flink-2.3-shaded/pom.xml | 150 +++++++++++++
.../src/main/resources/META-INF/LICENSE | 248 +++++++++++++++++++++
.../src/main/resources/META-INF/NOTICE | 45 ++++
.../META-INF/licenses/LICENSE-protobuf.txt | 42 ++++
client-flink/flink-2.3/pom.xml | 94 ++++++++
.../plugin/flink/CelebornFlinkShimV23.java | 209 +++++++++++++++++
.../plugin/flink/tiered/CelebornTierFactory.java | 78 +++++++
....apache.celeborn.plugin.flink.CelebornFlinkShim | 16 ++
.../plugin/flink/RemoteShuffleMasterSuiteJV23.java | 20 ++
...moteShuffleResultPartitionFactorySuiteJV23.java | 22 ++
.../RemoteShuffleResultPartitionSuiteJV23.java | 20 ++
.../RemoteShuffleServiceFactorySuiteJV23.java | 20 ++
.../flink/ShuffleResourceTrackerSuiteJV23.java | 20 ++
.../tiered/CelebornTierMasterAgentSuiteJV23.java | 20 ++
dev/dependencies.sh | 4 +
dev/deps/dependencies-client-flink-2.3 | 82 +++++++
dev/reformat | 1 +
docs/README.md | 4 +-
docs/deploy.md | 2 +-
docs/developers/overview.md | 2 +-
docs/developers/sbt.md | 1 +
pom.xml | 20 ++
project/CelebornBuild.scala | 11 +
.../apache/celeborn/tests/flink/FlinkVersion.java | 5 +-
33 files changed, 1158 insertions(+), 7 deletions(-)
diff --git a/.github/workflows/deps.yml b/.github/workflows/deps.yml
index 6578c7b3c0..075eda28e7 100644
--- a/.github/workflows/deps.yml
+++ b/.github/workflows/deps.yml
@@ -103,6 +103,7 @@ jobs:
- 'flink-2.0'
- 'flink-2.1'
- 'flink-2.2'
+ - 'flink-2.3'
steps:
- uses: actions/checkout@v4
- name: Setup JDK 11
diff --git a/.github/workflows/license.yml b/.github/workflows/license.yml
index 40cad4ad9e..67f7b175a8 100644
--- a/.github/workflows/license.yml
+++ b/.github/workflows/license.yml
@@ -49,6 +49,7 @@ jobs:
build/mvn org.apache.rat:apache-rat-plugin:check
-Pgoogle-mirror,flink-2.0
build/mvn org.apache.rat:apache-rat-plugin:check
-Pgoogle-mirror,flink-2.1
build/mvn org.apache.rat:apache-rat-plugin:check
-Pgoogle-mirror,flink-2.2
+ build/mvn org.apache.rat:apache-rat-plugin:check
-Pgoogle-mirror,flink-2.3
build/mvn org.apache.rat:apache-rat-plugin:check
-Pgoogle-mirror,spark-3.3
build/mvn org.apache.rat:apache-rat-plugin:check -Pgoogle-mirror,mr
- name: Upload rat report
diff --git a/.github/workflows/maven.yml b/.github/workflows/maven.yml
index 540d630fe0..8cedf8c968 100644
--- a/.github/workflows/maven.yml
+++ b/.github/workflows/maven.yml
@@ -203,6 +203,7 @@ jobs:
- '2.0'
- '2.1'
- '2.2'
+ - '2.3'
steps:
- uses: actions/checkout@v4
- name: Setup JDK ${{ matrix.java }}
diff --git a/.github/workflows/sbt.yml b/.github/workflows/sbt.yml
index 5ec630b2fb..ec58b11f10 100644
--- a/.github/workflows/sbt.yml
+++ b/.github/workflows/sbt.yml
@@ -252,6 +252,7 @@ jobs:
- '2.0'
- '2.1'
- '2.2'
+ - '2.3'
steps:
- uses: actions/checkout@v4
- name: Setup JDK ${{ matrix.java }}
diff --git a/.github/workflows/style.yml b/.github/workflows/style.yml
index 55614d637a..41ab7b7a9a 100644
--- a/.github/workflows/style.yml
+++ b/.github/workflows/style.yml
@@ -53,5 +53,6 @@ jobs:
build/mvn spotless:check -Pgoogle-mirror,flink-2.0
build/mvn spotless:check -Pgoogle-mirror,flink-2.1
build/mvn spotless:check -Pgoogle-mirror,flink-2.2
+ build/mvn spotless:check -Pgoogle-mirror,flink-2.3
build/mvn spotless:check -Pgoogle-mirror,spark-3.3
build/mvn spotless:check -Pgoogle-mirror,mr
diff --git a/README.md b/README.md
index 4e435ce422..cf349404d6 100644
--- a/README.md
+++ b/README.md
@@ -41,12 +41,12 @@ Celeborn Worker's slot count is decided by `total usable
disk size / average shu
Celeborn worker's slot count decreases when a partition is allocated and
increments when a partition is freed.
## Build
-1. Celeborn supports Spark 3.0/3.1/3.2/3.3/3.4/3.5/4.0, Flink
1.18/1.19/1.20/2.0/2.1/2.2 and Hadoop MapReduce 3.
+1. Celeborn supports Spark 3.0/3.1/3.2/3.3/3.4/3.5/4.0, Flink
1.18/1.19/1.20/2.0/2.1/2.2/2.3 and Hadoop MapReduce 3.
2. Celeborn tested under Scala 2.11/2.12/2.13 and Java 8/11/17 environment.
Build Celeborn via `make-distribution.sh`:
```shell
-./build/make-distribution.sh
-Pspark-3.0/-Pspark-3.1/-Pspark-3.2/-Pspark-3.3/-Pspark-3.4/-Pspark-3.5/-Pspark-4.0/-Pspark-4.1/-Pflink-1.18/-Pflink-1.19/-Pflink-1.20/-Pflink-2.0/-Pflink-2.1/-Pflink-2.2/-Pmr
+./build/make-distribution.sh
-Pspark-3.0/-Pspark-3.1/-Pspark-3.2/-Pspark-3.3/-Pspark-3.4/-Pspark-3.5/-Pspark-4.0/-Pspark-4.1/-Pflink-1.18/-Pflink-1.19/-Pflink-1.20/-Pflink-2.0/-Pflink-2.1/-Pflink-2.2/-Pflink-2.3/-Pmr
```
Package `apache-celeborn-${project.version}-bin.tgz` will be generated.
@@ -69,6 +69,7 @@ Package `apache-celeborn-${project.version}-bin.tgz` will be
generated.
| Flink 2.0 | ❌ | ❌ | ✔ |
✔ | ❌ | ✔ | ✔
|
| Flink 2.1 | ❌ | ❌ | ✔ |
✔ | ❌ | ✔ | ✔
|
| Flink 2.2 | ❌ | ❌ | ✔ |
✔ | ❌ | ✔ | ✔
|
+| Flink 2.3 | ❌ | ❌ | ✔ |
✔ | ❌ | ✔ | ✔
|
To compile for Spark 3.5 with Java21, please use the following command
```shell
diff --git a/build/make-distribution.sh b/build/make-distribution.sh
index 721673b42c..5817f7247b 100755
--- a/build/make-distribution.sh
+++ b/build/make-distribution.sh
@@ -387,6 +387,7 @@ if [ "$SBT_ENABLED" == "true" ]; then
sbt_build_client -Pflink-2.0
sbt_build_client -Pflink-2.1
sbt_build_client -Pflink-2.2
+ sbt_build_client -Pflink-2.3
export JAVA_HOME=$JAVA8_HOME
sbt_build_client -Pmr
# sbt_build_client -Ptez
@@ -428,6 +429,7 @@ else
build_flink_client -Pflink-2.0
build_flink_client -Pflink-2.1
build_flink_client -Pflink-2.2
+ build_flink_client -Pflink-2.3
export JAVA_HOME=$JAVA8_HOME
build_mr_client -Pmr
build_tez_client -Ptez
diff --git a/build/release/release.sh b/build/release/release.sh
index 95bc6bb9c9..5aff151567 100755
--- a/build/release/release.sh
+++ b/build/release/release.sh
@@ -133,6 +133,9 @@ upload_nexus_staging() {
echo "Deploying celeborn-client-flink-2.2-shaded_2.12"
${PROJECT_DIR}/build/sbt -Pflink-2.2
"clean;celeborn-client-flink-2_2-shaded/publishSigned"
+
+ echo "Deploying celeborn-client-flink-2.3-shaded_2.12"
+ ${PROJECT_DIR}/build/sbt -Pflink-2.3
"clean;celeborn-client-flink-2_3-shaded/publishSigned"
export JAVA_HOME=$JAVA8_HOME
echo "Deploying celeborn-client-mr-shaded_2.12"
diff --git
a/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleInputGate.java
b/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleInputGate.java
index cad869eff7..046912952e 100644
---
a/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleInputGate.java
+++
b/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleInputGate.java
@@ -48,6 +48,8 @@ public class RemoteShuffleInputGate extends IndexedInputGate {
public final RemoteShuffleInputGateDelegation inputGateDelegation;
private final ResultPartitionType consumedPartitionType;
+ private volatile boolean checkpointingDuringRecoveryEnabled = false;
+
public RemoteShuffleInputGate(
CelebornConf celebornConf,
ShuffleIOOwnerContext ownerContext,
@@ -206,4 +208,16 @@ public class RemoteShuffleInputGate extends
IndexedInputGate {
// TODO only used in 1.19
return consumedPartitionType;
}
+
+ public CompletableFuture<Void> getBufferFilteringCompleteFuture() {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ public void setCheckpointingDuringRecoveryEnabled(boolean enabled) {
+ this.checkpointingDuringRecoveryEnabled = enabled;
+ }
+
+ public boolean isCheckpointingDuringRecoveryEnabled() {
+ return checkpointingDuringRecoveryEnabled;
+ }
}
diff --git a/client-flink/flink-2.3-shaded/pom.xml
b/client-flink/flink-2.3-shaded/pom.xml
new file mode 100644
index 0000000000..a41c5361de
--- /dev/null
+++ b/client-flink/flink-2.3-shaded/pom.xml
@@ -0,0 +1,150 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ Licensed to the Apache Software Foundation (ASF) under one or more
+ ~ contributor license agreements. See the NOTICE file distributed with
+ ~ this work for additional information regarding copyright ownership.
+ ~ The ASF licenses this file to You under the Apache License, Version 2.0
+ ~ (the "License"); you may not use this file except in compliance with
+ ~ the License. You may obtain a copy of the License at
+ ~
+ ~ http://www.apache.org/licenses/LICENSE-2.0
+ ~
+ ~ Unless required by applicable law or agreed to in writing, software
+ ~ distributed under the License is distributed on an "AS IS" BASIS,
+ ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ ~ See the License for the specific language governing permissions and
+ ~ limitations under the License.
+ -->
+<project xmlns="http://maven.apache.org/POM/4.0.0";
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.celeborn</groupId>
+ <artifactId>celeborn-parent_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <relativePath>../../pom.xml</relativePath>
+ </parent>
+
+
<artifactId>celeborn-client-flink-2.3-shaded_${scala.binary.version}</artifactId>
+ <packaging>jar</packaging>
+ <name>Celeborn Shaded Client for Flink 2.3</name>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.celeborn</groupId>
+
<artifactId>celeborn-client-flink-2.3_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-shade-plugin</artifactId>
+ <configuration>
+ <relocations>
+ <relocation>
+ <pattern>com.google.protobuf</pattern>
+
<shadedPattern>${shading.prefix}.com.google.protobuf</shadedPattern>
+ </relocation>
+ <relocation>
+ <pattern>com.google.common</pattern>
+
<shadedPattern>${shading.prefix}.com.google.common</shadedPattern>
+ </relocation>
+ <relocation>
+ <pattern>io.netty</pattern>
+ <shadedPattern>${shading.prefix}.io.netty</shadedPattern>
+ </relocation>
+ <relocation>
+ <pattern>org.apache.commons</pattern>
+
<shadedPattern>${shading.prefix}.org.apache.commons</shadedPattern>
+ </relocation>
+ <relocation>
+ <pattern>org.roaringbitmap</pattern>
+
<shadedPattern>${shading.prefix}.org.roaringbitmap</shadedPattern>
+ </relocation>
+ </relocations>
+ <artifactSet>
+ <includes>
+ <include>org.apache.celeborn:*</include>
+ <include>com.google.protobuf:protobuf-java</include>
+ <include>com.google.guava:guava</include>
+ <include>com.google.guava:failureaccess</include>
+ <include>io.netty:*</include>
+ <include>org.apache.commons:commons-lang3</include>
+ <include>org.roaringbitmap:RoaringBitmap</include>
+ </includes>
+ </artifactSet>
+ <filters>
+ <filter>
+ <artifact>*:*</artifact>
+ <excludes>
+ <exclude>**/*.proto</exclude>
+ <exclude>META-INF/*.SF</exclude>
+ <exclude>META-INF/*.DSA</exclude>
+ <exclude>META-INF/*.RSA</exclude>
+ <exclude>**/log4j.properties</exclude>
+ <exclude>META-INF/LICENSE.txt</exclude>
+ <exclude>META-INF/NOTICE.txt</exclude>
+ <exclude>LICENSE.txt</exclude>
+ <exclude>NOTICE.txt</exclude>
+ </excludes>
+ </filter>
+ </filters>
+ <transformers>
+ <transformer
implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"></transformer>
+ </transformers>
+ </configuration>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-antrun-plugin</artifactId>
+ <version>${maven.plugin.antrun.version}</version>
+ <executions>
+ <execution>
+ <id>rename-native-library</id>
+ <goals>
+ <goal>run</goal>
+ </goals>
+ <phase>package</phase>
+ <configuration>
+ <target>
+ <echo message="unpacking netty jar"></echo>
+ <unzip dest="${project.build.directory}/unpacked/"
src="${project.build.directory}/${artifactId}-${version}.jar"></unzip>
+ <echo message="renaming native epoll library"></echo>
+ <move includeemptydirs="false"
todir="${project.build.directory}/unpacked/META-INF/native">
+ <fileset
dir="${project.build.directory}/unpacked/META-INF/native"></fileset>
+ <mapper from="libnetty_transport_native_epoll_x86_64.so"
to="liborg_apache_celeborn_shaded_netty_transport_native_epoll_x86_64.so"
type="glob"></mapper>
+ </move>
+ <move includeemptydirs="false"
todir="${project.build.directory}/unpacked/META-INF/native">
+ <fileset
dir="${project.build.directory}/unpacked/META-INF/native"></fileset>
+ <mapper from="libnetty_transport_native_epoll_aarch_64.so"
to="liborg_apache_celeborn_shaded_netty_transport_native_epoll_aarch_64.so"
type="glob"></mapper>
+ </move>
+ <echo message="renaming native kqueue library"></echo>
+ <move includeemptydirs="false"
todir="${project.build.directory}/unpacked/META-INF/native">
+ <fileset
dir="${project.build.directory}/unpacked/META-INF/native"></fileset>
+ <mapper
from="libnetty_transport_native_kqueue_x86_64.jnilib"
to="liborg_apache_celeborn_shaded_netty_transport_native_kqueue_x86_64.jnilib"
type="glob"></mapper>
+ </move>
+ <move includeemptydirs="false"
todir="${project.build.directory}/unpacked/META-INF/native">
+ <fileset
dir="${project.build.directory}/unpacked/META-INF/native"></fileset>
+ <mapper
from="libnetty_transport_native_kqueue_aarch_64.jnilib"
to="liborg_apache_celeborn_shaded_netty_transport_native_kqueue_aarch_64.jnilib"
type="glob"></mapper>
+ </move>
+ <echo message="renaming native dns resolver library"></echo>
+ <move includeemptydirs="false"
todir="${project.build.directory}/unpacked/META-INF/native">
+ <fileset
dir="${project.build.directory}/unpacked/META-INF/native"></fileset>
+ <mapper
from="libnetty_resolver_dns_native_macos_aarch_64.jnilib"
to="liborg_apache_celeborn_shaded_netty_resolver_dns_native_macos_aarch_64.jnilib"
type="glob"></mapper>
+ </move>
+ <move includeemptydirs="false"
todir="${project.build.directory}/unpacked/META-INF/native">
+ <fileset
dir="${project.build.directory}/unpacked/META-INF/native"></fileset>
+ <mapper
from="libnetty_resolver_dns_native_macos_x86_64.jnilib"
to="liborg_apache_celeborn_shaded_netty_resolver_dns_native_macos_x86_64.jnilib"
type="glob"></mapper>
+ </move>
+ <echo message="repackaging netty jar"></echo>
+ <jar basedir="${project.build.directory}/unpacked"
destfile="${project.build.directory}/${artifactId}-${version}.jar"></jar>
+ </target>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+</project>
diff --git a/client-flink/flink-2.3-shaded/src/main/resources/META-INF/LICENSE
b/client-flink/flink-2.3-shaded/src/main/resources/META-INF/LICENSE
new file mode 100644
index 0000000000..924ef2c85f
--- /dev/null
+++ b/client-flink/flink-2.3-shaded/src/main/resources/META-INF/LICENSE
@@ -0,0 +1,248 @@
+ Apache License
+ Version 2.0, January 2004
+ http://www.apache.org/licenses/
+
+ TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
+
+ 1. Definitions.
+
+ "License" shall mean the terms and conditions for use, reproduction,
+ and distribution as defined by Sections 1 through 9 of this document.
+
+ "Licensor" shall mean the copyright owner or entity authorized by
+ the copyright owner that is granting the License.
+
+ "Legal Entity" shall mean the union of the acting entity and all
+ other entities that control, are controlled by, or are under common
+ control with that entity. For the purposes of this definition,
+ "control" means (i) the power, direct or indirect, to cause the
+ direction or management of such entity, whether by contract or
+ otherwise, or (ii) ownership of fifty percent (50%) or more of the
+ outstanding shares, or (iii) beneficial ownership of such entity.
+
+ "You" (or "Your") shall mean an individual or Legal Entity
+ exercising permissions granted by this License.
+
+ "Source" form shall mean the preferred form for making modifications,
+ including but not limited to software source code, documentation
+ source, and configuration files.
+
+ "Object" form shall mean any form resulting from mechanical
+ transformation or translation of a Source form, including but
+ not limited to compiled object code, generated documentation,
+ and conversions to other media types.
+
+ "Work" shall mean the work of authorship, whether in Source or
+ Object form, made available under the License, as indicated by a
+ copyright notice that is included in or attached to the work
+ (an example is provided in the Appendix below).
+
+ "Derivative Works" shall mean any work, whether in Source or Object
+ form, that is based on (or derived from) the Work and for which the
+ editorial revisions, annotations, elaborations, or other modifications
+ represent, as a whole, an original work of authorship. For the purposes
+ of this License, Derivative Works shall not include works that remain
+ separable from, or merely link (or bind by name) to the interfaces of,
+ the Work and Derivative Works thereof.
+
+ "Contribution" shall mean any work of authorship, including
+ the original version of the Work and any modifications or additions
+ to that Work or Derivative Works thereof, that is intentionally
+ submitted to Licensor for inclusion in the Work by the copyright owner
+ or by an individual or Legal Entity authorized to submit on behalf of
+ the copyright owner. For the purposes of this definition, "submitted"
+ means any form of electronic, verbal, or written communication sent
+ to the Licensor or its representatives, including but not limited to
+ communication on electronic mailing lists, source code control systems,
+ and issue tracking systems that are managed by, or on behalf of, the
+ Licensor for the purpose of discussing and improving the Work, but
+ excluding communication that is conspicuously marked or otherwise
+ designated in writing by the copyright owner as "Not a Contribution."
+
+ "Contributor" shall mean Licensor and any individual or Legal Entity
+ on behalf of whom a Contribution has been received by Licensor and
+ subsequently incorporated within the Work.
+
+ 2. Grant of Copyright License. Subject to the terms and conditions of
+ this License, each Contributor hereby grants to You a perpetual,
+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+ copyright license to reproduce, prepare Derivative Works of,
+ publicly display, publicly perform, sublicense, and distribute the
+ Work and such Derivative Works in Source or Object form.
+
+ 3. Grant of Patent License. Subject to the terms and conditions of
+ this License, each Contributor hereby grants to You a perpetual,
+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+ (except as stated in this section) patent license to make, have made,
+ use, offer to sell, sell, import, and otherwise transfer the Work,
+ where such license applies only to those patent claims licensable
+ by such Contributor that are necessarily infringed by their
+ Contribution(s) alone or by combination of their Contribution(s)
+ with the Work to which such Contribution(s) was submitted. If You
+ institute patent litigation against any entity (including a
+ cross-claim or counterclaim in a lawsuit) alleging that the Work
+ or a Contribution incorporated within the Work constitutes direct
+ or contributory patent infringement, then any patent licenses
+ granted to You under this License for that Work shall terminate
+ as of the date such litigation is filed.
+
+ 4. Redistribution. You may reproduce and distribute copies of the
+ Work or Derivative Works thereof in any medium, with or without
+ modifications, and in Source or Object form, provided that You
+ meet the following conditions:
+
+ (a) You must give any other recipients of the Work or
+ Derivative Works a copy of this License; and
+
+ (b) You must cause any modified files to carry prominent notices
+ stating that You changed the files; and
+
+ (c) You must retain, in the Source form of any Derivative Works
+ that You distribute, all copyright, patent, trademark, and
+ attribution notices from the Source form of the Work,
+ excluding those notices that do not pertain to any part of
+ the Derivative Works; and
+
+ (d) If the Work includes a "NOTICE" text file as part of its
+ distribution, then any Derivative Works that You distribute must
+ include a readable copy of the attribution notices contained
+ within such NOTICE file, excluding those notices that do not
+ pertain to any part of the Derivative Works, in at least one
+ of the following places: within a NOTICE text file distributed
+ as part of the Derivative Works; within the Source form or
+ documentation, if provided along with the Derivative Works; or,
+ within a display generated by the Derivative Works, if and
+ wherever such third-party notices normally appear. The contents
+ of the NOTICE file are for informational purposes only and
+ do not modify the License. You may add Your own attribution
+ notices within Derivative Works that You distribute, alongside
+ or as an addendum to the NOTICE text from the Work, provided
+ that such additional attribution notices cannot be construed
+ as modifying the License.
+
+ You may add Your own copyright statement to Your modifications and
+ may provide additional or different license terms and conditions
+ for use, reproduction, or distribution of Your modifications, or
+ for any such Derivative Works as a whole, provided Your use,
+ reproduction, and distribution of the Work otherwise complies with
+ the conditions stated in this License.
+
+ 5. Submission of Contributions. Unless You explicitly state otherwise,
+ any Contribution intentionally submitted for inclusion in the Work
+ by You to the Licensor shall be under the terms and conditions of
+ this License, without any additional terms or conditions.
+ Notwithstanding the above, nothing herein shall supersede or modify
+ the terms of any separate license agreement you may have executed
+ with Licensor regarding such Contributions.
+
+ 6. Trademarks. This License does not grant permission to use the trade
+ names, trademarks, service marks, or product names of the Licensor,
+ except as required for reasonable and customary use in describing the
+ origin of the Work and reproducing the content of the NOTICE file.
+
+ 7. Disclaimer of Warranty. Unless required by applicable law or
+ agreed to in writing, Licensor provides the Work (and each
+ Contributor provides its Contributions) on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+ implied, including, without limitation, any warranties or conditions
+ of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
+ PARTICULAR PURPOSE. You are solely responsible for determining the
+ appropriateness of using or redistributing the Work and assume any
+ risks associated with Your exercise of permissions under this License.
+
+ 8. Limitation of Liability. In no event and under no legal theory,
+ whether in tort (including negligence), contract, or otherwise,
+ unless required by applicable law (such as deliberate and grossly
+ negligent acts) or agreed to in writing, shall any Contributor be
+ liable to You for damages, including any direct, indirect, special,
+ incidental, or consequential damages of any character arising as a
+ result of this License or out of the use or inability to use the
+ Work (including but not limited to damages for loss of goodwill,
+ work stoppage, computer failure or malfunction, or any and all
+ other commercial damages or losses), even if such Contributor
+ has been advised of the possibility of such damages.
+
+ 9. Accepting Warranty or Additional Liability. While redistributing
+ the Work or Derivative Works thereof, You may choose to offer,
+ and charge a fee for, acceptance of support, warranty, indemnity,
+ or other liability obligations and/or rights consistent with this
+ License. However, in accepting such obligations, You may act only
+ on Your own behalf and on Your sole responsibility, not on behalf
+ of any other Contributor, and only if You agree to indemnify,
+ defend, and hold each Contributor harmless for any liability
+ incurred by, or claims asserted against, such Contributor by reason
+ of your accepting any such warranty or additional liability.
+
+ END OF TERMS AND CONDITIONS
+
+ APPENDIX: How to apply the Apache License to your work.
+
+ To apply the Apache License to your work, attach the following
+ boilerplate notice, with the fields enclosed by brackets "[]"
+ replaced with your own identifying information. (Don't include
+ the brackets!) The text should be enclosed in the appropriate
+ comment syntax for the file format. We also recommend that a
+ file or class name and description of purpose be included on the
+ same "printed page" as the copyright notice for easier
+ identification within third-party archives.
+
+ Copyright [yyyy] [name of copyright owner]
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+
+
+------------------------------------------------------------------------------------
+This project bundles the following dependencies under the Apache License 2.0
(http://www.apache.org/licenses/LICENSE-2.0.txt):
+
+
+Apache License 2.0
+--------------------------------------
+
+com.google.guava:failureaccess
+com.google.guava:guava
+io.netty:netty-all
+io.netty:netty-buffer
+io.netty:netty-codec
+io.netty:netty-codec-dns
+io.netty:netty-codec-haproxy
+io.netty:netty-codec-http
+io.netty:netty-codec-http2
+io.netty:netty-codec-memcache
+io.netty:netty-codec-mqtt
+io.netty:netty-codec-redis
+io.netty:netty-codec-smtp
+io.netty:netty-codec-socks
+io.netty:netty-codec-stomp
+io.netty:netty-codec-xml
+io.netty:netty-common
+io.netty:netty-handler
+io.netty:netty-handler-proxy
+io.netty:netty-resolver
+io.netty:netty-resolver-dns
+io.netty:netty-transport
+io.netty:netty-transport-classes-epoll
+io.netty:netty-transport-classes-kqueue
+io.netty:netty-transport-native-epoll
+io.netty:netty-transport-native-kqueue
+io.netty:netty-transport-native-unix-common
+io.netty:netty-transport-rxtx
+io.netty:netty-transport-sctp
+io.netty:netty-transport-udt
+org.apache.commons:commons-lang3
+org.roaringbitmap:RoaringBitmap
+
+
+BSD 3-clause
+------------
+See licenses/LICENSE-protobuf.txt for details.
+com.google.protobuf:protobuf-java
diff --git a/client-flink/flink-2.3-shaded/src/main/resources/META-INF/NOTICE
b/client-flink/flink-2.3-shaded/src/main/resources/META-INF/NOTICE
new file mode 100644
index 0000000000..43452a38af
--- /dev/null
+++ b/client-flink/flink-2.3-shaded/src/main/resources/META-INF/NOTICE
@@ -0,0 +1,45 @@
+
+Apache Celeborn
+Copyright 2022-2024 The Apache Software Foundation.
+
+This product includes software developed at
+The Apache Software Foundation (https://www.apache.org/).
+
+Apache Spark
+Copyright 2014 and onwards The Apache Software Foundation
+
+Apache Kyuubi
+Copyright 2021-2023 The Apache Software Foundation
+
+Apache Iceberg
+Copyright 2017-2022 The Apache Software Foundation
+
+Apache Parquet MR
+Copyright 2014-2024 The Apache Software Foundation
+
+This project includes code from Kite, developed at Cloudera, Inc. with
+the following copyright notice:
+
+| Copyright 2013 Cloudera Inc.
+|
+| Licensed under the Apache License, Version 2.0 (the "License");
+| you may not use this file except in compliance with the License.
+| You may obtain a copy of the License at
+|
+| http://www.apache.org/licenses/LICENSE-2.0
+|
+| Unless required by applicable law or agreed to in writing, software
+| distributed under the License is distributed on an "AS IS" BASIS,
+| WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+| See the License for the specific language governing permissions and
+| limitations under the License.
+
+Remote Shuffle Service for Flink
+Copyright 2021 The Flink Remote Shuffle Project
+
+=============================================================================
+= NOTICE file corresponding to section 4d of the Apache License Version 2.0 =
+=============================================================================
+
+Apache Commons Lang
+Copyright 2001-2021 The Apache Software Foundation
diff --git
a/client-flink/flink-2.3-shaded/src/main/resources/META-INF/licenses/LICENSE-protobuf.txt
b/client-flink/flink-2.3-shaded/src/main/resources/META-INF/licenses/LICENSE-protobuf.txt
new file mode 100644
index 0000000000..b4350ec83c
--- /dev/null
+++
b/client-flink/flink-2.3-shaded/src/main/resources/META-INF/licenses/LICENSE-protobuf.txt
@@ -0,0 +1,42 @@
+This license applies to all parts of Protocol Buffers except the following:
+
+ - Atomicops support for generic gcc, located in
+ src/google/protobuf/stubs/atomicops_internals_generic_gcc.h.
+ This file is copyrighted by Red Hat Inc.
+
+ - Atomicops support for AIX/POWER, located in
+ src/google/protobuf/stubs/atomicops_internals_aix.h.
+ This file is copyrighted by Bloomberg Finance LP.
+
+Copyright 2014, Google Inc. All rights reserved.
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are
+met:
+
+ * Redistributions of source code must retain the above copyright
+notice, this list of conditions and the following disclaimer.
+ * Redistributions in binary form must reproduce the above
+copyright notice, this list of conditions and the following disclaimer
+in the documentation and/or other materials provided with the
+distribution.
+ * Neither the name of Google Inc. nor the names of its
+contributors may be used to endorse or promote products derived from
+this software without specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+Code generated by the Protocol Buffer compiler is owned by the owner
+of the input file used when generating it. This code is not
+standalone and requires a support library to be linked with it. This
+support library is itself covered by the above license.
\ No newline at end of file
diff --git a/client-flink/flink-2.3/pom.xml b/client-flink/flink-2.3/pom.xml
new file mode 100644
index 0000000000..16301d5cfb
--- /dev/null
+++ b/client-flink/flink-2.3/pom.xml
@@ -0,0 +1,94 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ Licensed to the Apache Software Foundation (ASF) under one or more
+ ~ contributor license agreements. See the NOTICE file distributed with
+ ~ this work for additional information regarding copyright ownership.
+ ~ The ASF licenses this file to You under the Apache License, Version 2.0
+ ~ (the "License"); you may not use this file except in compliance with
+ ~ the License. You may obtain a copy of the License at
+ ~
+ ~ http://www.apache.org/licenses/LICENSE-2.0
+ ~
+ ~ Unless required by applicable law or agreed to in writing, software
+ ~ distributed under the License is distributed on an "AS IS" BASIS,
+ ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ ~ See the License for the specific language governing permissions and
+ ~ limitations under the License.
+ -->
+<project xmlns="http://maven.apache.org/POM/4.0.0";
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.celeborn</groupId>
+ <artifactId>celeborn-parent_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <relativePath>../../pom.xml</relativePath>
+ </parent>
+
+ <artifactId>celeborn-client-flink-2.3_${scala.binary.version}</artifactId>
+ <packaging>jar</packaging>
+ <name>Celeborn Client for Flink 2.3</name>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.celeborn</groupId>
+ <artifactId>celeborn-common_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.celeborn</groupId>
+ <artifactId>celeborn-client_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.celeborn</groupId>
+
<artifactId>celeborn-client-flink-common_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.celeborn</groupId>
+
<artifactId>celeborn-client-flink-common-tiered_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.celeborn</groupId>
+
<artifactId>celeborn-client-flink-common_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.celeborn</groupId>
+
<artifactId>celeborn-client-flink-common-tiered_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-runtime</artifactId>
+ <scope>provided</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>org.xerial.snappy</groupId>
+ <artifactId>snappy-java</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <!-- Test dependencies -->
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-core</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-slf4j-impl</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-1.2-api</artifactId>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+</project>
diff --git
a/client-flink/flink-2.3/src/main/java/org/apache/celeborn/plugin/flink/CelebornFlinkShimV23.java
b/client-flink/flink-2.3/src/main/java/org/apache/celeborn/plugin/flink/CelebornFlinkShimV23.java
new file mode 100644
index 0000000000..55aeb63589
--- /dev/null
+++
b/client-flink/flink-2.3/src/main/java/org/apache/celeborn/plugin/flink/CelebornFlinkShimV23.java
@@ -0,0 +1,209 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.plugin.flink;
+
+import java.net.InetAddress;
+import java.time.Duration;
+import java.util.ArrayDeque;
+import java.util.Collection;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.stream.Collectors;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.NettyShuffleEnvironmentOptions;
+import org.apache.flink.core.memory.MemorySegmentProvider;
+import org.apache.flink.metrics.SimpleCounter;
+import org.apache.flink.runtime.checkpoint.channel.ChannelStateWriter;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.deployment.InputGateDeploymentDescriptor;
+import org.apache.flink.runtime.executiongraph.IndexRange;
+import org.apache.flink.runtime.io.network.ConnectionID;
+import org.apache.flink.runtime.io.network.LocalConnectionManager;
+import org.apache.flink.runtime.io.network.buffer.Buffer;
+import org.apache.flink.runtime.io.network.buffer.BufferCompressor;
+import org.apache.flink.runtime.io.network.buffer.BufferDecompressor;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionType;
+import
org.apache.flink.runtime.io.network.partition.ResultSubpartitionIndexSet;
+import org.apache.flink.runtime.io.network.partition.consumer.InputChannel;
+import
org.apache.flink.runtime.io.network.partition.consumer.RemoteInputChannel;
+import org.apache.flink.runtime.io.network.partition.consumer.SingleInputGate;
+import
org.apache.flink.runtime.io.network.partition.hybrid.tiered.common.TieredStorageIdMappingUtils;
+import
org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.TierShuffleHandler;
+import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
+import org.apache.flink.runtime.jobgraph.IntermediateResultPartitionID;
+import org.apache.flink.runtime.shuffle.JobShuffleContext;
+import org.apache.flink.runtime.shuffle.PartitionDescriptor;
+import org.apache.flink.runtime.shuffle.PartitionWithMetrics;
+import org.apache.flink.runtime.shuffle.TaskInputsOutputsDescriptor;
+import org.apache.flink.runtime.taskmanager.TaskManagerLocation;
+import org.apache.flink.runtime.throughput.ThroughputCalculator;
+import org.apache.flink.util.clock.SystemClock;
+
+import org.apache.celeborn.plugin.flink.tiered.CelebornFlinkTierShim;
+import org.apache.celeborn.plugin.flink.utils.ReflectionUtils;
+
+public class CelebornFlinkShimV23 extends CelebornFlinkTierShim {
+ @Override
+ protected BufferCompressor createBufferCompressorImpl(
+ int networkBufferSize, String compressionCodec) {
+ return new BufferCompressor(
+ networkBufferSize,
+
NettyShuffleEnvironmentOptions.CompressionCodec.valueOf(compressionCodec));
+ }
+
+ @Override
+ protected BufferDecompressor createBufferDecompressorImpl(
+ int networkBufferSize, String compressionCodec) {
+ return new BufferDecompressor(
+ networkBufferSize,
+
NettyShuffleEnvironmentOptions.CompressionCodec.valueOf(compressionCodec));
+ }
+
+ @Override
+ protected Duration getRequestSegmentsTimeoutImpl(Configuration
configuration) {
+ return
configuration.get(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_REQUEST_TIMEOUT);
+ }
+
+ @Override
+ protected InputChannel createFakeChannelImpl(
+ RemoteShuffleInputGateDelegation inputGateDelegation, int channelIndex) {
+ return new RemoteInputChannel(
+ new SingleInputGate(
+ inputGateDelegation.getTaskName(),
+ inputGateDelegation.getGateIndex(),
+ new IntermediateDataSetID(),
+ ResultPartitionType.BLOCKING,
+ 1,
+ (a, b, c) -> {},
+ () -> null,
+ null,
+
ReflectionUtils.createEmptyImplementation(MemorySegmentProvider.class),
+ 0,
+ new ThroughputCalculator(SystemClock.getInstance()),
+ null),
+ channelIndex,
+ new ResultPartitionID(),
+ new ResultSubpartitionIndexSet(new IndexRange(0, 0)),
+ new ConnectionID(
+ new TaskManagerLocation(ResourceID.generate(),
InetAddress.getLoopbackAddress(), 1), 0),
+ new LocalConnectionManager(),
+ 0,
+ 0,
+ 0,
+ 0,
+ new SimpleCounter(),
+ new SimpleCounter(),
+ ReflectionUtils.createEmptyImplementation(ChannelStateWriter.class),
+ new ArrayDeque<>());
+ }
+
+ @Override
+ protected Tuple2<Integer, Integer> getConsumedSubpartitionIndexRangeImpl(
+ InputGateDeploymentDescriptor gateDescriptor) {
+ IndexRange indexRange = gateDescriptor.getConsumedSubpartitionRange(0);
+ return Tuple2.of(indexRange.getStartIndex(), indexRange.getEndIndex());
+ }
+
+ @Override
+ protected ResultPartitionBytesCounter createResultPartitionBytesCounterImpl(
+ RemoteShuffleResultPartition remoteShuffleResultPartition) {
+ org.apache.flink.runtime.io.network.metrics.ResultPartitionBytesCounter
resultPartitionBytes =
+ ReflectionUtils.readDeclaredFieldRecursiveNoThrow(
+ remoteShuffleResultPartition, "resultPartitionBytes");
+ return (bufferWithSubpartition, isBroadcast, readableBytes) -> {
+ if (isBroadcast) {
+ resultPartitionBytes.incAll(readableBytes);
+ } else {
+
resultPartitionBytes.inc(bufferWithSubpartition.getSubpartitionIndex(),
readableBytes);
+ }
+ };
+ }
+
+ @Override
+ protected Buffer.DataType getEndOfPartitionDataTypeImpl() {
+ return Buffer.DataType.END_OF_PARTITION;
+ }
+
+ @Override
+ protected PartitionDescriptor createPartitionDescriptorImpl(
+ IntermediateDataSetID intermediateDataSetId, int partitionNum) {
+ IntermediateResultPartitionID intermediateResultPartitionId =
+ new IntermediateResultPartitionID(intermediateDataSetId, partitionNum);
+ return new PartitionDescriptor(
+ intermediateDataSetId,
+ 10,
+ intermediateResultPartitionId,
+ ResultPartitionType.BLOCKING,
+ 5,
+ 1,
+ false,
+ false,
+ false);
+ }
+
+ @Override
+ protected TaskInputsOutputsDescriptor createTaskInputsOutputsDescriptorImpl(
+ int inputGateNums,
+ Map<IntermediateDataSetID, Integer> inputChannelNums,
+ Map<IntermediateDataSetID, Integer> partitionReuseCount,
+ Map<IntermediateDataSetID, Integer> subpartitionNums,
+ Map<IntermediateDataSetID, ResultPartitionType> inputPartitionTypes,
+ Map<IntermediateDataSetID, ResultPartitionType> partitionTypes) {
+ return TaskInputsOutputsDescriptor.from(
+ inputGateNums,
+ inputChannelNums,
+ partitionReuseCount,
+ subpartitionNums,
+ inputPartitionTypes,
+ partitionTypes);
+ }
+
+ @Override
+ protected JobShuffleContext getJobShuffleContextImpl(
+ JobID jobID, TierShuffleHandler tierShuffleHandler) {
+ return new JobShuffleContext() {
+ @Override
+ public JobID getJobId() {
+ return jobID;
+ }
+
+ @Override
+ public CompletableFuture<?> stopTrackingAndReleasePartitions(
+ Collection<ResultPartitionID> resultPartitionIds) {
+ return tierShuffleHandler.onReleasePartitions(
+ resultPartitionIds.stream()
+ .map(TieredStorageIdMappingUtils::convertId)
+ .collect(Collectors.toList()));
+ }
+
+ @Override
+ public CompletableFuture<Collection<PartitionWithMetrics>>
getPartitionWithMetrics(
+ Duration duration, Set<ResultPartitionID> set) {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ @Override
+ public void notifyPartitionRecoveryStarted() {}
+ };
+ }
+}
diff --git
a/client-flink/flink-2.3/src/main/java/org/apache/celeborn/plugin/flink/tiered/CelebornTierFactory.java
b/client-flink/flink-2.3/src/main/java/org/apache/celeborn/plugin/flink/tiered/CelebornTierFactory.java
new file mode 100644
index 0000000000..5adf8adde6
--- /dev/null
+++
b/client-flink/flink-2.3/src/main/java/org/apache/celeborn/plugin/flink/tiered/CelebornTierFactory.java
@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.plugin.flink.tiered;
+
+import java.util.List;
+import java.util.concurrent.ScheduledExecutorService;
+
+import javax.annotation.Nullable;
+
+import org.apache.flink.runtime.io.disk.BatchShuffleReadBufferPool;
+import org.apache.flink.runtime.io.network.buffer.BufferCompressor;
+import
org.apache.flink.runtime.io.network.partition.hybrid.tiered.common.TieredStoragePartitionId;
+import
org.apache.flink.runtime.io.network.partition.hybrid.tiered.netty.TieredStorageNettyService;
+import
org.apache.flink.runtime.io.network.partition.hybrid.tiered.storage.TieredStorageConsumerSpec;
+import
org.apache.flink.runtime.io.network.partition.hybrid.tiered.storage.TieredStorageMemoryManager;
+import
org.apache.flink.runtime.io.network.partition.hybrid.tiered.storage.TieredStorageResourceRegistry;
+import
org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.TierConsumerAgent;
+import
org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.TierFactory;
+import
org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.TierProducerAgent;
+import
org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.TierShuffleDescriptor;
+
+/**
+ * The factory class of the Celeborn client, used as a tier of flink hybrid
shuffle tiered storage.
+ */
+public class CelebornTierFactory extends BaseCelebornTierFactory implements
TierFactory {
+
+ @Override
+ public TierProducerAgent createProducerAgent(
+ int numPartitions,
+ int numSubpartitions,
+ TieredStoragePartitionId partitionId,
+ String dataFileBasePath,
+ boolean isBroadcastOnly,
+ TieredStorageMemoryManager storageMemoryManager,
+ TieredStorageNettyService nettyService,
+ TieredStorageResourceRegistry resourceRegistry,
+ BatchShuffleReadBufferPool bufferPool,
+ ScheduledExecutorService ioExecutor,
+ List<TierShuffleDescriptor> shuffleDescriptors,
+ int maxRequestedBuffers,
+ @Nullable BufferCompressor bufferCompressor) {
+ return new CelebornTierProducerAgent(
+ conf,
+ partitionId,
+ numPartitions,
+ numSubpartitions,
+ MAX_BYTES_PER_SEGMENT,
+ bufferSizeBytes,
+ storageMemoryManager,
+ resourceRegistry,
+ shuffleDescriptors);
+ }
+
+ @Override
+ public TierConsumerAgent createConsumerAgent(
+ List<TieredStorageConsumerSpec> tieredStorageConsumerSpecs,
+ List<TierShuffleDescriptor> shuffleDescriptors,
+ TieredStorageNettyService nettyService) {
+ return new CelebornTierConsumerAgent(
+ conf, tieredStorageConsumerSpecs, shuffleDescriptors, bufferSizeBytes);
+ }
+}
diff --git
a/client-flink/flink-2.3/src/main/resources/META-INF/services/org.apache.celeborn.plugin.flink.CelebornFlinkShim
b/client-flink/flink-2.3/src/main/resources/META-INF/services/org.apache.celeborn.plugin.flink.CelebornFlinkShim
new file mode 100644
index 0000000000..4fd0a2213c
--- /dev/null
+++
b/client-flink/flink-2.3/src/main/resources/META-INF/services/org.apache.celeborn.plugin.flink.CelebornFlinkShim
@@ -0,0 +1,16 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+org.apache.celeborn.plugin.flink.CelebornFlinkShimV23
\ No newline at end of file
diff --git
a/client-flink/flink-2.3/src/test/java/org/apache/celeborn/plugin/flink/RemoteShuffleMasterSuiteJV23.java
b/client-flink/flink-2.3/src/test/java/org/apache/celeborn/plugin/flink/RemoteShuffleMasterSuiteJV23.java
new file mode 100644
index 0000000000..29745bdb69
--- /dev/null
+++
b/client-flink/flink-2.3/src/test/java/org/apache/celeborn/plugin/flink/RemoteShuffleMasterSuiteJV23.java
@@ -0,0 +1,20 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.plugin.flink;
+
+public class RemoteShuffleMasterSuiteJV23 extends RemoteShuffleMasterSuiteJ {}
diff --git
a/client-flink/flink-2.3/src/test/java/org/apache/celeborn/plugin/flink/RemoteShuffleResultPartitionFactorySuiteJV23.java
b/client-flink/flink-2.3/src/test/java/org/apache/celeborn/plugin/flink/RemoteShuffleResultPartitionFactorySuiteJV23.java
new file mode 100644
index 0000000000..0686443798
--- /dev/null
+++
b/client-flink/flink-2.3/src/test/java/org/apache/celeborn/plugin/flink/RemoteShuffleResultPartitionFactorySuiteJV23.java
@@ -0,0 +1,22 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.plugin.flink;
+
+/** Tests for {@link RemoteShuffleResultPartitionFactory}. */
+public class RemoteShuffleResultPartitionFactorySuiteJV23
+ extends RemoteShuffleResultPartitionFactorySuiteJ {}
diff --git
a/client-flink/flink-2.3/src/test/java/org/apache/celeborn/plugin/flink/RemoteShuffleResultPartitionSuiteJV23.java
b/client-flink/flink-2.3/src/test/java/org/apache/celeborn/plugin/flink/RemoteShuffleResultPartitionSuiteJV23.java
new file mode 100644
index 0000000000..db6a96a90f
--- /dev/null
+++
b/client-flink/flink-2.3/src/test/java/org/apache/celeborn/plugin/flink/RemoteShuffleResultPartitionSuiteJV23.java
@@ -0,0 +1,20 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.plugin.flink;
+
+public class RemoteShuffleResultPartitionSuiteJV23 extends
RemoteShuffleResultPartitionSuiteJ {}
diff --git
a/client-flink/flink-2.3/src/test/java/org/apache/celeborn/plugin/flink/RemoteShuffleServiceFactorySuiteJV23.java
b/client-flink/flink-2.3/src/test/java/org/apache/celeborn/plugin/flink/RemoteShuffleServiceFactorySuiteJV23.java
new file mode 100644
index 0000000000..e7da8e1196
--- /dev/null
+++
b/client-flink/flink-2.3/src/test/java/org/apache/celeborn/plugin/flink/RemoteShuffleServiceFactorySuiteJV23.java
@@ -0,0 +1,20 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.plugin.flink;
+
+public class RemoteShuffleServiceFactorySuiteJV23 extends
RemoteShuffleServiceFactorySuiteJ {}
diff --git
a/client-flink/flink-2.3/src/test/java/org/apache/celeborn/plugin/flink/ShuffleResourceTrackerSuiteJV23.java
b/client-flink/flink-2.3/src/test/java/org/apache/celeborn/plugin/flink/ShuffleResourceTrackerSuiteJV23.java
new file mode 100644
index 0000000000..30581094cc
--- /dev/null
+++
b/client-flink/flink-2.3/src/test/java/org/apache/celeborn/plugin/flink/ShuffleResourceTrackerSuiteJV23.java
@@ -0,0 +1,20 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.plugin.flink;
+
+public class ShuffleResourceTrackerSuiteJV23 extends
ShuffleResourceTrackerSuiteJ {}
diff --git
a/client-flink/flink-2.3/src/test/java/org/apache/celeborn/plugin/flink/tiered/CelebornTierMasterAgentSuiteJV23.java
b/client-flink/flink-2.3/src/test/java/org/apache/celeborn/plugin/flink/tiered/CelebornTierMasterAgentSuiteJV23.java
new file mode 100644
index 0000000000..f6bb1d2255
--- /dev/null
+++
b/client-flink/flink-2.3/src/test/java/org/apache/celeborn/plugin/flink/tiered/CelebornTierMasterAgentSuiteJV23.java
@@ -0,0 +1,20 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.plugin.flink.tiered;
+
+public class CelebornTierMasterAgentSuiteJV23 extends
CelebornTierMasterAgentSuiteJ {}
diff --git a/dev/dependencies.sh b/dev/dependencies.sh
index fe755e8acb..832be53f98 100755
--- a/dev/dependencies.sh
+++ b/dev/dependencies.sh
@@ -196,6 +196,10 @@ case "$MODULE" in
MVN_MODULES="client-flink/flink-2.2"
SBT_PROJECT="celeborn-client-flink-2_2"
;;
+ "flink-2.3")
+ MVN_MODULES="client-flink/flink-2.3"
+ SBT_PROJECT="celeborn-client-flink-2_3"
+ ;;
"mr")
MVN_MODULES="client-mr/mr"
SBT_PROJECT="celeborn-client-mr"
diff --git a/dev/deps/dependencies-client-flink-2.3
b/dev/deps/dependencies-client-flink-2.3
new file mode 100644
index 0000000000..b06979be85
--- /dev/null
+++ b/dev/deps/dependencies-client-flink-2.3
@@ -0,0 +1,82 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+RoaringBitmap/1.0.6//RoaringBitmap-1.0.6.jar
+commons-crypto/1.0.0//commons-crypto-1.0.0.jar
+commons-io/2.17.0//commons-io-2.17.0.jar
+commons-lang3/3.17.0//commons-lang3-3.17.0.jar
+commons-logging/1.1.3//commons-logging-1.1.3.jar
+failureaccess/1.0.2//failureaccess-1.0.2.jar
+guava/33.1.0-jre//guava-33.1.0-jre.jar
+hadoop-client-api/3.3.6//hadoop-client-api-3.3.6.jar
+hadoop-client-runtime/3.3.6//hadoop-client-runtime-3.3.6.jar
+jackson-annotations/2.15.3//jackson-annotations-2.15.3.jar
+jackson-core/2.15.3//jackson-core-2.15.3.jar
+jackson-databind/2.15.3//jackson-databind-2.15.3.jar
+jackson-module-scala_2.12/2.15.3//jackson-module-scala_2.12-2.15.3.jar
+jcl-over-slf4j/1.7.36//jcl-over-slf4j-1.7.36.jar
+jsr305/1.3.9//jsr305-1.3.9.jar
+jul-to-slf4j/1.7.36//jul-to-slf4j-1.7.36.jar
+leveldbjni-all/1.8//leveldbjni-all-1.8.jar
+lz4-java/1.10.4//lz4-java-1.10.4.jar
+metrics-core/4.2.25//metrics-core-4.2.25.jar
+metrics-graphite/4.2.25//metrics-graphite-4.2.25.jar
+metrics-jvm/4.2.25//metrics-jvm-4.2.25.jar
+netty-all/4.2.10.Final//netty-all-4.2.10.Final.jar
+netty-buffer/4.2.10.Final//netty-buffer-4.2.10.Final.jar
+netty-codec-base/4.2.10.Final//netty-codec-base-4.2.10.Final.jar
+netty-codec-classes-quic/4.2.10.Final//netty-codec-classes-quic-4.2.10.Final.jar
+netty-codec-compression/4.2.10.Final//netty-codec-compression-4.2.10.Final.jar
+netty-codec-dns/4.2.10.Final//netty-codec-dns-4.2.10.Final.jar
+netty-codec-http/4.2.10.Final//netty-codec-http-4.2.10.Final.jar
+netty-codec-http2/4.2.10.Final//netty-codec-http2-4.2.10.Final.jar
+netty-codec-http3/4.2.10.Final//netty-codec-http3-4.2.10.Final.jar
+netty-codec-marshalling/4.2.10.Final//netty-codec-marshalling-4.2.10.Final.jar
+netty-codec-native-quic/4.2.10.Final/linux-aarch_64/netty-codec-native-quic-4.2.10.Final-linux-aarch_64.jar
+netty-codec-native-quic/4.2.10.Final/linux-x86_64/netty-codec-native-quic-4.2.10.Final-linux-x86_64.jar
+netty-codec-native-quic/4.2.10.Final/osx-aarch_64/netty-codec-native-quic-4.2.10.Final-osx-aarch_64.jar
+netty-codec-native-quic/4.2.10.Final/osx-x86_64/netty-codec-native-quic-4.2.10.Final-osx-x86_64.jar
+netty-codec-native-quic/4.2.10.Final/windows-x86_64/netty-codec-native-quic-4.2.10.Final-windows-x86_64.jar
+netty-codec-protobuf/4.2.10.Final//netty-codec-protobuf-4.2.10.Final.jar
+netty-codec-socks/4.2.10.Final//netty-codec-socks-4.2.10.Final.jar
+netty-codec/4.2.10.Final//netty-codec-4.2.10.Final.jar
+netty-common/4.2.10.Final//netty-common-4.2.10.Final.jar
+netty-handler-proxy/4.2.10.Final//netty-handler-proxy-4.2.10.Final.jar
+netty-handler/4.2.10.Final//netty-handler-4.2.10.Final.jar
+netty-resolver-dns/4.2.10.Final//netty-resolver-dns-4.2.10.Final.jar
+netty-resolver/4.2.10.Final//netty-resolver-4.2.10.Final.jar
+netty-transport-classes-epoll/4.2.10.Final//netty-transport-classes-epoll-4.2.10.Final.jar
+netty-transport-classes-io_uring/4.2.10.Final//netty-transport-classes-io_uring-4.2.10.Final.jar
+netty-transport-classes-kqueue/4.2.10.Final//netty-transport-classes-kqueue-4.2.10.Final.jar
+netty-transport-native-epoll/4.2.10.Final/linux-aarch_64/netty-transport-native-epoll-4.2.10.Final-linux-aarch_64.jar
+netty-transport-native-epoll/4.2.10.Final/linux-riscv64/netty-transport-native-epoll-4.2.10.Final-linux-riscv64.jar
+netty-transport-native-epoll/4.2.10.Final/linux-x86_64/netty-transport-native-epoll-4.2.10.Final-linux-x86_64.jar
+netty-transport-native-io_uring/4.2.10.Final/linux-aarch_64/netty-transport-native-io_uring-4.2.10.Final-linux-aarch_64.jar
+netty-transport-native-io_uring/4.2.10.Final/linux-riscv64/netty-transport-native-io_uring-4.2.10.Final-linux-riscv64.jar
+netty-transport-native-io_uring/4.2.10.Final/linux-x86_64/netty-transport-native-io_uring-4.2.10.Final-linux-x86_64.jar
+netty-transport-native-kqueue/4.2.10.Final/osx-aarch_64/netty-transport-native-kqueue-4.2.10.Final-osx-aarch_64.jar
+netty-transport-native-kqueue/4.2.10.Final/osx-x86_64/netty-transport-native-kqueue-4.2.10.Final-osx-x86_64.jar
+netty-transport-native-unix-common/4.2.10.Final//netty-transport-native-unix-common-4.2.10.Final.jar
+netty-transport/4.2.10.Final//netty-transport-4.2.10.Final.jar
+paranamer/2.8//paranamer-2.8.jar
+protobuf-java/3.25.5//protobuf-java-3.25.5.jar
+scala-library/2.12.18//scala-library-2.12.18.jar
+scala-reflect/2.12.18//scala-reflect-2.12.18.jar
+slf4j-api/1.7.36//slf4j-api-1.7.36.jar
+snakeyaml/2.2//snakeyaml-2.2.jar
+snappy-java/1.1.10.5//snappy-java-1.1.10.5.jar
+zstd-jni/1.5.7-1//zstd-jni-1.5.7-1.jar
diff --git a/dev/reformat b/dev/reformat
index 5a1b11370c..53aee8d347 100755
--- a/dev/reformat
+++ b/dev/reformat
@@ -29,6 +29,7 @@ else
${PROJECT_DIR}/build/mvn spotless:apply -Pflink-2.0
${PROJECT_DIR}/build/mvn spotless:apply -Pflink-2.1
${PROJECT_DIR}/build/mvn spotless:apply -Pflink-2.2
+ ${PROJECT_DIR}/build/mvn spotless:apply -Pflink-2.3
${PROJECT_DIR}/build/mvn spotless:apply -Pspark-3.3
${PROJECT_DIR}/build/mvn spotless:apply -Pspark-3.5
${PROJECT_DIR}/build/mvn spotless:apply -Pspark-4.0
diff --git a/docs/README.md b/docs/README.md
index 4879014f70..ae144bcc74 100644
--- a/docs/README.md
+++ b/docs/README.md
@@ -122,7 +122,7 @@ INFO [async-reply] Controller: CommitFiles for
local-1690000152711-0 success wit
**Important: Only Flink batch jobs are supported for now.**
#### Copy Celeborn Client to Flink's lib
-Celeborn release binary contains clients for Flink 1.18.x, Flink 1.19.x, Flink
1.20.x, Flink 2.0.x, Flink 2.1.x, Flink 2.2.x, copy the corresponding client
jar into Flink's
+Celeborn release binary contains clients for Flink 1.18.x, Flink 1.19.x, Flink
1.20.x, Flink 2.0.x, Flink 2.1.x, Flink 2.2.x, Flink 2.3.x, copy the
corresponding client jar into Flink's
`lib/` directory:
```shell
cp
$CELEBORN_HOME/flink/celeborn-client-flink-<flink.version>-shaded_<scala.binary.version>-<celeborn.version>.jar
$FLINK_HOME/lib/
@@ -136,7 +136,7 @@ cd $FLINK_HOME
vi conf/flink-conf.yaml
```
-- Flink 1.19.x, Flink 1.20.x, Flink 2.0.x, Flink 2.1.x, Flink 2.2.x
+- Flink 1.19.x, Flink 1.20.x, Flink 2.0.x, Flink 2.1.x, Flink 2.2.x, Flink
2.3.x
```shell
cd $FLINK_HOME
vi conf/config.yaml
diff --git a/docs/deploy.md b/docs/deploy.md
index 09a831928b..25f1f78ce6 100644
--- a/docs/deploy.md
+++ b/docs/deploy.md
@@ -212,7 +212,7 @@ spark.executor.userClassPathFirst false
**Important: Only Flink batch jobs are supported for now. Due to the Shuffle
Service in Flink is cluster-granularity, if you want to use Celeborn in a
session cluster, it will not be able to submit both streaming and batch job to
the same cluster. We plan to get rid of this restriction for Hybrid Shuffle
mode in a future release.**
-Celeborn release binary contains clients for Flink 1.18.x, Flink 1.19.x, Flink
1.20.x, Flink 2.0.x, Flink 2.1.x, Flink 2.2.x, copy the corresponding client
jar into Flink's
+Celeborn release binary contains clients for Flink 1.18.x, Flink 1.19.x, Flink
1.20.x, Flink 2.0.x, Flink 2.1.x, Flink 2.2.x, Flink 2.3.x, copy the
corresponding client jar into Flink's
`lib/` directory:
Copy
`$CELEBORN_HOME/flink/celeborn-client-flink-<flink.version>-shaded_<scala.binary.version>-<celeborn.version>.jar`
to `$FLINK_HOME/lib/`.
diff --git a/docs/developers/overview.md b/docs/developers/overview.md
index f5a0543843..950dbb49a7 100644
--- a/docs/developers/overview.md
+++ b/docs/developers/overview.md
@@ -89,7 +89,7 @@ Celeborn's primary components(i.e. Master, Worker, Client)
are engine irrelevant
and easy to implement plugins for various engines.
Currently, Celeborn officially supports
[Spark](https://spark.apache.org/)(both Spark 2.x, Spark 3.x and Spark 4.x),
-[Flink](https://flink.apache.org/)(1.18/1.19/1.20/2.0/2.1/2.2),
+[Flink](https://flink.apache.org/)(1.18/1.19/1.20/2.0/2.1/2.2/2.3),
[Gluten](https://github.com/apache/gluten) and
[Auron](https://github.com/apache/auron). Also, developers are integrating
Celeborn with other engines,
for example [MR3](https://mr3docs.datamonad.com/docs/mr3/).
diff --git a/docs/developers/sbt.md b/docs/developers/sbt.md
index 7cad08de5a..1c0c77c10d 100644
--- a/docs/developers/sbt.md
+++ b/docs/developers/sbt.md
@@ -42,6 +42,7 @@ The following table indicates the compatibility of Celeborn
Spark and Flink clie
| Flink 2.0 | ❌ | ❌ | ✔ |
✔ | ❌ | ✔ | ✔
|
| Flink 2.1 | ❌ | ❌ | ✔ |
✔ | ❌ | ✔ | ✔
|
| Flink 2.2 | ❌ | ❌ | ✔ |
✔ | ❌ | ✔ | ✔
|
+| Flink 2.3 | ❌ | ❌ | ✔ |
✔ | ❌ | ✔ | ✔
|
## Useful SBT commands
diff --git a/pom.xml b/pom.xml
index b2e1a9cacd..ad47cb9f23 100644
--- a/pom.xml
+++ b/pom.xml
@@ -1902,6 +1902,26 @@
</properties>
</profile>
+ <profile>
+ <id>flink-2.3</id>
+ <modules>
+ <module>client-flink/common</module>
+ <module>client-flink/common-tiered</module>
+ <module>client-flink/flink-2.3</module>
+ <module>client-flink/flink-2.3-shaded</module>
+ <module>tests/flink-it</module>
+ </modules>
+ <properties>
+ <flink.version>2.3.0</flink.version>
+ <flink.binary.version>2.3</flink.binary.version>
+ <scala.binary.version>2.12</scala.binary.version>
+
<celeborn.flink.plugin.artifact>celeborn-client-flink-2.3_${scala.binary.version}</celeborn.flink.plugin.artifact>
+
<flink.streaming.artifact>flink-streaming-java</flink.streaming.artifact>
+ <flink.clients.artifact>flink-clients</flink.clients.artifact>
+
<flink.runtime.web.artifact>flink-runtime-web</flink.runtime.web.artifact>
+ </properties>
+ </profile>
+
<profile>
<id>mr</id>
<modules>
diff --git a/project/CelebornBuild.scala b/project/CelebornBuild.scala
index ec82fd1d67..6d497868ce 100644
--- a/project/CelebornBuild.scala
+++ b/project/CelebornBuild.scala
@@ -540,6 +540,7 @@ object Utils {
case Some("flink-2.0") => Some(Flink20)
case Some("flink-2.1") => Some(Flink21)
case Some("flink-2.2") => Some(Flink22)
+ case Some("flink-2.3") => Some(Flink23)
case _ => None
}
@@ -1281,6 +1282,16 @@ object Flink22 extends FlinkClientProjects {
val flinkClientShadedProjectName: String = "celeborn-client-flink-2_2-shaded"
}
+object Flink23 extends FlinkClientProjects {
+ val flinkVersion = "2.3.0"
+
+ // note that SBT does not allow using the period symbol (.) in project names.
+ val flinkClientProjectPath = "client-flink/flink-2.3"
+ val flinkClientProjectName = "celeborn-client-flink-2_3"
+ val flinkClientShadedProjectPath: String = "client-flink/flink-2.3-shaded"
+ val flinkClientShadedProjectName: String = "celeborn-client-flink-2_3-shaded"
+}
+
trait FlinkClientProjects {
val flinkVersion: String
diff --git
a/tests/flink-it/src/test/java/org/apache/celeborn/tests/flink/FlinkVersion.java
b/tests/flink-it/src/test/java/org/apache/celeborn/tests/flink/FlinkVersion.java
index 43fee92f69..ab1246c792 100644
---
a/tests/flink-it/src/test/java/org/apache/celeborn/tests/flink/FlinkVersion.java
+++
b/tests/flink-it/src/test/java/org/apache/celeborn/tests/flink/FlinkVersion.java
@@ -28,7 +28,8 @@ public enum FlinkVersion {
v1_20("1.20"),
v2_0("2.0"),
v2_1("2.1"),
- v2_2("2.2");
+ v2_2("2.2"),
+ v2_3("2.3");
private final String versionStr;
@@ -50,6 +51,8 @@ public enum FlinkVersion {
return v2_1;
case "2.2":
return v2_2;
+ case "2.3":
+ return v2_3;
default:
throw new IllegalArgumentException("Unsupported flink version: " +
versionStr);
}
