[#2093][followup] feat: Add support of partition split for grpc (#2396)

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

Add support of partition split for grpc

### Why are the changes needed?

followup for #2093 

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

No.

### How was this patch tested?

Existing unit tests

Author: zuston

client-spark/common/src/main/java/org/apache/uniffle/shuffle/manager/RssShuffleManagerBase.java

@@ -271,6 +271,7 @@ public RssShuffleManagerBase(SparkConf conf, boolean isDriver) {
         throw new RssException(
             "The feature of task partition reassign is incompatible with multiple replicas mechanism.");
       }
+      LOG.info("Partition reassign is enabled.");
     }
     this.blockIdSelfManagedEnabled = rssConf.getBoolean(RSS_BLOCK_ID_SELF_MANAGEMENT_ENABLED);
     this.shuffleManagerRpcServiceEnabled =

internal-client/src/main/java/org/apache/uniffle/client/impl/grpc/ShuffleServerGrpcClient.java

@@ -23,11 +23,13 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Random;
+import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
 import com.google.protobuf.ByteString;
 import com.google.protobuf.UnsafeByteOperations;
 import io.grpc.StatusRuntimeException;
@@ -540,7 +542,7 @@ public RssSendShuffleDataResponse sendShuffleData(RssSendShuffleDataRequest requ
 
     boolean isSuccessful = true;
     AtomicReference<StatusCode> failedStatusCode = new AtomicReference<>(StatusCode.INTERNAL_ERROR);
-
+    Set<Integer> needSplitPartitionIds = Sets.newHashSet();
     // prepare rpc request based on shuffleId -> partitionId -> blocks
     for (Map.Entry<Integer, Map<Integer, List<ShuffleBlockInfo>>> stb :
         shuffleIdToBlocks.entrySet()) {
@@ -583,17 +585,18 @@ public RssSendShuffleDataResponse sendShuffleData(RssSendShuffleDataRequest requ
         RetryUtils.retryWithCondition(
             () -> {
               // TODO(baoloongmao): support partition split follow netty client
-              long requireId =
+              Pair<Long, List<Integer>> allocationResult =
                   requirePreAllocation(
-                          appId,
-                          shuffleId,
-                          partitionIds,
-                          partitionRequireSizes,
-                          allocateSize,
-                          request.getRetryMax() / maxRetryAttempts,
-                          request.getRetryIntervalMax(),
-                          failedStatusCode)
-                      .getLeft();
+                      appId,
+                      shuffleId,
+                      partitionIds,
+                      partitionRequireSizes,
+                      allocateSize,
+                      request.getRetryMax() / maxRetryAttempts,
+                      request.getRetryIntervalMax(),
+                      failedStatusCode);
+              long requireId = allocationResult.getLeft();
+              needSplitPartitionIds.addAll(allocationResult.getRight());
               if (requireId == FAILED_REQUIRE_ID) {
                 throw new RssException(
                     String.format(
@@ -661,6 +664,7 @@ public RssSendShuffleDataResponse sendShuffleData(RssSendShuffleDataRequest requ
     } else {
       response = new RssSendShuffleDataResponse(failedStatusCode.get());
     }
+    response.setNeedSplitPartitionIds(needSplitPartitionIds);
     return response;
   }