[ 
https://issues.apache.org/jira/browse/HDFS-17532?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17849233#comment-17849233
 ] 

ASF GitHub Bot commented on HDFS-17532:
---------------------------------------

ZanderXu commented on code in PR #6839:
URL: https://github.com/apache/hadoop/pull/6839#discussion_r1613173537


##########
hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/StateStoreDriver.java:
##########
@@ -88,6 +101,13 @@ public boolean init(final Configuration config, final 
String id,
         return false;
       }
     }
+
+    if (conf.getBoolean(
+        RBFConfigKeys.FEDERATION_STORE_MEMBERSHIP_ASYNC_OVERRIDE,
+        RBFConfigKeys.FEDERATION_STORE_MEMBERSHIP_ASYNC_OVERRIDE_DEFAULT)) {
+      executor = new ThreadPoolExecutor(2, 2, 1L, TimeUnit.MINUTES, new 
LinkedBlockingQueue<>());

Review Comment:
   You can refer HDFS-16848 to change this configuration to the number of 
threads.



##########
hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/StateStoreDriver.java:
##########
@@ -17,13 +17,23 @@
  */
 package org.apache.hadoop.hdfs.server.federation.store.driver;
 
+import java.io.IOException;
 import java.net.InetAddress;
+import java.util.ArrayList;
 import java.util.Collection;
+import java.util.HashMap;

Review Comment:
   unused.



##########
hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/StateStoreDriver.java:
##########
@@ -206,4 +231,48 @@ private String getHostname() {
     }
     return hostname;
   }
+
+  /**
+   * Try to overwrite records in commitRecords and remove records in 
deleteRecords.
+   * Should return null if async mode is used. Else return removed records.
+   * @param commitRecords records to overwrite in state store
+   * @param deleteRecords records to remove from state store
+   * @param <R> record class
+   * @return null if async mode is used, else removed records
+   */
+  public <R extends BaseRecord> List<R> handleOverwriteAndDelete(List<R> 
commitRecords,
+      List<R> deleteRecords) throws IOException {
+    Callable<StateStoreOperationResult> overwriteCallable =
+        () -> putAll(commitRecords, true, false);
+    Callable<Map<R, Boolean>> deletionCallable = () -> 
removeMultiple(deleteRecords);
+
+    if (executor != null) {
+      // In async mode, just submit and let the tasks do their work and return 
asap.
+      if (!commitRecords.isEmpty()) {
+        executor.submit(overwriteCallable);
+      }
+      if (!deleteRecords.isEmpty()) {
+        executor.submit(deletionCallable);
+      }
+      return null;
+    } else {
+      try {
+        List<R> result = new ArrayList<>();
+        if (!commitRecords.isEmpty()) {
+          overwriteCallable.call();
+        }
+        if (!deleteRecords.isEmpty()) {
+          Map<R, Boolean> removedRecords = deletionCallable.call();
+          for (Map.Entry<R, Boolean> entry : removedRecords.entrySet()) {
+            if (entry.getValue()) {
+              result.add(entry.getKey());
+            }
+          }
+        }
+        return result;
+      } catch (Exception e) {
+        throw new IOException(e);
+      }
+    }
+  }

Review Comment:
   ```
   public <R extends BaseRecord> List<R> handleOverwriteAndDelete(List<R> 
commitRecords,
         List<R> deleteRecords) throws IOException {
       List<R> result = null;
       try {
         // overwrite all expired records.
         if (commitRecords != null && !commitRecords.isEmpty()) {
           Callable<StateStoreOperationResult> overwriteCallable =
               () -> putAll(commitRecords, true, false);
           if (executor != null) {
             executor.submit(overwriteCallable);
           } else {
             overwriteCallable.call();
           }
         }
         
         // delete all deletable records.
         if (deleteRecords != null && !deleteRecords.isEmpty()) {
           Callable<Map<R, Boolean>> deletionCallable = () -> 
removeMultiple(deleteRecords);
           if (executor != null) {
             executor.submit(deletionCallable);
           } else {
             result = new ArrayList<>();
             Map<R, Boolean> removedRecords = deletionCallable.call();
             for (Map.Entry<R, Boolean> entry : removedRecords.entrySet()) {
               if (entry.getValue()) {
                 result.add(entry.getKey());
               }
             }
           }
         }
       } catch (Exception e) {
         throw new IOException(e);
       }
       return result;
     }
   ```





> RBF: Allow router state store cache update to overwrite and delete in parallel
> ------------------------------------------------------------------------------
>
>                 Key: HDFS-17532
>                 URL: https://issues.apache.org/jira/browse/HDFS-17532
>             Project: Hadoop HDFS
>          Issue Type: Improvement
>          Components: hdfs, rbf
>            Reporter: Felix N
>            Assignee: Felix N
>            Priority: Minor
>              Labels: pull-request-available
>
> Current implementation for router state store update is quite inefficient, so 
> much that when routers are removed and a lot of NameNodeMembership records 
> are deleted in a short burst, the deletions triggered a router safemode in 
> our cluster and caused a lot of troubles.
> This ticket aims to allow the overwrite part and delete part of 
> org.apache.hadoop.hdfs.server.federation.store.CachedRecordStore#overrideExpiredRecords
>  to run in parallel.
> See HDFS-17529 for the other half of this improvement.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)

---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]

Reply via email to