This is an automated email from the ASF dual-hosted git repository.
omalley pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/hadoop.git
The following commit(s) were added to refs/heads/trunk by this push:
new cd19da13099 HDFS-16877: Enables state context for namenode in
TestObserverWithRouter (#5257)
cd19da13099 is described below
commit cd19da130991b3db02fcc13e1552a5af0bb61c54
Author: Simbarashe Dzinamarira <sdzinamar...@linkedin.com>
AuthorDate: Fri Jan 6 16:18:35 2023 -0800
HDFS-16877: Enables state context for namenode in TestObserverWithRouter
(#5257)
---
.../server/federation/router/RouterRpcServer.java | 14 +++++++++++++-
.../federation/router/TestObserverWithRouter.java | 19 +++++++++++++++++++
2 files changed, 32 insertions(+), 1 deletion(-)
diff --git
a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterRpcServer.java
b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterRpcServer.java
index c4173163436..4f56362f39f 100644
---
a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterRpcServer.java
+++
b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterRpcServer.java
@@ -203,6 +203,9 @@ public class RouterRpcServer extends AbstractService
implements ClientProtocol,
/** Router using this RPC server. */
private final Router router;
+ /** Alignment context storing state IDs for all namespaces this router
serves. */
+ private final RouterStateIdContext routerStateIdContext;
+
/** The RPC server that listens to requests from clients. */
private final Server rpcServer;
/** The address for this RPC server. */
@@ -321,7 +324,7 @@ public class RouterRpcServer extends AbstractService
implements ClientProtocol,
// Create security manager
this.securityManager = new RouterSecurityManager(this.conf);
- RouterStateIdContext routerStateIdContext = new RouterStateIdContext(conf);
+ routerStateIdContext = new RouterStateIdContext(conf);
this.rpcServer = new RPC.Builder(this.conf)
.setProtocol(ClientNamenodeProtocolPB.class)
@@ -510,6 +513,15 @@ public class RouterRpcServer extends AbstractService
implements ClientProtocol,
return this.fedRenameScheduler;
}
+ /**
+ * Get the routerStateIdContext used by this server.
+ * @return routerStateIdContext
+ */
+ @VisibleForTesting
+ protected RouterStateIdContext getRouterStateIdContext() {
+ return routerStateIdContext;
+ }
+
/**
* Get the RPC security manager.
*
diff --git
a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestObserverWithRouter.java
b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestObserverWithRouter.java
index 23e72546aac..48515047faf 100644
---
a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestObserverWithRouter.java
+++
b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestObserverWithRouter.java
@@ -24,6 +24,7 @@ import static org.junit.Assert.assertTrue;
import static org.junit.Assert.assertThrows;
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_HA_NAMENODE_ID_KEY;
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMESERVICE_ID;
+import static
org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMENODE_STATE_CONTEXT_ENABLED_KEY;
import static
org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_MONITOR_NAMENODE;
import java.io.IOException;
@@ -31,6 +32,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.LongAccumulator;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
@@ -91,6 +93,7 @@ public class TestObserverWithRouter {
conf.setBoolean(RBFConfigKeys.DFS_ROUTER_OBSERVER_READ_DEFAULT_KEY, true);
conf.setBoolean(DFSConfigKeys.DFS_HA_TAILEDITS_INPROGRESS_KEY, true);
conf.set(DFSConfigKeys.DFS_HA_TAILEDITS_PERIOD_KEY, "0ms");
+ conf.setBoolean(DFS_NAMENODE_STATE_CONTEXT_ENABLED_KEY, true);
if (confOverrides != null) {
conf.addResource(confOverrides);
}
@@ -545,4 +548,20 @@ public class TestObserverWithRouter {
Assertions.assertEquals(1, latestFederateState.size());
Assertions.assertEquals(10L, latestFederateState.get("ns0"));
}
+
+ @Test
+ public void testStateIdProgressionInRouter() throws Exception {
+ Path rootPath = new Path("/");
+ fileSystem = routerContext.getFileSystem(getConfToEnableObserverReads());
+ RouterStateIdContext routerStateIdContext = routerContext
+ .getRouterRpcServer()
+ .getRouterStateIdContext();
+ for (int i = 0; i < 10; i++) {
+ fileSystem.create(new Path(rootPath, "file" + i)).close();
+ }
+
+ // Get object storing state of the namespace in the shared
RouterStateIdContext
+ LongAccumulator namespaceStateId =
routerStateIdContext.getNamespaceStateId("ns0");
+ assertEquals("Router's shared should have progressed.", 21,
namespaceStateId.get());
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscr...@hadoop.apache.org
For additional commands, e-mail: common-commits-h...@hadoop.apache.org
