[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #775: [#757] feat(server): separate flush thread pools for different storage types

["jerqi (via GitHub)" <gi...@apache.org>]

jerqi commented on code in PR #775:
URL: https://github.com/apache/incubator-uniffle/pull/775#discussion_r1154572080


##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -163,101 +168,75 @@ private void processEvent(ShuffleDataFlushEvent event) {
     }
   }
 
-  private void flushToFile(ShuffleDataFlushEvent event) {
-    long start = System.currentTimeMillis();
+  private boolean flushToFile(ShuffleDataFlushEvent event) {
     boolean writeSuccess = false;
 
-    while (event.getRetryTimes() <= retryMax) {
-      try {
-        if (!event.isValid()) {
-          writeSuccess = true;
-          LOG.warn("AppId {} was removed already, event {} should be dropped", event.getAppId(), event);
-          break;
-        }
+    try {
+      if (!event.isValid()) {
+        LOG.warn("AppId {} was removed already, event {} should be dropped", event.getAppId(), event);
+        return true;
+      }
 
-        List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
-        if (blocks == null || blocks.isEmpty()) {
-          LOG.info("There is no block to be flushed: {}", event);
-          break;
-        }
+      List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
+      if (blocks == null || blocks.isEmpty()) {
+        LOG.info("There is no block to be flushed: {}", event);
+        return true;
+      }
 
-        Storage storage = storageManager.selectStorage(event);
-        if (storage == null) {
-          LOG.error("Storage selected is null and this should not happen. event: {}", event);
-          break;
-        }
+      Storage storage = event.getUnderStorage();
+      if (storage == null) {
+        LOG.error("Storage selected is null and this should not happen. event: {}", event);
+        return true;
+      }
 
-        if (!storage.canWrite()) {
-          // todo: Could we add an interface supportPending for storageManager
-          //       to unify following logic of multiple different storage managers
-          if (storageManager instanceof MultiStorageManager) {
-            event.increaseRetryTimes();
-            ShuffleServerMetrics.incStorageRetryCounter(storage.getStorageHost());
-            continue;
-          } else {
-            // To avoid being re-pushed to pending queue and make the server too much pressure,
-            // it's better to drop directly.
-            if (event.isPended()) {
-              LOG.error("Drop this event directly due to already having entered pending queue. event: {}", event);
-              break;
-            }
-            event.increaseRetryTimes();
-            event.markPended();
-            addPendingEvents(event);
-            return;
+      if (!storage.canWrite()) {
+        // todo: Could we add an interface supportPending for storageManager
+        //       to unify following logic of multiple different storage managers
+        if (event.getRetryTimes() <= retryMax) {
+          if (event.isPended()) {

Review Comment:
   I'm ok.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org