[apache#1668] fix: Replace assertions with exception-throwing codes (apache#1804)

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

Replace assertions with exception-throwing codes to make them work in production.

### Why are the changes needed?

For: apache#1668.
This will make the exception checks work and make the code more reliable.

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

No.

### How was this patch tested?

Existing UTs.

Author: rickyma

client-mr/core/src/main/java/org/apache/hadoop/mapreduce/task/reduce/RssEventFetcher.java

@@ -163,7 +163,9 @@ public void acceptMapCompletionEvents() throws IOException {
         LOG.debug("Got " + events.length + " map completion events from " + fromEventIdx);
       }
 
-      assert !update.shouldReset() : "Unexpected legacy state";
+      if (update.shouldReset()) {
+        throw new RssException("Unexpected legacy state");
+      }
 
       // Update the last seen event ID
       fromEventIdx += events.length;

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

@@ -321,7 +321,10 @@ private void checkSentBlockCount() {
     long expected = blockIds.size();
     long bufferManagerTracked = bufferManager.getBlockCount();
 
-    assert serverToPartitionToBlockIds != null;
+    if (serverToPartitionToBlockIds == null) {
+      throw new RssException("serverToPartitionToBlockIds should not be null");
+    }
+
     // to filter the multiple replica's duplicate blockIds
     Set<Long> blockIds = new HashSet<>();
     for (Map<Integer, Set<Long>> partitionBlockIds : serverToPartitionToBlockIds.values()) {

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

@@ -368,7 +368,10 @@ private void checkSentBlockCount() {
     long expected = blockIds.size();
     long bufferManagerTracked = bufferManager.getBlockCount();
 
-    assert serverToPartitionToBlockIds != null;
+    if (serverToPartitionToBlockIds == null) {
+      throw new RssException("serverToPartitionToBlockIds should not be null");
+    }
+
     // to filter the multiple replica's duplicate blockIds
     Set<Long> blockIds = new HashSet<>();
     for (Map<Integer, Set<Long>> partitionBlockIds : serverToPartitionToBlockIds.values()) {

client-tez/src/main/java/org/apache/tez/common/RssTezUtils.java

@@ -362,7 +362,10 @@ private static void parseRssWorkerFromHostInfo(
           new ShuffleServerInfo(info[0].split(":")[0], Integer.parseInt(info[0].split(":")[1]));
 
       String[] partitions = info[1].split("_");
-      assert (partitions.length > 0);
+
+      if (partitions.length <= 0) {
+        throw new RssException("The length of partitions should not be less than 0");
+      }
       for (String partitionId : partitions) {
         rssWorker.computeIfAbsent(Integer.parseInt(partitionId), k -> new HashSet<>());
         rssWorker.get(Integer.parseInt(partitionId)).add(serverInfo);

client-tez/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/RssShuffleManager.java

@@ -99,6 +99,7 @@
 import org.slf4j.LoggerFactory;
 
 import org.apache.uniffle.common.ShuffleServerInfo;
+import org.apache.uniffle.common.exception.RssException;
 import org.apache.uniffle.common.util.JavaUtils;
 
 // This only knows how to deal with a single srcIndex for a given targetIndex.
@@ -1109,7 +1110,9 @@ private void registerCompletedInputForPipelinedShuffle(
       shuffleInfoEventsMap.put(inputIdentifier, eventInfo);
     }
 
-    assert (eventInfo != null);
+    if (eventInfo == null) {
+      throw new RssException("eventInfo should not be null");
+    }
     eventInfo.spillProcessed(srcAttemptIdentifier.getSpillEventId());
     numFetchedSpills.getAndIncrement();
 

client-tez/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/RssTezFetcherTask.java

@@ -24,6 +24,7 @@
 import java.util.Objects;
 import java.util.Set;
 
+import org.apache.commons.collections4.CollectionUtils;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
 import org.apache.tez.common.CallableWithNdc;
@@ -44,6 +45,7 @@
 import org.apache.uniffle.client.factory.ShuffleClientFactory;
 import org.apache.uniffle.common.RemoteStorageInfo;
 import org.apache.uniffle.common.ShuffleServerInfo;
+import org.apache.uniffle.common.exception.RssException;
 import org.apache.uniffle.common.util.UnitConverter;
 
 public class RssTezFetcherTask extends CallableWithNdc<FetchResult> {
@@ -89,7 +91,9 @@ public RssTezFetcherTask(
       Map<Integer, Roaring64NavigableMap> rssSuccessBlockIdBitmapMap,
       int numPhysicalInputs,
       int partitionNum) {
-    assert (inputs != null && inputs.size() > 0);
+    if (CollectionUtils.isEmpty(inputs)) {
+      throw new RssException("inputs should not be empty");
+    }
     this.fetcherCallback = fetcherCallback;
     this.inputContext = inputContext;
     this.conf = conf;

client-tez/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/RssShuffleScheduler.java

@@ -804,7 +804,9 @@ public synchronized void copySucceeded(
           pipelinedShuffleInfoEventsMap.put(inputIdentifier, eventInfo);
         }
 
-        assert (eventInfo != null);
+        if (eventInfo == null) {
+          throw new RssException("eventInfo should not be null");
+        }
         eventInfo.spillProcessed(srcAttemptIdentifier.getSpillEventId());
         numFetchedSpills++;
 

client-tez/src/main/java/org/apache/tez/runtime/library/input/RssOrderedGroupedKVInput.java

@@ -136,8 +136,12 @@ public synchronized List<Event> initialize() throws IOException {
     TezDAGID tezDAGID = tezVertexID.getDAGId();
     int sourceVertexId = this.conf.getInt(RSS_SHUFFLE_SOURCE_VERTEX_ID, -1);
     int destinationVertexId = this.conf.getInt(RSS_SHUFFLE_DESTINATION_VERTEX_ID, -1);
-    assert sourceVertexId != -1;
-    assert destinationVertexId != -1;
+    if (sourceVertexId == -1) {
+      throw new RssException("sourceVertexId should not be -1");
+    }
+    if (destinationVertexId == -1) {
+      throw new RssException("destinationVertexId should not be -1");
+    }
     this.shuffleId =
         RssTezUtils.computeShuffleId(tezDAGID.getId(), sourceVertexId, destinationVertexId);
     this.applicationAttemptId =

client-tez/src/main/java/org/apache/tez/runtime/library/input/RssUnorderedKVInput.java

@@ -130,8 +130,12 @@ public synchronized List<Event> initialize() throws Exception {
     TezDAGID tezDAGID = tezVertexID.getDAGId();
     int sourceVertexId = this.conf.getInt(RSS_SHUFFLE_SOURCE_VERTEX_ID, -1);
     int destinationVertexId = this.conf.getInt(RSS_SHUFFLE_DESTINATION_VERTEX_ID, -1);
-    assert sourceVertexId != -1;
-    assert destinationVertexId != -1;
+    if (sourceVertexId == -1) {
+      throw new RssException("sourceVertexId should not be -1");
+    }
+    if (destinationVertexId == -1) {
+      throw new RssException("destinationVertexId should not be -1");
+    }
     this.shuffleId =
         RssTezUtils.computeShuffleId(tezDAGID.getId(), sourceVertexId, destinationVertexId);
     this.applicationAttemptId =

client-tez/src/main/java/org/apache/tez/runtime/library/output/RssOrderedPartitionedKVOutput.java

@@ -68,6 +68,7 @@
 import org.slf4j.LoggerFactory;
 
 import org.apache.uniffle.common.ShuffleServerInfo;
+import org.apache.uniffle.common.exception.RssException;
 
 import static org.apache.tez.common.RssTezConfig.RSS_AM_SHUFFLE_MANAGER_ADDRESS;
 import static org.apache.tez.common.RssTezConfig.RSS_AM_SHUFFLE_MANAGER_PORT;
@@ -166,8 +167,12 @@ public TezRemoteShuffleUmbilicalProtocol run() throws Exception {
     TezDAGID tezDAGID = tezVertexID.getDAGId();
     int sourceVertexId = this.conf.getInt(RSS_SHUFFLE_SOURCE_VERTEX_ID, -1);
     int destinationVertexId = this.conf.getInt(RSS_SHUFFLE_DESTINATION_VERTEX_ID, -1);
-    assert sourceVertexId != -1;
-    assert destinationVertexId != -1;
+    if (sourceVertexId == -1) {
+      throw new RssException("sourceVertexId should not be -1");
+    }
+    if (destinationVertexId == -1) {
+      throw new RssException("destinationVertexId should not be -1");
+    }
     this.shuffleId =
         RssTezUtils.computeShuffleId(tezDAGID.getId(), sourceVertexId, destinationVertexId);
     GetShuffleServerRequest request =

client-tez/src/main/java/org/apache/tez/runtime/library/output/RssUnorderedKVOutput.java

@@ -68,6 +68,7 @@
 import org.slf4j.LoggerFactory;
 
 import org.apache.uniffle.common.ShuffleServerInfo;
+import org.apache.uniffle.common.exception.RssException;
 
 import static org.apache.tez.common.RssTezConfig.RSS_AM_SHUFFLE_MANAGER_ADDRESS;
 import static org.apache.tez.common.RssTezConfig.RSS_AM_SHUFFLE_MANAGER_PORT;
@@ -166,8 +167,12 @@ public TezRemoteShuffleUmbilicalProtocol run() throws Exception {
     TezDAGID tezDAGID = tezVertexID.getDAGId();
     int sourceVertexId = this.conf.getInt(RSS_SHUFFLE_SOURCE_VERTEX_ID, -1);
     int destinationVertexId = this.conf.getInt(RSS_SHUFFLE_DESTINATION_VERTEX_ID, -1);
-    assert sourceVertexId != -1;
-    assert destinationVertexId != -1;
+    if (sourceVertexId == -1) {
+      throw new RssException("sourceVertexId should not be -1");
+    }
+    if (destinationVertexId == -1) {
+      throw new RssException("destinationVertexId should not be -1");
+    }
     this.shuffleId =
         RssTezUtils.computeShuffleId(tezDAGID.getId(), sourceVertexId, destinationVertexId);
     this.applicationAttemptId =

client-tez/src/main/java/org/apache/tez/runtime/library/output/RssUnorderedPartitionedKVOutput.java

@@ -68,6 +68,7 @@
 import org.slf4j.LoggerFactory;
 
 import org.apache.uniffle.common.ShuffleServerInfo;
+import org.apache.uniffle.common.exception.RssException;
 
 import static org.apache.tez.common.RssTezConfig.RSS_AM_SHUFFLE_MANAGER_ADDRESS;
 import static org.apache.tez.common.RssTezConfig.RSS_AM_SHUFFLE_MANAGER_PORT;
@@ -164,8 +165,12 @@ public TezRemoteShuffleUmbilicalProtocol run() throws Exception {
     TezDAGID tezDAGID = tezVertexID.getDAGId();
     int sourceVertexId = this.conf.getInt(RSS_SHUFFLE_SOURCE_VERTEX_ID, -1);
     int destinationVertexId = this.conf.getInt(RSS_SHUFFLE_DESTINATION_VERTEX_ID, -1);
-    assert sourceVertexId != -1;
-    assert destinationVertexId != -1;
+    if (sourceVertexId == -1) {
+      throw new RssException("sourceVertexId should not be -1");
+    }
+    if (destinationVertexId == -1) {
+      throw new RssException("destinationVertexId should not be -1");
+    }
     this.shuffleId =
         RssTezUtils.computeShuffleId(tezDAGID.getId(), sourceVertexId, destinationVertexId);
     this.applicationAttemptId =

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

@@ -26,6 +26,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.apache.uniffle.common.exception.RssException;
 import org.apache.uniffle.common.netty.protocol.Message;
 import org.apache.uniffle.common.netty.protocol.MessageWithHeader;
 import org.apache.uniffle.common.netty.protocol.RpcResponse;
@@ -84,7 +85,9 @@ public void encode(ChannelHandlerContext ctx, Message in, List<Object> out) thro
     msgType.encode(header);
     header.writeInt(bodyLength);
     in.encode(header);
-    assert header.writableBytes() == 0;
+    if (header.writableBytes() != 0) {
+      throw new RssException("header's writable bytes should be 0");
+    }
 
     if (body != null) {
       // We transfer ownership of the reference on in.body() to MessageWithHeader.

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

@@ -28,6 +28,7 @@
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
 
+import org.apache.uniffle.common.exception.RssException;
 import org.apache.uniffle.common.netty.protocol.Message;
 
 /**
@@ -160,7 +161,9 @@ private ByteBuf decodeNext() {
       remaining -= next.readableBytes();
       frame.addComponent(next).writerIndex(frame.writerIndex() + next.readableBytes());
     }
-    assert remaining == 0;
+    if (remaining != 0) {
+      throw new RssException("The remaining should be 0");
+    }
     return frame;
   }
 

common/src/main/java/org/apache/uniffle/common/netty/client/TransportClientFactory.java

@@ -38,6 +38,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.apache.uniffle.common.exception.RssException;
 import org.apache.uniffle.common.netty.IOMode;
 import org.apache.uniffle.common.netty.TransportFrameDecoder;
 import org.apache.uniffle.common.netty.handle.TransportChannelHandler;
@@ -233,7 +234,9 @@ public void initChannel(SocketChannel ch) {
     }
 
     TransportClient client = clientRef.get();
-    assert client != null : "Channel future completed successfully with null client";
+    if (client == null) {
+      throw new RssException("Channel future completed unsuccessfully with null client");
+    }
 
     if (logger.isDebugEnabled()) {
       logger.debug("Connection to {} successful", address);

common/src/main/java/org/apache/uniffle/common/netty/protocol/Message.java

@@ -19,6 +19,7 @@
 
 import io.netty.buffer.ByteBuf;
 
+import org.apache.uniffle.common.exception.RssException;
 import org.apache.uniffle.common.netty.buffer.ManagedBuffer;
 
 public abstract class Message implements Encodable {
@@ -66,7 +67,9 @@ public enum Type implements Encodable {
     private final byte id;
 
     Type(int id) {
-      assert id < 128 : "Cannot have more than 128 message types";
+      if (id >= 128) {
+        throw new RssException("Cannot have more than 128 message types");
+      }
       this.id = (byte) id;
     }
 

common/src/main/java/org/apache/uniffle/common/storage/StorageMedia.java

@@ -17,6 +17,7 @@
 
 package org.apache.uniffle.common.storage;
 
+import org.apache.uniffle.common.exception.RssException;
 import org.apache.uniffle.proto.RssProtos.StorageInfo;
 
 public enum StorageMedia {
@@ -29,7 +30,9 @@ public enum StorageMedia {
   private final byte val;
 
   StorageMedia(int code) {
-    assert (code >= -1 && code < 256);
+    if (code < -1 || code >= 256) {
+      throw new RssException("The code should be within [-1, 256)");
+    }
     this.val = (byte) code;
   }
 

common/src/main/java/org/apache/uniffle/common/storage/StorageStatus.java

@@ -17,6 +17,7 @@
 
 package org.apache.uniffle.common.storage;
 
+import org.apache.uniffle.common.exception.RssException;
 import org.apache.uniffle.proto.RssProtos.StorageInfo;
 
 public enum StorageStatus {
@@ -28,7 +29,9 @@ public enum StorageStatus {
   private final byte val;
 
   StorageStatus(int code) {
-    assert (code >= -1 && code < 256);
+    if (code < -1 || code >= 256) {
+      throw new RssException("The code should be within [-1, 256)");
+    }
     this.val = (byte) code;
   }
 

coordinator/src/main/java/org/apache/uniffle/coordinator/web/resource/ServerResource.java

@@ -39,6 +39,7 @@
 
 import org.apache.uniffle.common.Application;
 import org.apache.uniffle.common.ServerStatus;
+import org.apache.uniffle.common.exception.RssException;
 import org.apache.uniffle.common.web.resource.BaseResource;
 import org.apache.uniffle.common.web.resource.Response;
 import org.apache.uniffle.coordinator.ApplicationManager;
@@ -100,8 +101,9 @@ public Response<List<ServerNode>> nodes(@QueryParam("status") String status) {
   public Response<Object> cancelDecommission(CancelDecommissionRequest params) {
     return execute(
         () -> {
-          assert CollectionUtils.isNotEmpty(params.getServerIds())
-              : "Parameter[serverIds] should not be null!";
+          if (CollectionUtils.isEmpty(params.getServerIds())) {
+            throw new RssException("Parameter[serverIds] should not be empty!");
+          }
           params.getServerIds().forEach(getClusterManager()::cancelDecommission);
           return null;
         });
@@ -122,8 +124,9 @@ public Response<Object> cancelDecommission(@PathParam("id") String serverId) {
   public Response<Object> decommission(DecommissionRequest params) {
     return execute(
         () -> {
-          assert CollectionUtils.isNotEmpty(params.getServerIds())
-              : "Parameter[serverIds] should not be null!";
+          if (CollectionUtils.isEmpty(params.getServerIds())) {
+            throw new RssException("Parameter[serverIds] should not be empty!");
+          }
           params.getServerIds().forEach(getClusterManager()::decommission);
           return null;
         });

server/src/main/java/org/apache/uniffle/server/ShuffleServer.java

@@ -284,7 +284,12 @@ private void initialization() throws Exception {
     nettyServerEnabled =
         shuffleServerConf.get(ShuffleServerConf.RPC_SERVER_TYPE) == ServerType.GRPC_NETTY;
     if (nettyServerEnabled) {
-      assert nettyPort >= 0;
+      if (nettyPort < 0) {
+        throw new RssException(
+            String.format(
+                "%s must be set during startup when using GRPC_NETTY",
+                ShuffleServerConf.NETTY_SERVER_PORT.key()));
+      }
       streamServer = new StreamServer(this);
     }