This is an automated email from the ASF dual-hosted git repository.
RexXiong 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 f43f424c04 [CELEBORN-2471] Fix Flink uncompressed shuffle failing to
load codec NONE
f43f424c04 is described below
commit f43f424c04c4259a18251987965ddb25745e28e4
Author: senthh <[email protected]>
AuthorDate: Fri Oct 9 10:13:32 2026 +0800
[CELEBORN-2471] Fix Flink uncompressed shuffle failing to load codec NONE
### What changes were proposed in this pull request?
Celeborn lets you turn off shuffle compression for Flink by setting the
codec to `NONE` (uncompressed shuffle).
**Bug:** when the codec is `NONE`, the read side of the shuffle passes the
literal string `"NONE"` to CelebornFlinkShim.createBufferDecompressor, which
builds a Flink `BufferDecompressor`. Flink resolves that value as a class name
(`Class.forName("NONE")`) and throws `ClassNotFoundException: NONE`, so the
Flink job fails to start. Uncompressed shuffle was unusable.
**Fix:** mirror the write-side NONE guard into the read path — when the
codec is `NONE`, pass a null decompressor instead of constructing one. This is
safe because `RemoteShuffleInputGateDelegation.decompressBufferIfNeeded` only
dereferences the decompressor inside `if (buffer.isCompressed())`, and NONE
writes never produce compressed buffers.
### Why are the changes needed?
The write and read paths handled the `NONE` sentinel asymmetrically:
- Write side (`RemoteShuffleResultPartitionFactory.getBufferCompressor`):
correctly guards `NONE` and returns a null compressor (no compression).
- Read side (`RemoteShuffleInputGateFactory.create` / `createInputGate`):
did not guard `NONE`, so it passed the literal string to Flink's class
resolution and crashed.
A simple mismatch — one half of the code knew what `NONE` meant, the other
half didn't.
### Does this PR resolve a correctness bug?
- [x] Yes
### Does this PR introduce _any_ user-facing change?
- [x] Yes
### How was this patch tested?
- Compiled the affected module: `mvn -Pflink-1.18 compile -pl
client-flink/common -am` → BUILD SUCCESS (covers the shared factory used by all
Flink modules).
- Full end-to-end repro (`TestCelebornFlinkShuffle > Uncompressed shuffle
produces the correct result`) to be run on a cluster once the shaded Flink jar
is rebuilt and redeployed.
Fixes CELEBORN-2471.
Closes #3861 from senthh/CELEBORN-2471.
Authored-by: senthh <[email protected]>
Signed-off-by: Shuang <[email protected]>
---
.../celeborn/plugin/flink/RemoteShuffleInputGateFactory.java | 11 ++++++++---
1 file changed, 8 insertions(+), 3 deletions(-)
diff --git
a/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleInputGateFactory.java
b/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleInputGateFactory.java
index 71b2a41b80..3c55bafd43 100644
---
a/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleInputGateFactory.java
+++
b/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleInputGateFactory.java
@@ -34,6 +34,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.celeborn.common.CelebornConf;
+import org.apache.celeborn.common.protocol.CompressionCodec;
import org.apache.celeborn.plugin.flink.utils.Utils;
/** Factory class to create {@link RemoteShuffleInputGate}. */
@@ -96,9 +97,11 @@ public class RemoteShuffleInputGateFactory {
SupplierWithException<BufferPool, IOException> bufferPoolFactory =
createBufferPoolFactory(networkBufferPool, numBuffersPerGate,
supportFloatingBuffers);
+ String compressionCodec = celebornConf.shuffleCompressionCodec().name();
BufferDecompressor bufferDecompressor =
- CelebornFlinkShim.createBufferDecompressor(
- networkBufferSize, celebornConf.shuffleCompressionCodec().name());
+ CompressionCodec.NONE.name().equals(compressionCodec)
+ ? null
+ : CelebornFlinkShim.createBufferDecompressor(networkBufferSize,
compressionCodec);
return new RemoteShuffleInputGate(
celebornConf,
ownerContext,
@@ -118,7 +121,9 @@ public class RemoteShuffleInputGateFactory {
String compressionCodec,
Map<Integer, ShuffleIOMetricGroup> shuffleIOMetricGroupMap) {
BufferDecompressor bufferDecompressor =
- CelebornFlinkShim.createBufferDecompressor(networkBufferSize,
compressionCodec);
+ CompressionCodec.NONE.name().equals(compressionCodec)
+ ? null
+ : CelebornFlinkShim.createBufferDecompressor(networkBufferSize,
compressionCodec);
return new RemoteShuffleInputGate(
celebornConf,
ownerContext,
