[#2724] refactor(spark): Introduce ReassignExecutor to simplify shuffle writer logic

client-spark/common/src/main/java/org/apache/spark/shuffle/writer/TaskAttemptAssignment.java

@@ -88,7 +88,7 @@ private boolean hasBeenLoadBalanced(int partitionId) {
    * @param exclusiveServers
    * @return
    */
-  public boolean updatePartitionSplitAssignment(
+  public boolean tryNextServerForSplitPartition(
       int partitionId, List<ShuffleServerInfo> exclusiveServers) {
     if (hasBeenLoadBalanced(partitionId)) {
       Set<ShuffleServerInfo> servers =

client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriteBufferManager.java

@@ -678,6 +678,11 @@ protected long getRecordCount() {
     return recordCounter.get();
   }
 
+  @VisibleForTesting
+  protected void resetRecordCount() {
+    recordCounter.set(0);
+  }
+
   public long getBlockCount() {
     return blockCounter.get();
   }