[
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]
