[#2750] feat(spark): Support Spark 4.1 (#2751)

### What changes were proposed in this pull request?

Add a `spark4.1` Maven profile that reuses the existing `client-spark/spark4` module to compile and run against **Spark 4.1.1**.

#### API shims

Spark 4.1 introduced two binary-incompatible additions to APIs called from this module:

- `org.apache.spark.scheduler.MapStatus.apply` — gained a trailing `checksumVal: Long` parameter.
- `org.apache.spark.util.collection.ExternalSorter` — constructor gained a trailing `rowBasedChecksums: Array[RowBasedChecksum]` parameter.

Scala default parameters do not surface to Java callers, so a single `Spark4Compat.java` cannot satisfy both 4.0.x and 4.1.x signatures. Two parallel source roots ship a matching `Spark4Compat`:

- `src/main/java-spark4_0/` — 3-arg `mapStatus`, 5-arg `ExternalSorter` ctor (current Spark 4.0 shape).
- `src/main/java-spark4_1/` — 4-arg `mapStatus` (`checksumVal = 0L`), 6-arg `ExternalSorter` ctor (empty `RowBasedChecksum[]`).

`build-helper-maven-plugin` selects the source root via `${spark4.compat.source.dir}`; the existing `spark4` profile keeps the 4.0 default, the new `spark4.1` profile overrides it to the 4.1 variant.

`RssShuffleReader` now calls `Spark4Compat.newExternalSorter(...)` and `RssShuffleWriter` calls `Spark4Compat.mapStatus(...)` instead of constructing those Spark types directly.

The `client-spark/extension` (Spark UI) module already uses `scala-jakarta` under the `spark4` profile; Spark 4.1's `WebUIPage.render` keeps the same `jakarta.servlet.http.HttpServletRequest` signature, so the existing source root works for `spark4.1` as well — only a `spark4.1` profile body that mirrors the `spark4` one is added.

`spark4` and `spark4.1` are mutually exclusive at build time; they share a Maven coordinate but produce incompatible bytecode against different Spark majors. Run `mvn clean` between profile switches locally.

#### Runtime dependency adjustments under `-Pspark4.1`

Spark 4.1.1 bumped several runtime libraries; the `spark4.1` profile pins matching versions and the code makes the necessary lifecycle adjustments:

- **Netty 4.1.118 → 4.2.7.Final**. Spark 4.1's `NettyUtils.createEventLoop` references `io.netty.channel.nio.NioIoHandler` (new in Netty 4.2). Netty 4.2 also tightens `PooledByteBuf.nioBuffer()` to require `refCnt > 0`. Response decoders that wrap the frame buffer in a `NettyManagedBuffer` body now call `retain()` explicitly, so the body's reference count is independent of the frame's. `TransportFrameDecoder.shouldRelease` is simplified to always return true, and `TransportFrameDecoderTest` is updated accordingly. Netty 4.1 silently tolerated the previous shape; 4.2 throws `IllegalReferenceCountException`.
- **Jetty 9.3.24 → 9.4.53.v20231009**. Aligns with Spark 4.1's bundled Jetty. `JettyServer.createThreadPool` is already source-compatible with both lines.
- **Jackson 2.18.2 → 2.20.0**. Spark 4.1.1 ships `jackson-module-scala_2.13:2.20.0`, which validates the classpath `jackson-databind` version on registration and rejects anything outside `[2.20.0, 2.21.0)`. Without the bump, `RDDOperationScope`'s static initializer (`ObjectMapper.registerModule(DefaultScalaModule)`) throws `JsonMappingException`, surfacing as `ExceptionInInitializerError` on the first SQL/RDD-touching test and `NoClassDefFoundError` on every later one.

These adjustments live entirely inside the `spark4.1` profile and the response-decoder ref-count fix; the default build and `-Pspark4` (Spark 4.0.2) paths are unchanged.

### Why are the changes needed?

Following #1805 (Spark 4.0.2 support, now closed), users on Spark 4.1 currently cannot link Uniffle's Spark 4 client because of the API additions above. Closes #2750.

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

No (build / packaging only). A new `-Pspark4.1` build flag is available; default builds and the existing `-Pspark4` flag are unchanged.

### How was this patch tested?

- Local: `mvn clean package -Pspark4.1 -DskipTests` against Spark 4.1.1 — passes.
- Local: `mvn clean package -Pspark4 -DskipTests` against Spark 4.0.2 — still passes.
- Verified the resulting `Spark4Compat.class` differs across the two profiles (different bytecode size / arity), confirming the multi-source-root selection works.
- Verified API signatures via `javap -p` on Spark 4.0.2 / 4.1.1 jars from Maven Central match the shim implementations.
- Local: `mvn -Pspark4.1 -pl integration-test/spark4 -am test` — `AQERepartitionTest` and `MapSideCombineTest` pass after the netty refCnt + jackson alignment fixes.
- Local: `TransportFrameDecoderTest` passes against Netty 4.2 (`-Pspark4.1`) and continues to pass on the default Netty 4.1 line.
- CI: `parallel.yml` now runs `-Pspark4.1` (java-version 17) in the matrix; `sequential.yml` adds an `Execute -Pspark4.1` step gated on java-version 17. Latest run on this branch ([26386496434](https://github.com/apache/uniffle/actions/runs/26386496434)) is **all-green** across the full 45-job matrix.

Author: LuciferYang

.github/workflows/parallel.yml

@@ -77,6 +77,8 @@ jobs:
             java-version: 8
           - profile: spark4
             java-version: 17
+          - profile: spark4.1
+            java-version: 17
           - profile: mr-hadoop2.8
             java-version: 8
           - profile: mr-hadoop3.2

.github/workflows/sequential.yml

@@ -72,6 +72,10 @@ jobs:
       if: inputs.java-version == '17'
       run: ./mvnw -B -fae ${{ inputs.maven-args }} -Pspark4 | tee -a /tmp/maven.log;
       shell: bash
+    - name: Execute `./mvnw ${{ inputs.maven-args }} -Pspark4.1`
+      if: inputs.java-version == '17'
+      run: ./mvnw -B -fae ${{ inputs.maven-args }} -Pspark4.1 | tee -a /tmp/maven.log;
+      shell: bash
     - name: Execute `./mvnw ${{ inputs.maven-args }} -Pspark3`
       run: ./mvnw -B -fae ${{ inputs.maven-args }} -Pspark3 | tee -a /tmp/maven.log;
       shell: bash

client-spark/extension/pom.xml

@@ -136,11 +136,26 @@
 
     <profiles>
         <profile>
-            <!-- Merged with the root pom's spark4 profile by shared id. -->
+            <!--
+                Co-activated with the root pom's spark4 profile via the shared
+                -Pspark4 flag. Each pom contributes its own profile body; Maven
+                does not merge profiles across poms by id.
+            -->
             <id>spark4</id>
             <properties>
                 <extension.servlet.source.dir>src/main/scala-jakarta</extension.servlet.source.dir>
             </properties>
         </profile>
+        <profile>
+            <!--
+                Co-activated with the root pom's spark4.1 profile via the shared
+                -Pspark4.1 flag. Each pom contributes its own profile body; Maven
+                does not merge profiles across poms by id.
+            -->
+            <id>spark4.1</id>
+            <properties>
+                <extension.servlet.source.dir>src/main/scala-jakarta</extension.servlet.source.dir>
+            </properties>
+        </profile>
     </profiles>
 </project>

client-spark/spark4-shaded/pom.xml

@@ -208,6 +208,20 @@
                   <pattern>org.checkerframework</pattern>
                   <shadedPattern>${rss.shade.packageName}.org.checkerframework</shadedPattern>
                 </relocation>
+                <relocation>
+                  <!--
+                      Spark 4.1's hadoop 3.4.1 pulls bcprov-jdk18on transitively,
+                      and netty 4.2 stopped internally shading jctools. Both leak
+                      into the shaded jar; relocate them so checkshade.sh stays
+                      happy and we don't conflict with users' classpath.
+                  -->
+                  <pattern>org.bouncycastle</pattern>
+                  <shadedPattern>${rss.shade.packageName}.org.bouncycastle</shadedPattern>
+                </relocation>
+                <relocation>
+                  <pattern>org.jctools</pattern>
+                  <shadedPattern>${rss.shade.packageName}.org.jctools</shadedPattern>
+                </relocation>
                 <relocation>
                   <pattern>org.eclipse</pattern>
                   <shadedPattern>${rss.shade.packageName}.org.eclipse</shadedPattern>

client-spark/spark4/pom.xml

@@ -31,6 +31,27 @@
     <packaging>jar</packaging>
     <name>Apache Uniffle Client (Spark 4)</name>
 
+    <!--
+        Spark 4.1 added new (defaulted) parameters to MapStatus.apply and to the
+        ExternalSorter constructor (checksumVal, rowBasedChecksums). Scala defaults
+        do not surface to Java callers, so a single Spark4Compat.java cannot
+        satisfy both 4.0 and 4.1 signatures. Two parallel source roots
+        (src/main/java-spark4_0, src/main/java-spark4_1) each ship a Spark4Compat
+        with the matching call shape. build-helper-maven-plugin picks one via
+        ${spark4.compat.source.dir}; the spark4 profile keeps the 4.0 default
+        while the spark4.1 profile in this pom (co-activated with the root
+        pom's spark4.1 profile via the shared -Pspark4.1 flag — each pom
+        contributes its own properties/build sections) switches it to the
+        4.1 variant.
+
+        Keep the public method surface of Spark4Compat in lock-step across both
+        variants; only the bodies differ.
+    -->
+    <properties>
+        <!-- Overridden by the spark4.1 profile below. -->
+        <spark4.compat.source.dir>src/main/java-spark4_0</spark4.compat.source.dir>
+    </properties>
+
     <dependencies>
         <dependency>
             <groupId>com.google.protobuf</groupId>
@@ -107,10 +128,42 @@
 
     <build>
         <plugins>
+            <plugin>
+                <groupId>org.codehaus.mojo</groupId>
+                <artifactId>build-helper-maven-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <id>add-spark4-compat-source</id>
+                        <phase>generate-sources</phase>
+                        <goals>
+                            <goal>add-source</goal>
+                        </goals>
+                        <configuration>
+                            <sources>
+                                <source>${spark4.compat.source.dir}</source>
+                            </sources>
+                        </configuration>
+                    </execution>
+                </executions>
+            </plugin>
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-surefire-plugin</artifactId>
             </plugin>
         </plugins>
     </build>
+
+    <profiles>
+        <profile>
+            <!--
+                Co-activated with the root pom's spark4.1 profile via the shared
+                -Pspark4.1 flag. Each pom contributes its own profile body; Maven
+                does not merge profiles across poms by id.
+            -->
+            <id>spark4.1</id>
+            <properties>
+                <spark4.compat.source.dir>src/main/java-spark4_1</spark4.compat.source.dir>
+            </properties>
+        </profile>
+    </profiles>
 </project>

client-spark/spark4/src/main/java-spark4_0/org/apache/spark/shuffle/compat/Spark4Compat.java

@@ -0,0 +1,54 @@
+/*
+ * 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.spark.shuffle.compat;
+
+import scala.Option;
+import scala.math.Ordering;
+
+import org.apache.spark.Aggregator;
+import org.apache.spark.Partitioner;
+import org.apache.spark.TaskContext;
+import org.apache.spark.scheduler.MapStatus;
+import org.apache.spark.serializer.Serializer;
+import org.apache.spark.storage.BlockManagerId;
+import org.apache.spark.util.collection.ExternalSorter;
+
+/**
+ * Compatibility shim for Spark 4.0.x. Selected by the build via the
+ * {@code src/main/java-spark4_0} source root under the {@code spark4} profile.
+ *
+ * <p>Mirror of {@code java-spark4_1/.../Spark4Compat.java}; keep the public surface in lock-step.
+ */
+public final class Spark4Compat {
+
+  private Spark4Compat() {}
+
+  public static MapStatus mapStatus(
+      BlockManagerId loc, long[] uncompressedSizes, long mapTaskId) {
+    return MapStatus.apply(loc, uncompressedSizes, mapTaskId);
+  }
+
+  public static <K, V, C> ExternalSorter<K, V, C> newExternalSorter(
+      TaskContext context,
+      Option<Aggregator<K, V, C>> aggregator,
+      Option<Partitioner> partitioner,
+      Option<Ordering<K>> ordering,
+      Serializer serializer) {
+    return new ExternalSorter<>(context, aggregator, partitioner, ordering, serializer);
+  }
+}

client-spark/spark4/src/main/java-spark4_1/org/apache/spark/shuffle/compat/Spark4Compat.java

@@ -0,0 +1,62 @@
+/*
+ * 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.spark.shuffle.compat;
+
+import scala.Option;
+import scala.math.Ordering;
+
+import org.apache.spark.Aggregator;
+import org.apache.spark.Partitioner;
+import org.apache.spark.TaskContext;
+import org.apache.spark.scheduler.MapStatus;
+import org.apache.spark.serializer.Serializer;
+import org.apache.spark.shuffle.checksum.RowBasedChecksum;
+import org.apache.spark.storage.BlockManagerId;
+import org.apache.spark.util.collection.ExternalSorter;
+
+/**
+ * Compatibility shim for Spark 4.1.x. Selected by the build via the
+ * {@code src/main/java-spark4_1} source root under the {@code spark4.1} profile.
+ *
+ * <p>Spark 4.1 added a {@code checksumVal} parameter to {@code MapStatus.apply} and a
+ * {@code rowBasedChecksums} parameter to {@link ExternalSorter}'s constructor. Both have Scala
+ * defaults but are required from Java; pass disabled defaults (0L and an empty array).
+ *
+ * <p>Mirror of {@code java-spark4_0/.../Spark4Compat.java}; keep the public surface in lock-step.
+ */
+public final class Spark4Compat {
+
+  private Spark4Compat() {}
+
+  public static MapStatus mapStatus(
+      BlockManagerId loc, long[] uncompressedSizes, long mapTaskId) {
+    return MapStatus.apply(loc, uncompressedSizes, mapTaskId, 0L);
+  }
+
+  public static <K, V, C> ExternalSorter<K, V, C> newExternalSorter(
+      TaskContext context,
+      Option<Aggregator<K, V, C>> aggregator,
+      Option<Partitioner> partitioner,
+      Option<Ordering<K>> ordering,
+      Serializer serializer) {
+    // Spark 4.1's ExternalSorter requires a non-null RowBasedChecksum[]; pass an empty
+    // array to disable row-based checksums.
+    return new ExternalSorter<>(
+        context, aggregator, partitioner, ordering, serializer, new RowBasedChecksum[0]);
+  }
+}

client-spark/spark4/src/main/java/org/apache/spark/shuffle/reader/RssShuffleReader.java

@@ -48,6 +48,7 @@
 import org.apache.spark.shuffle.RssShuffleManager;
 import org.apache.spark.shuffle.RssSparkConfig;
 import org.apache.spark.shuffle.ShuffleReader;
+import org.apache.spark.shuffle.compat.Spark4Compat;
 import org.apache.spark.util.CompletionIterator;
 import org.apache.spark.util.CompletionIterator$;
 import org.apache.spark.util.collection.ExternalSorter;
@@ -223,7 +224,7 @@ public Iterator<Product2<K, C>> read() {
         }
       }
       ExternalSorter<K, Object, C> sorter =
-          new ExternalSorter<>(
+          Spark4Compat.newExternalSorter(
               context, aggregator, Option.empty(), shuffleDependency.keyOrdering(), serializer);
       long startTime = System.currentTimeMillis();
       sorter.insertAll(rssShuffleDataIterator);

client-spark/spark4/src/main/java/org/apache/spark/shuffle/writer/RssShuffleWriter.java

@@ -59,6 +59,7 @@
 import org.apache.spark.shuffle.RssSparkConfig;
 import org.apache.spark.shuffle.RssSparkShuffleUtils;
 import org.apache.spark.shuffle.ShuffleWriter;
+import org.apache.spark.shuffle.compat.Spark4Compat;
 import org.apache.spark.shuffle.handle.ShuffleHandleInfo;
 import org.apache.spark.storage.BlockManagerId;
 import org.roaringbitmap.longlong.Roaring64NavigableMap;
@@ -720,7 +721,8 @@ public Option<MapStatus> stop(boolean success) {
                         ? shuffleTaskStats.encode()
                         : Long.toString(taskAttemptId)));
         MapStatus mapStatus =
-            MapStatus.apply(blockManagerId, shuffleTaskStats.getPartitionLengths(), taskAttemptId);
+            Spark4Compat.mapStatus(
+                blockManagerId, shuffleTaskStats.getPartitionLengths(), taskAttemptId);
         return Option.apply(mapStatus);
       } else {
         return Option.empty();

common/src/main/java/org/apache/uniffle/common/netty/TransportFrameDecoder.java

@@ -23,7 +23,6 @@
 import com.google.common.base.Preconditions;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.CompositeByteBuf;
-import io.netty.buffer.EmptyByteBuf;
 import io.netty.buffer.Unpooled;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
@@ -85,10 +84,10 @@ public void channelRead(ChannelHandlerContext ctx, Object data) {
 
   @VisibleForTesting
   static boolean shouldRelease(Message msg) {
-    if (msg == null || msg.body() == null || msg.body().byteBuf() == null) {
-      return true;
-    }
-    return msg.body().byteBuf() instanceof EmptyByteBuf;
+    // The frame buffer is always released here. Response decoders that wrap the frame
+    // buffer in a NettyManagedBuffer body must call retain() on the buffer themselves
+    // so the body can be consumed independently of the frame's reference count.
+    return true;
   }
 
   private void clear() {