You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by ha...@apache.org on 2018/03/21 23:48:01 UTC
[25/50] [abbrv] hadoop git commit: HDFS-13215. RBF: Move Router to
its own module. Contributed by Wei Yan
http://git-wip-us.apache.org/repos/asf/hadoop/blob/87700d45/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/store/TestStateStoreMembershipState.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/store/TestStateStoreMembershipState.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/store/TestStateStoreMembershipState.java
new file mode 100644
index 0000000..dd349da
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/store/TestStateStoreMembershipState.java
@@ -0,0 +1,463 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hdfs.server.federation.store;
+
+import static org.apache.hadoop.hdfs.server.federation.FederationTestUtils.NAMENODES;
+import static org.apache.hadoop.hdfs.server.federation.FederationTestUtils.NAMESERVICES;
+import static org.apache.hadoop.hdfs.server.federation.FederationTestUtils.ROUTERS;
+import static org.apache.hadoop.hdfs.server.federation.FederationTestUtils.verifyException;
+import static org.apache.hadoop.hdfs.server.federation.store.FederationStateStoreTestUtils.clearRecords;
+import static org.apache.hadoop.hdfs.server.federation.store.FederationStateStoreTestUtils.createMockRegistrationForNamenode;
+import static org.apache.hadoop.hdfs.server.federation.store.FederationStateStoreTestUtils.synchronizeRecords;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.hadoop.hdfs.server.federation.resolver.FederationNamenodeServiceState;
+import org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.GetNamenodeRegistrationsRequest;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.GetNamenodeRegistrationsResponse;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.NamenodeHeartbeatRequest;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.NamenodeHeartbeatResponse;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.UpdateNamenodeRegistrationRequest;
+import org.apache.hadoop.hdfs.server.federation.store.records.MembershipState;
+import org.apache.hadoop.util.Time;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+/**
+ * Test the basic {@link MembershipStore} membership functionality.
+ */
+public class TestStateStoreMembershipState extends TestStateStoreBase {
+
+ private static MembershipStore membershipStore;
+
+ @BeforeClass
+ public static void create() {
+ // Reduce expirations to 5 seconds
+ getConf().setLong(
+ RBFConfigKeys.FEDERATION_STORE_MEMBERSHIP_EXPIRATION_MS,
+ TimeUnit.SECONDS.toMillis(5));
+ }
+
+ @Before
+ public void setup() throws IOException, InterruptedException {
+
+ membershipStore =
+ getStateStore().getRegisteredRecordStore(MembershipStore.class);
+
+ // Clear NN registrations
+ assertTrue(clearRecords(getStateStore(), MembershipState.class));
+ }
+
+ @Test
+ public void testNamenodeStateOverride() throws Exception {
+ // Populate the state store
+ // 1) ns0:nn0 - Standby
+ String ns = "ns0";
+ String nn = "nn0";
+ MembershipState report = createRegistration(
+ ns, nn, ROUTERS[1], FederationNamenodeServiceState.STANDBY);
+ assertTrue(namenodeHeartbeat(report));
+
+ // Load data into cache and calculate quorum
+ assertTrue(getStateStore().loadCache(MembershipStore.class, true));
+
+ MembershipState existingState = getNamenodeRegistration(ns, nn);
+ assertEquals(
+ FederationNamenodeServiceState.STANDBY, existingState.getState());
+
+ // Override cache
+ UpdateNamenodeRegistrationRequest request =
+ UpdateNamenodeRegistrationRequest.newInstance(
+ ns, nn, FederationNamenodeServiceState.ACTIVE);
+ assertTrue(membershipStore.updateNamenodeRegistration(request).getResult());
+
+ MembershipState newState = getNamenodeRegistration(ns, nn);
+ assertEquals(FederationNamenodeServiceState.ACTIVE, newState.getState());
+ }
+
+ @Test
+ public void testStateStoreDisconnected() throws Exception {
+
+ // Close the data store driver
+ getStateStore().closeDriver();
+ assertFalse(getStateStore().isDriverReady());
+
+ NamenodeHeartbeatRequest hbRequest = NamenodeHeartbeatRequest.newInstance();
+ hbRequest.setNamenodeMembership(
+ createMockRegistrationForNamenode(
+ "test", "test", FederationNamenodeServiceState.UNAVAILABLE));
+ verifyException(membershipStore, "namenodeHeartbeat",
+ StateStoreUnavailableException.class,
+ new Class[] {NamenodeHeartbeatRequest.class},
+ new Object[] {hbRequest });
+
+ // Information from cache, no exception should be triggered for these
+ // TODO - should cached info expire at some point?
+ GetNamenodeRegistrationsRequest getRequest =
+ GetNamenodeRegistrationsRequest.newInstance();
+ verifyException(membershipStore,
+ "getNamenodeRegistrations", null,
+ new Class[] {GetNamenodeRegistrationsRequest.class},
+ new Object[] {getRequest});
+
+ verifyException(membershipStore,
+ "getExpiredNamenodeRegistrations", null,
+ new Class[] {GetNamenodeRegistrationsRequest.class},
+ new Object[] {getRequest});
+
+ UpdateNamenodeRegistrationRequest overrideRequest =
+ UpdateNamenodeRegistrationRequest.newInstance();
+ verifyException(membershipStore,
+ "updateNamenodeRegistration", null,
+ new Class[] {UpdateNamenodeRegistrationRequest.class},
+ new Object[] {overrideRequest});
+ }
+
+ private void registerAndLoadRegistrations(
+ List<MembershipState> registrationList) throws IOException {
+ // Populate
+ assertTrue(synchronizeRecords(
+ getStateStore(), registrationList, MembershipState.class));
+
+ // Load into cache
+ assertTrue(getStateStore().loadCache(MembershipStore.class, true));
+ }
+
+ private MembershipState createRegistration(String ns, String nn,
+ String router, FederationNamenodeServiceState state) throws IOException {
+ MembershipState record = MembershipState.newInstance(
+ router, ns,
+ nn, "testcluster", "testblock-" + ns, "testrpc-"+ ns + nn,
+ "testservice-"+ ns + nn, "testlifeline-"+ ns + nn,
+ "testweb-" + ns + nn, state, false);
+ return record;
+ }
+
+ @Test
+ public void testRegistrationMajorityQuorum()
+ throws InterruptedException, IOException {
+
+ // Populate the state store with a set of non-matching elements
+ // 1) ns0:nn0 - Standby (newest)
+ // 2) ns0:nn0 - Active (oldest)
+ // 3) ns0:nn0 - Active (2nd oldest)
+ // 4) ns0:nn0 - Active (3nd oldest element, newest active element)
+ // Verify the selected entry is the newest majority opinion (4)
+ String ns = "ns0";
+ String nn = "nn0";
+
+ // Active - oldest
+ MembershipState report = createRegistration(
+ ns, nn, ROUTERS[1], FederationNamenodeServiceState.ACTIVE);
+ assertTrue(namenodeHeartbeat(report));
+ Thread.sleep(1000);
+
+ // Active - 2nd oldest
+ report = createRegistration(
+ ns, nn, ROUTERS[2], FederationNamenodeServiceState.ACTIVE);
+ assertTrue(namenodeHeartbeat(report));
+ Thread.sleep(1000);
+
+ // Active - 3rd oldest, newest active element
+ report = createRegistration(
+ ns, nn, ROUTERS[3], FederationNamenodeServiceState.ACTIVE);
+ assertTrue(namenodeHeartbeat(report));
+
+ // standby - newest overall
+ report = createRegistration(
+ ns, nn, ROUTERS[0], FederationNamenodeServiceState.STANDBY);
+ assertTrue(namenodeHeartbeat(report));
+
+ // Load and calculate quorum
+ assertTrue(getStateStore().loadCache(MembershipStore.class, true));
+
+ // Verify quorum entry
+ MembershipState quorumEntry = getNamenodeRegistration(
+ report.getNameserviceId(), report.getNamenodeId());
+ assertNotNull(quorumEntry);
+ assertEquals(quorumEntry.getRouterId(), ROUTERS[3]);
+ }
+
+ @Test
+ public void testRegistrationQuorumExcludesExpired()
+ throws InterruptedException, IOException {
+
+ // Populate the state store with some expired entries and verify the expired
+ // entries are ignored.
+ // 1) ns0:nn0 - Active
+ // 2) ns0:nn0 - Expired
+ // 3) ns0:nn0 - Expired
+ // 4) ns0:nn0 - Expired
+ // Verify the selected entry is the active entry
+ List<MembershipState> registrationList = new ArrayList<>();
+ String ns = "ns0";
+ String nn = "nn0";
+ String rpcAddress = "testrpcaddress";
+ String serviceAddress = "testserviceaddress";
+ String lifelineAddress = "testlifelineaddress";
+ String blockPoolId = "testblockpool";
+ String clusterId = "testcluster";
+ String webAddress = "testwebaddress";
+ boolean safemode = false;
+
+ // Active
+ MembershipState record = MembershipState.newInstance(
+ ROUTERS[0], ns, nn, clusterId, blockPoolId,
+ rpcAddress, serviceAddress, lifelineAddress, webAddress,
+ FederationNamenodeServiceState.ACTIVE, safemode);
+ registrationList.add(record);
+
+ // Expired
+ record = MembershipState.newInstance(
+ ROUTERS[1], ns, nn, clusterId, blockPoolId,
+ rpcAddress, serviceAddress, lifelineAddress, webAddress,
+ FederationNamenodeServiceState.EXPIRED, safemode);
+ registrationList.add(record);
+
+ // Expired
+ record = MembershipState.newInstance(
+ ROUTERS[2], ns, nn, clusterId, blockPoolId,
+ rpcAddress, serviceAddress, lifelineAddress, webAddress,
+ FederationNamenodeServiceState.EXPIRED, safemode);
+ registrationList.add(record);
+
+ // Expired
+ record = MembershipState.newInstance(
+ ROUTERS[3], ns, nn, clusterId, blockPoolId,
+ rpcAddress, serviceAddress, lifelineAddress, webAddress,
+ FederationNamenodeServiceState.EXPIRED, safemode);
+ registrationList.add(record);
+ registerAndLoadRegistrations(registrationList);
+
+ // Verify quorum entry chooses active membership
+ MembershipState quorumEntry = getNamenodeRegistration(
+ record.getNameserviceId(), record.getNamenodeId());
+ assertNotNull(quorumEntry);
+ assertEquals(ROUTERS[0], quorumEntry.getRouterId());
+ }
+
+ @Test
+ public void testRegistrationQuorumAllExpired() throws IOException {
+
+ // 1) ns0:nn0 - Expired (oldest)
+ // 2) ns0:nn0 - Expired
+ // 3) ns0:nn0 - Expired
+ // 4) ns0:nn0 - Expired
+ // Verify no entry is either selected or cached
+ List<MembershipState> registrationList = new ArrayList<>();
+ String ns = NAMESERVICES[0];
+ String nn = NAMENODES[0];
+ String rpcAddress = "testrpcaddress";
+ String serviceAddress = "testserviceaddress";
+ String lifelineAddress = "testlifelineaddress";
+ String blockPoolId = "testblockpool";
+ String clusterId = "testcluster";
+ String webAddress = "testwebaddress";
+ boolean safemode = false;
+ long startingTime = Time.now();
+
+ // Expired
+ MembershipState record = MembershipState.newInstance(
+ ROUTERS[0], ns, nn, clusterId, blockPoolId,
+ rpcAddress, webAddress, lifelineAddress, webAddress,
+ FederationNamenodeServiceState.EXPIRED, safemode);
+ record.setDateModified(startingTime - 10000);
+ registrationList.add(record);
+
+ // Expired
+ record = MembershipState.newInstance(
+ ROUTERS[1], ns, nn, clusterId, blockPoolId,
+ rpcAddress, serviceAddress, lifelineAddress, webAddress,
+ FederationNamenodeServiceState.EXPIRED, safemode);
+ record.setDateModified(startingTime);
+ registrationList.add(record);
+
+ // Expired
+ record = MembershipState.newInstance(
+ ROUTERS[2], ns, nn, clusterId, blockPoolId,
+ rpcAddress, serviceAddress, lifelineAddress, webAddress,
+ FederationNamenodeServiceState.EXPIRED, safemode);
+ record.setDateModified(startingTime);
+ registrationList.add(record);
+
+ // Expired
+ record = MembershipState.newInstance(
+ ROUTERS[3], ns, nn, clusterId, blockPoolId,
+ rpcAddress, serviceAddress, lifelineAddress, webAddress,
+ FederationNamenodeServiceState.EXPIRED, safemode);
+ record.setDateModified(startingTime);
+ registrationList.add(record);
+
+ registerAndLoadRegistrations(registrationList);
+
+ // Verify no entry is found for this nameservice
+ assertNull(getNamenodeRegistration(
+ record.getNameserviceId(), record.getNamenodeId()));
+ }
+
+ @Test
+ public void testRegistrationNoQuorum()
+ throws InterruptedException, IOException {
+
+ // Populate the state store with a set of non-matching elements
+ // 1) ns0:nn0 - Standby (newest)
+ // 2) ns0:nn0 - Standby (oldest)
+ // 3) ns0:nn0 - Active (2nd oldest)
+ // 4) ns0:nn0 - Active (3nd oldest element, newest active element)
+ // Verify the selected entry is the newest entry (1)
+ MembershipState report1 = createRegistration(
+ NAMESERVICES[0], NAMENODES[0], ROUTERS[1],
+ FederationNamenodeServiceState.STANDBY);
+ assertTrue(namenodeHeartbeat(report1));
+ Thread.sleep(100);
+ MembershipState report2 = createRegistration(
+ NAMESERVICES[0], NAMENODES[0], ROUTERS[2],
+ FederationNamenodeServiceState.ACTIVE);
+ assertTrue(namenodeHeartbeat(report2));
+ Thread.sleep(100);
+ MembershipState report3 = createRegistration(
+ NAMESERVICES[0], NAMENODES[0], ROUTERS[3],
+ FederationNamenodeServiceState.ACTIVE);
+ assertTrue(namenodeHeartbeat(report3));
+ Thread.sleep(100);
+ MembershipState report4 = createRegistration(
+ NAMESERVICES[0], NAMENODES[0], ROUTERS[0],
+ FederationNamenodeServiceState.STANDBY);
+ assertTrue(namenodeHeartbeat(report4));
+
+ // Load and calculate quorum
+ assertTrue(getStateStore().loadCache(MembershipStore.class, true));
+
+ // Verify quorum entry uses the newest data, even though it is standby
+ MembershipState quorumEntry = getNamenodeRegistration(
+ report1.getNameserviceId(), report1.getNamenodeId());
+ assertNotNull(quorumEntry);
+ assertEquals(ROUTERS[0], quorumEntry.getRouterId());
+ assertEquals(
+ FederationNamenodeServiceState.STANDBY, quorumEntry.getState());
+ }
+
+ @Test
+ public void testRegistrationExpired()
+ throws InterruptedException, IOException {
+
+ // Populate the state store with a single NN element
+ // 1) ns0:nn0 - Active
+ // Wait for the entry to expire without heartbeating
+ // Verify the NN entry is populated as EXPIRED internally in the state store
+
+ MembershipState report = createRegistration(
+ NAMESERVICES[0], NAMENODES[0], ROUTERS[0],
+ FederationNamenodeServiceState.ACTIVE);
+ assertTrue(namenodeHeartbeat(report));
+
+ // Load cache
+ assertTrue(getStateStore().loadCache(MembershipStore.class, true));
+
+ // Verify quorum and entry
+ MembershipState quorumEntry = getNamenodeRegistration(
+ report.getNameserviceId(), report.getNamenodeId());
+ assertNotNull(quorumEntry);
+ assertEquals(ROUTERS[0], quorumEntry.getRouterId());
+ assertEquals(FederationNamenodeServiceState.ACTIVE, quorumEntry.getState());
+
+ // Wait past expiration (set in conf to 5 seconds)
+ Thread.sleep(6000);
+ // Reload cache
+ assertTrue(getStateStore().loadCache(MembershipStore.class, true));
+
+ // Verify entry is now expired and is no longer in the cache
+ quorumEntry = getNamenodeRegistration(NAMESERVICES[0], NAMENODES[0]);
+ assertNull(quorumEntry);
+
+ // Verify entry is now expired and can't be used by RPC service
+ quorumEntry = getNamenodeRegistration(
+ report.getNameserviceId(), report.getNamenodeId());
+ assertNull(quorumEntry);
+
+ // Heartbeat again, updates dateModified
+ assertTrue(namenodeHeartbeat(report));
+ // Reload cache
+ assertTrue(getStateStore().loadCache(MembershipStore.class, true));
+
+ // Verify updated entry marked as active and is accessible to RPC server
+ quorumEntry = getNamenodeRegistration(
+ report.getNameserviceId(), report.getNamenodeId());
+ assertNotNull(quorumEntry);
+ assertEquals(ROUTERS[0], quorumEntry.getRouterId());
+ assertEquals(FederationNamenodeServiceState.ACTIVE, quorumEntry.getState());
+ }
+
+ /**
+ * Get a single namenode membership record from the store.
+ *
+ * @param nsId The HDFS nameservice ID to search for
+ * @param nnId The HDFS namenode ID to search for
+ * @return The single NamenodeMembershipRecord that matches the query or null
+ * if not found.
+ * @throws IOException if the query could not be executed.
+ */
+ private MembershipState getNamenodeRegistration(
+ final String nsId, final String nnId) throws IOException {
+
+ MembershipState partial = MembershipState.newInstance();
+ partial.setNameserviceId(nsId);
+ partial.setNamenodeId(nnId);
+ GetNamenodeRegistrationsRequest request =
+ GetNamenodeRegistrationsRequest.newInstance(partial);
+ GetNamenodeRegistrationsResponse response =
+ membershipStore.getNamenodeRegistrations(request);
+
+ List<MembershipState> results = response.getNamenodeMemberships();
+ if (results != null && results.size() == 1) {
+ MembershipState record = results.get(0);
+ return record;
+ }
+ return null;
+ }
+
+ /**
+ * Register a namenode heartbeat with the state store.
+ *
+ * @param store FederationMembershipStateStore instance to retrieve the
+ * membership data records.
+ * @param namenode A fully populated namenode membership record to be
+ * committed to the data store.
+ * @return True if successful, false otherwise.
+ * @throws IOException if the state store query could not be performed.
+ */
+ private boolean namenodeHeartbeat(MembershipState namenode)
+ throws IOException {
+
+ NamenodeHeartbeatRequest request =
+ NamenodeHeartbeatRequest.newInstance(namenode);
+ NamenodeHeartbeatResponse response =
+ membershipStore.namenodeHeartbeat(request);
+ return response.getResult();
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/87700d45/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/store/TestStateStoreMountTable.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/store/TestStateStoreMountTable.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/store/TestStateStoreMountTable.java
new file mode 100644
index 0000000..d30d6ba
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/store/TestStateStoreMountTable.java
@@ -0,0 +1,250 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hdfs.server.federation.store;
+
+import static org.apache.hadoop.hdfs.server.federation.FederationTestUtils.NAMESERVICES;
+import static org.apache.hadoop.hdfs.server.federation.FederationTestUtils.verifyException;
+import static org.apache.hadoop.hdfs.server.federation.store.FederationStateStoreTestUtils.clearRecords;
+import static org.apache.hadoop.hdfs.server.federation.store.FederationStateStoreTestUtils.createMockMountTable;
+import static org.apache.hadoop.hdfs.server.federation.store.FederationStateStoreTestUtils.synchronizeRecords;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.hadoop.hdfs.server.federation.store.protocol.AddMountTableEntryRequest;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.AddMountTableEntryResponse;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.GetMountTableEntriesRequest;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.GetMountTableEntriesResponse;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.RemoveMountTableEntryRequest;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.UpdateMountTableEntryRequest;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.UpdateMountTableEntryResponse;
+import org.apache.hadoop.hdfs.server.federation.store.records.MountTable;
+import org.apache.hadoop.hdfs.server.federation.store.records.QueryResult;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+/**
+ * Test the basic {@link StateStoreService}
+ * {@link MountTableStore} functionality.
+ */
+public class TestStateStoreMountTable extends TestStateStoreBase {
+
+ private static List<String> nameservices;
+ private static MountTableStore mountStore;
+
+ @BeforeClass
+ public static void create() throws IOException {
+ nameservices = new ArrayList<>();
+ nameservices.add(NAMESERVICES[0]);
+ nameservices.add(NAMESERVICES[1]);
+ }
+
+ @Before
+ public void setup() throws IOException, InterruptedException {
+ mountStore =
+ getStateStore().getRegisteredRecordStore(MountTableStore.class);
+ // Clear Mount table registrations
+ assertTrue(clearRecords(getStateStore(), MountTable.class));
+ }
+
+ @Test
+ public void testStateStoreDisconnected() throws Exception {
+
+ // Close the data store driver
+ getStateStore().closeDriver();
+ assertFalse(getStateStore().isDriverReady());
+
+ // Test APIs that access the store to check they throw the correct exception
+ AddMountTableEntryRequest addRequest =
+ AddMountTableEntryRequest.newInstance();
+ verifyException(mountStore, "addMountTableEntry",
+ StateStoreUnavailableException.class,
+ new Class[] {AddMountTableEntryRequest.class},
+ new Object[] {addRequest});
+
+ UpdateMountTableEntryRequest updateRequest =
+ UpdateMountTableEntryRequest.newInstance();
+ verifyException(mountStore, "updateMountTableEntry",
+ StateStoreUnavailableException.class,
+ new Class[] {UpdateMountTableEntryRequest.class},
+ new Object[] {updateRequest});
+
+ RemoveMountTableEntryRequest removeRequest =
+ RemoveMountTableEntryRequest.newInstance();
+ verifyException(mountStore, "removeMountTableEntry",
+ StateStoreUnavailableException.class,
+ new Class[] {RemoveMountTableEntryRequest.class},
+ new Object[] {removeRequest});
+
+ GetMountTableEntriesRequest getRequest =
+ GetMountTableEntriesRequest.newInstance();
+ mountStore.loadCache(true);
+ verifyException(mountStore, "getMountTableEntries",
+ StateStoreUnavailableException.class,
+ new Class[] {GetMountTableEntriesRequest.class},
+ new Object[] {getRequest});
+ }
+
+ @Test
+ public void testSynchronizeMountTable() throws IOException {
+ // Synchronize and get mount table entries
+ List<MountTable> entries = createMockMountTable(nameservices);
+ assertTrue(synchronizeRecords(getStateStore(), entries, MountTable.class));
+ for (MountTable e : entries) {
+ mountStore.loadCache(true);
+ MountTable entry = getMountTableEntry(e.getSourcePath());
+ assertNotNull(entry);
+ assertEquals(e.getDefaultLocation().getDest(),
+ entry.getDefaultLocation().getDest());
+ }
+ }
+
+ @Test
+ public void testAddMountTableEntry() throws IOException {
+
+ // Add 1
+ List<MountTable> entries = createMockMountTable(nameservices);
+ List<MountTable> entries1 = getMountTableEntries("/").getRecords();
+ assertEquals(0, entries1.size());
+ MountTable entry0 = entries.get(0);
+ AddMountTableEntryRequest request =
+ AddMountTableEntryRequest.newInstance(entry0);
+ AddMountTableEntryResponse response =
+ mountStore.addMountTableEntry(request);
+ assertTrue(response.getStatus());
+
+ mountStore.loadCache(true);
+ List<MountTable> entries2 = getMountTableEntries("/").getRecords();
+ assertEquals(1, entries2.size());
+ }
+
+ @Test
+ public void testRemoveMountTableEntry() throws IOException {
+
+ // Add many
+ List<MountTable> entries = createMockMountTable(nameservices);
+ synchronizeRecords(getStateStore(), entries, MountTable.class);
+ mountStore.loadCache(true);
+ List<MountTable> entries1 = getMountTableEntries("/").getRecords();
+ assertEquals(entries.size(), entries1.size());
+
+ // Remove 1
+ RemoveMountTableEntryRequest request =
+ RemoveMountTableEntryRequest.newInstance();
+ request.setSrcPath(entries.get(0).getSourcePath());
+ assertTrue(mountStore.removeMountTableEntry(request).getStatus());
+
+ // Verify remove
+ mountStore.loadCache(true);
+ List<MountTable> entries2 = getMountTableEntries("/").getRecords();
+ assertEquals(entries.size() - 1, entries2.size());
+ }
+
+ @Test
+ public void testUpdateMountTableEntry() throws IOException {
+
+ // Add 1
+ List<MountTable> entries = createMockMountTable(nameservices);
+ MountTable entry0 = entries.get(0);
+ String srcPath = entry0.getSourcePath();
+ String nsId = entry0.getDefaultLocation().getNameserviceId();
+ AddMountTableEntryRequest request =
+ AddMountTableEntryRequest.newInstance(entry0);
+ AddMountTableEntryResponse response =
+ mountStore.addMountTableEntry(request);
+ assertTrue(response.getStatus());
+
+ // Verify
+ mountStore.loadCache(true);
+ MountTable matchingEntry0 = getMountTableEntry(srcPath);
+ assertNotNull(matchingEntry0);
+ assertEquals(nsId, matchingEntry0.getDefaultLocation().getNameserviceId());
+
+ // Edit destination nameservice for source path
+ Map<String, String> destMap =
+ Collections.singletonMap("testnameservice", "/");
+ MountTable replacement =
+ MountTable.newInstance(srcPath, destMap);
+ UpdateMountTableEntryRequest updateRequest =
+ UpdateMountTableEntryRequest.newInstance(replacement);
+ UpdateMountTableEntryResponse updateResponse =
+ mountStore.updateMountTableEntry(updateRequest);
+ assertTrue(updateResponse.getStatus());
+
+ // Verify
+ mountStore.loadCache(true);
+ MountTable matchingEntry1 = getMountTableEntry(srcPath);
+ assertNotNull(matchingEntry1);
+ assertEquals("testnameservice",
+ matchingEntry1.getDefaultLocation().getNameserviceId());
+ }
+
+ /**
+ * Gets an existing mount table record in the state store.
+ *
+ * @param mount The mount point of the record to remove.
+ * @return The matching record if found, null if it is not found.
+ * @throws IOException If the state store could not be accessed.
+ */
+ private MountTable getMountTableEntry(String mount) throws IOException {
+ GetMountTableEntriesRequest request =
+ GetMountTableEntriesRequest.newInstance(mount);
+ GetMountTableEntriesResponse response =
+ mountStore.getMountTableEntries(request);
+ List<MountTable> results = response.getEntries();
+ if (results.size() > 0) {
+ // First result is sorted to have the shortest mount string length
+ return results.get(0);
+ }
+ return null;
+ }
+
+ /**
+ * Fetch all mount table records beneath a root path.
+ *
+ * @param store FederationMountTableStore instance to commit the data.
+ * @param mount The root search path, enter "/" to return all mount table
+ * records.
+ *
+ * @return A list of all mount table records found below the root mount.
+ *
+ * @throws IOException If the state store could not be accessed.
+ */
+ private QueryResult<MountTable> getMountTableEntries(String mount)
+ throws IOException {
+ if (mount == null) {
+ throw new IOException("Please specify a root search path");
+ }
+ GetMountTableEntriesRequest request =
+ GetMountTableEntriesRequest.newInstance();
+ request.setSrcPath(mount);
+ GetMountTableEntriesResponse response =
+ mountStore.getMountTableEntries(request);
+ List<MountTable> records = response.getEntries();
+ long timestamp = response.getTimestamp();
+ return new QueryResult<MountTable>(records, timestamp);
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/87700d45/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/store/TestStateStoreRouterState.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/store/TestStateStoreRouterState.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/store/TestStateStoreRouterState.java
new file mode 100644
index 0000000..db1df19
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/store/TestStateStoreRouterState.java
@@ -0,0 +1,195 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hdfs.server.federation.store;
+
+import static org.apache.hadoop.hdfs.server.federation.FederationTestUtils.verifyException;
+import static org.apache.hadoop.hdfs.server.federation.store.FederationStateStoreTestUtils.clearRecords;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.hadoop.hdfs.server.federation.router.FederationUtil;
+import org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys;
+import org.apache.hadoop.hdfs.server.federation.router.RouterServiceState;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.GetRouterRegistrationRequest;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.GetRouterRegistrationsRequest;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.RouterHeartbeatRequest;
+import org.apache.hadoop.hdfs.server.federation.store.records.RouterState;
+import org.apache.hadoop.util.Time;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+/**
+ * Test the basic {@link StateStoreService} {@link RouterStore} functionality.
+ */
+public class TestStateStoreRouterState extends TestStateStoreBase {
+
+ private static RouterStore routerStore;
+
+ @BeforeClass
+ public static void create() {
+ // Reduce expirations to 5 seconds
+ getConf().setTimeDuration(
+ RBFConfigKeys.FEDERATION_STORE_ROUTER_EXPIRATION_MS,
+ 5, TimeUnit.SECONDS);
+ }
+
+ @Before
+ public void setup() throws IOException, InterruptedException {
+
+ if (routerStore == null) {
+ routerStore =
+ getStateStore().getRegisteredRecordStore(RouterStore.class);
+ }
+
+ // Clear router status registrations
+ assertTrue(clearRecords(getStateStore(), RouterState.class));
+ }
+
+ @Test
+ public void testStateStoreDisconnected() throws Exception {
+
+ // Close the data store driver
+ getStateStore().closeDriver();
+ assertEquals(false, getStateStore().isDriverReady());
+
+ // Test all APIs that access the data store to ensure they throw the correct
+ // exception.
+ GetRouterRegistrationRequest getSingleRequest =
+ GetRouterRegistrationRequest.newInstance();
+ verifyException(routerStore, "getRouterRegistration",
+ StateStoreUnavailableException.class,
+ new Class[] {GetRouterRegistrationRequest.class},
+ new Object[] {getSingleRequest});
+
+ GetRouterRegistrationsRequest getRequest =
+ GetRouterRegistrationsRequest.newInstance();
+ routerStore.loadCache(true);
+ verifyException(routerStore, "getRouterRegistrations",
+ StateStoreUnavailableException.class,
+ new Class[] {GetRouterRegistrationsRequest.class},
+ new Object[] {getRequest});
+
+ RouterHeartbeatRequest hbRequest = RouterHeartbeatRequest.newInstance(
+ RouterState.newInstance("test", 0, RouterServiceState.UNINITIALIZED));
+ verifyException(routerStore, "routerHeartbeat",
+ StateStoreUnavailableException.class,
+ new Class[] {RouterHeartbeatRequest.class},
+ new Object[] {hbRequest});
+ }
+
+ //
+ // Router
+ //
+ @Test
+ public void testUpdateRouterStatus()
+ throws IllegalStateException, IOException {
+
+ long dateStarted = Time.now();
+ String address = "testaddress";
+
+ // Set
+ RouterHeartbeatRequest request = RouterHeartbeatRequest.newInstance(
+ RouterState.newInstance(
+ address, dateStarted, RouterServiceState.RUNNING));
+ assertTrue(routerStore.routerHeartbeat(request).getStatus());
+
+ // Verify
+ GetRouterRegistrationRequest getRequest =
+ GetRouterRegistrationRequest.newInstance(address);
+ RouterState record =
+ routerStore.getRouterRegistration(getRequest).getRouter();
+ assertNotNull(record);
+ assertEquals(RouterServiceState.RUNNING, record.getStatus());
+ assertEquals(address, record.getAddress());
+ assertEquals(FederationUtil.getCompileInfo(), record.getCompileInfo());
+ // Build version may vary a bit
+ assertFalse(record.getVersion().isEmpty());
+ }
+
+ @Test
+ public void testRouterStateExpired()
+ throws IOException, InterruptedException {
+
+ long dateStarted = Time.now();
+ String address = "testaddress";
+
+ RouterHeartbeatRequest request = RouterHeartbeatRequest.newInstance(
+ RouterState.newInstance(
+ address, dateStarted, RouterServiceState.RUNNING));
+ // Set
+ assertTrue(routerStore.routerHeartbeat(request).getStatus());
+
+ // Verify
+ GetRouterRegistrationRequest getRequest =
+ GetRouterRegistrationRequest.newInstance(address);
+ RouterState record =
+ routerStore.getRouterRegistration(getRequest).getRouter();
+ assertNotNull(record);
+
+ // Wait past expiration (set to 5 sec in config)
+ Thread.sleep(6000);
+
+ // Verify expired
+ RouterState r = routerStore.getRouterRegistration(getRequest).getRouter();
+ assertEquals(RouterServiceState.EXPIRED, r.getStatus());
+
+ // Heartbeat again and this shouldn't be EXPIRED anymore
+ assertTrue(routerStore.routerHeartbeat(request).getStatus());
+ r = routerStore.getRouterRegistration(getRequest).getRouter();
+ assertEquals(RouterServiceState.RUNNING, r.getStatus());
+ }
+
+ @Test
+ public void testGetAllRouterStates()
+ throws StateStoreUnavailableException, IOException {
+
+ // Set 2 entries
+ RouterHeartbeatRequest heartbeatRequest1 =
+ RouterHeartbeatRequest.newInstance(
+ RouterState.newInstance(
+ "testaddress1", Time.now(), RouterServiceState.RUNNING));
+ assertTrue(routerStore.routerHeartbeat(heartbeatRequest1).getStatus());
+
+ RouterHeartbeatRequest heartbeatRequest2 =
+ RouterHeartbeatRequest.newInstance(
+ RouterState.newInstance(
+ "testaddress2", Time.now(), RouterServiceState.RUNNING));
+ assertTrue(routerStore.routerHeartbeat(heartbeatRequest2).getStatus());
+
+ // Verify
+ routerStore.loadCache(true);
+ GetRouterRegistrationsRequest request =
+ GetRouterRegistrationsRequest.newInstance();
+ List<RouterState> entries =
+ routerStore.getRouterRegistrations(request).getRouters();
+ assertEquals(2, entries.size());
+ Collections.sort(entries);
+ assertEquals("testaddress1", entries.get(0).getAddress());
+ assertEquals("testaddress2", entries.get(1).getAddress());
+ assertEquals(RouterServiceState.RUNNING, entries.get(0).getStatus());
+ assertEquals(RouterServiceState.RUNNING, entries.get(1).getStatus());
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/87700d45/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/store/driver/TestStateStoreDriverBase.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/store/driver/TestStateStoreDriverBase.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/store/driver/TestStateStoreDriverBase.java
new file mode 100644
index 0000000..fd29e37
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/store/driver/TestStateStoreDriverBase.java
@@ -0,0 +1,613 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hdfs.server.federation.store.driver;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+import java.io.IOException;
+import java.lang.reflect.Method;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Random;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdfs.server.federation.metrics.StateStoreMetrics;
+import org.apache.hadoop.hdfs.server.federation.resolver.FederationNamenodeServiceState;
+import org.apache.hadoop.hdfs.server.federation.router.RouterServiceState;
+import org.apache.hadoop.hdfs.server.federation.store.FederationStateStoreTestUtils;
+import org.apache.hadoop.hdfs.server.federation.store.StateStoreService;
+import org.apache.hadoop.hdfs.server.federation.store.records.BaseRecord;
+import org.apache.hadoop.hdfs.server.federation.store.records.MembershipState;
+import org.apache.hadoop.hdfs.server.federation.store.records.MountTable;
+import org.apache.hadoop.hdfs.server.federation.store.records.Query;
+import org.apache.hadoop.hdfs.server.federation.store.records.QueryResult;
+import org.apache.hadoop.hdfs.server.federation.store.records.RouterState;
+import org.apache.hadoop.hdfs.server.federation.store.records.StateStoreVersion;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Base tests for the driver. The particular implementations will use this to
+ * test their functionality.
+ */
+public class TestStateStoreDriverBase {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(TestStateStoreDriverBase.class);
+
+ private static StateStoreService stateStore;
+ private static Configuration conf;
+
+ private static final Random RANDOM = new Random();
+
+
+ /**
+ * Get the State Store driver.
+ * @return State Store driver.
+ */
+ protected StateStoreDriver getStateStoreDriver() {
+ return stateStore.getDriver();
+ }
+
+ @After
+ public void cleanMetrics() {
+ if (stateStore != null) {
+ StateStoreMetrics metrics = stateStore.getMetrics();
+ metrics.reset();
+ }
+ }
+
+ @AfterClass
+ public static void tearDownCluster() {
+ if (stateStore != null) {
+ stateStore.stop();
+ }
+ }
+
+ /**
+ * Get a new State Store using this configuration.
+ *
+ * @param config Configuration for the State Store.
+ * @throws Exception If we cannot get the State Store.
+ */
+ public static void getStateStore(Configuration config) throws Exception {
+ conf = config;
+ stateStore = FederationStateStoreTestUtils.newStateStore(conf);
+ }
+
+ private String generateRandomString() {
+ String randomString = "randomString-" + RANDOM.nextInt();
+ return randomString;
+ }
+
+ private long generateRandomLong() {
+ return RANDOM.nextLong();
+ }
+
+ @SuppressWarnings("rawtypes")
+ private <T extends Enum> T generateRandomEnum(Class<T> enumClass) {
+ int x = RANDOM.nextInt(enumClass.getEnumConstants().length);
+ T data = enumClass.getEnumConstants()[x];
+ return data;
+ }
+
+ @SuppressWarnings("unchecked")
+ private <T extends BaseRecord> T generateFakeRecord(Class<T> recordClass)
+ throws IllegalArgumentException, IllegalAccessException, IOException {
+
+ if (recordClass == MembershipState.class) {
+ return (T) MembershipState.newInstance(generateRandomString(),
+ generateRandomString(), generateRandomString(),
+ generateRandomString(), generateRandomString(),
+ generateRandomString(), generateRandomString(),
+ generateRandomString(), generateRandomString(),
+ generateRandomEnum(FederationNamenodeServiceState.class), false);
+ } else if (recordClass == MountTable.class) {
+ String src = "/" + generateRandomString();
+ Map<String, String> destMap = Collections.singletonMap(
+ generateRandomString(), "/" + generateRandomString());
+ return (T) MountTable.newInstance(src, destMap);
+ } else if (recordClass == RouterState.class) {
+ RouterState routerState = RouterState.newInstance(generateRandomString(),
+ generateRandomLong(), generateRandomEnum(RouterServiceState.class));
+ StateStoreVersion version = generateFakeRecord(StateStoreVersion.class);
+ routerState.setStateStoreVersion(version);
+ return (T) routerState;
+ }
+
+ return null;
+ }
+
+ /**
+ * Validate if a record is the same.
+ *
+ * @param original Original record.
+ * @param committed Committed record.
+ * @param assertEquals Assert if the records are equal or just return.
+ * @return If the record is successfully validated.
+ */
+ private boolean validateRecord(
+ BaseRecord original, BaseRecord committed, boolean assertEquals) {
+
+ boolean ret = true;
+
+ Map<String, Class<?>> fields = getFields(original);
+ for (String key : fields.keySet()) {
+ if (key.equals("dateModified") ||
+ key.equals("dateCreated") ||
+ key.equals("proto")) {
+ // Fields are updated/set on commit and fetch and may not match
+ // the fields that are initialized in a non-committed object.
+ continue;
+ }
+ Object data1 = getField(original, key);
+ Object data2 = getField(committed, key);
+ if (assertEquals) {
+ assertEquals("Field " + key + " does not match", data1, data2);
+ } else if (!data1.equals(data2)) {
+ ret = false;
+ }
+ }
+
+ long now = stateStore.getDriver().getTime();
+ assertTrue(
+ committed.getDateCreated() <= now && committed.getDateCreated() > 0);
+ assertTrue(committed.getDateModified() >= committed.getDateCreated());
+
+ return ret;
+ }
+
+ public static void removeAll(StateStoreDriver driver) throws IOException {
+ driver.removeAll(MembershipState.class);
+ driver.removeAll(MountTable.class);
+ }
+
+ public <T extends BaseRecord> void testInsert(
+ StateStoreDriver driver, Class<T> recordClass)
+ throws IllegalArgumentException, IllegalAccessException, IOException {
+
+ assertTrue(driver.removeAll(recordClass));
+ QueryResult<T> queryResult0 = driver.get(recordClass);
+ List<T> records0 = queryResult0.getRecords();
+ assertTrue(records0.isEmpty());
+
+ // Insert single
+ BaseRecord record = generateFakeRecord(recordClass);
+ driver.put(record, true, false);
+
+ // Verify
+ QueryResult<T> queryResult1 = driver.get(recordClass);
+ List<T> records1 = queryResult1.getRecords();
+ assertEquals(1, records1.size());
+ T record0 = records1.get(0);
+ validateRecord(record, record0, true);
+
+ // Insert multiple
+ List<T> insertList = new ArrayList<>();
+ for (int i = 0; i < 10; i++) {
+ T newRecord = generateFakeRecord(recordClass);
+ insertList.add(newRecord);
+ }
+ driver.putAll(insertList, true, false);
+
+ // Verify
+ QueryResult<T> queryResult2 = driver.get(recordClass);
+ List<T> records2 = queryResult2.getRecords();
+ assertEquals(11, records2.size());
+ }
+
+ public <T extends BaseRecord> void testFetchErrors(StateStoreDriver driver,
+ Class<T> clazz) throws IllegalAccessException, IOException {
+
+ // Fetch empty list
+ driver.removeAll(clazz);
+ QueryResult<T> result0 = driver.get(clazz);
+ assertNotNull(result0);
+ List<T> records0 = result0.getRecords();
+ assertEquals(records0.size(), 0);
+
+ // Insert single
+ BaseRecord record = generateFakeRecord(clazz);
+ assertTrue(driver.put(record, true, false));
+
+ // Verify
+ QueryResult<T> result1 = driver.get(clazz);
+ List<T> records1 = result1.getRecords();
+ assertEquals(1, records1.size());
+ validateRecord(record, records1.get(0), true);
+
+ // Test fetch single object with a bad query
+ final T fakeRecord = generateFakeRecord(clazz);
+ final Query<T> query = new Query<T>(fakeRecord);
+ T getRecord = driver.get(clazz, query);
+ assertNull(getRecord);
+
+ // Test fetch multiple objects does not exist returns empty list
+ assertEquals(driver.getMultiple(clazz, query).size(), 0);
+ }
+
+ public <T extends BaseRecord> void testPut(
+ StateStoreDriver driver, Class<T> clazz)
+ throws IllegalArgumentException, ReflectiveOperationException,
+ IOException, SecurityException {
+
+ driver.removeAll(clazz);
+ QueryResult<T> records = driver.get(clazz);
+ assertTrue(records.getRecords().isEmpty());
+
+ // Insert multiple
+ List<T> insertList = new ArrayList<>();
+ for (int i = 0; i < 10; i++) {
+ T newRecord = generateFakeRecord(clazz);
+ insertList.add(newRecord);
+ }
+
+ // Verify
+ assertTrue(driver.putAll(insertList, false, true));
+ records = driver.get(clazz);
+ assertEquals(records.getRecords().size(), 10);
+
+ // Generate a new record with the same PK fields as an existing record
+ BaseRecord updatedRecord = generateFakeRecord(clazz);
+ BaseRecord existingRecord = records.getRecords().get(0);
+ Map<String, String> primaryKeys = existingRecord.getPrimaryKeys();
+ for (Entry<String, String> entry : primaryKeys.entrySet()) {
+ String key = entry.getKey();
+ String value = entry.getValue();
+ Class<?> fieldType = getFieldType(existingRecord, key);
+ Object field = fromString(value, fieldType);
+ assertTrue(setField(updatedRecord, key, field));
+ }
+
+ // Attempt an update of an existing entry, but it is not allowed.
+ assertFalse(driver.put(updatedRecord, false, true));
+
+ // Verify no update occurred, all original records are unchanged
+ QueryResult<T> newRecords = driver.get(clazz);
+ assertTrue(newRecords.getRecords().size() == 10);
+ assertEquals("A single entry was improperly updated in the store", 10,
+ countMatchingEntries(records.getRecords(), newRecords.getRecords()));
+
+ // Update the entry (allowing updates)
+ assertTrue(driver.put(updatedRecord, true, false));
+
+ // Verify that one entry no longer matches the original set
+ newRecords = driver.get(clazz);
+ assertEquals(10, newRecords.getRecords().size());
+ assertEquals(
+ "Record of type " + clazz + " not updated in the store", 9,
+ countMatchingEntries(records.getRecords(), newRecords.getRecords()));
+ }
+
+ private int countMatchingEntries(
+ Collection<? extends BaseRecord> committedList,
+ Collection<? extends BaseRecord> matchList) {
+
+ int matchingCount = 0;
+ for (BaseRecord committed : committedList) {
+ for (BaseRecord match : matchList) {
+ try {
+ if (match.getPrimaryKey().equals(committed.getPrimaryKey())) {
+ if (validateRecord(match, committed, false)) {
+ matchingCount++;
+ }
+ break;
+ }
+ } catch (Exception ex) {
+ }
+ }
+ }
+ return matchingCount;
+ }
+
+ public <T extends BaseRecord> void testRemove(
+ StateStoreDriver driver, Class<T> clazz)
+ throws IllegalArgumentException, IllegalAccessException, IOException {
+
+ // Remove all
+ assertTrue(driver.removeAll(clazz));
+ QueryResult<T> records = driver.get(clazz);
+ assertTrue(records.getRecords().isEmpty());
+
+ // Insert multiple
+ List<T> insertList = new ArrayList<>();
+ for (int i = 0; i < 10; i++) {
+ T newRecord = generateFakeRecord(clazz);
+ insertList.add(newRecord);
+ }
+
+ // Verify
+ assertTrue(driver.putAll(insertList, false, true));
+ records = driver.get(clazz);
+ assertEquals(records.getRecords().size(), 10);
+
+ // Remove Single
+ assertTrue(driver.remove(records.getRecords().get(0)));
+
+ // Verify
+ records = driver.get(clazz);
+ assertEquals(records.getRecords().size(), 9);
+
+ // Remove with filter
+ final T firstRecord = records.getRecords().get(0);
+ final Query<T> query0 = new Query<T>(firstRecord);
+ assertTrue(driver.remove(clazz, query0) > 0);
+
+ final T secondRecord = records.getRecords().get(1);
+ final Query<T> query1 = new Query<T>(secondRecord);
+ assertTrue(driver.remove(clazz, query1) > 0);
+
+ // Verify
+ records = driver.get(clazz);
+ assertEquals(records.getRecords().size(), 7);
+
+ // Remove all
+ assertTrue(driver.removeAll(clazz));
+
+ // Verify
+ records = driver.get(clazz);
+ assertTrue(records.getRecords().isEmpty());
+ }
+
+ public void testInsert(StateStoreDriver driver)
+ throws IllegalArgumentException, IllegalAccessException, IOException {
+ testInsert(driver, MembershipState.class);
+ testInsert(driver, MountTable.class);
+ }
+
+ public void testPut(StateStoreDriver driver)
+ throws IllegalArgumentException, ReflectiveOperationException,
+ IOException, SecurityException {
+ testPut(driver, MembershipState.class);
+ testPut(driver, MountTable.class);
+ }
+
+ public void testRemove(StateStoreDriver driver)
+ throws IllegalArgumentException, IllegalAccessException, IOException {
+ testRemove(driver, MembershipState.class);
+ testRemove(driver, MountTable.class);
+ }
+
+ public void testFetchErrors(StateStoreDriver driver)
+ throws IllegalArgumentException, IllegalAccessException, IOException {
+ testFetchErrors(driver, MembershipState.class);
+ testFetchErrors(driver, MountTable.class);
+ }
+
+ public void testMetrics(StateStoreDriver driver)
+ throws IOException, IllegalArgumentException, IllegalAccessException {
+
+ MountTable insertRecord =
+ this.generateFakeRecord(MountTable.class);
+
+ // Put single
+ StateStoreMetrics metrics = stateStore.getMetrics();
+ assertEquals(0, metrics.getWriteOps());
+ driver.put(insertRecord, true, false);
+ assertEquals(1, metrics.getWriteOps());
+
+ // Put multiple
+ metrics.reset();
+ assertEquals(0, metrics.getWriteOps());
+ driver.put(insertRecord, true, false);
+ assertEquals(1, metrics.getWriteOps());
+
+ // Get Single
+ metrics.reset();
+ assertEquals(0, metrics.getReadOps());
+
+ final String querySourcePath = insertRecord.getSourcePath();
+ MountTable partial = MountTable.newInstance();
+ partial.setSourcePath(querySourcePath);
+ final Query<MountTable> query = new Query<>(partial);
+ driver.get(MountTable.class, query);
+ assertEquals(1, metrics.getReadOps());
+
+ // GetAll
+ metrics.reset();
+ assertEquals(0, metrics.getReadOps());
+ driver.get(MountTable.class);
+ assertEquals(1, metrics.getReadOps());
+
+ // GetMultiple
+ metrics.reset();
+ assertEquals(0, metrics.getReadOps());
+ driver.getMultiple(MountTable.class, query);
+ assertEquals(1, metrics.getReadOps());
+
+ // Insert fails
+ metrics.reset();
+ assertEquals(0, metrics.getFailureOps());
+ driver.put(insertRecord, false, true);
+ assertEquals(1, metrics.getFailureOps());
+
+ // Remove single
+ metrics.reset();
+ assertEquals(0, metrics.getRemoveOps());
+ driver.remove(insertRecord);
+ assertEquals(1, metrics.getRemoveOps());
+
+ // Remove multiple
+ metrics.reset();
+ driver.put(insertRecord, true, false);
+ assertEquals(0, metrics.getRemoveOps());
+ driver.remove(MountTable.class, query);
+ assertEquals(1, metrics.getRemoveOps());
+
+ // Remove all
+ metrics.reset();
+ driver.put(insertRecord, true, false);
+ assertEquals(0, metrics.getRemoveOps());
+ driver.removeAll(MountTable.class);
+ assertEquals(1, metrics.getRemoveOps());
+ }
+
+ /**
+ * Sets the value of a field on the object.
+ *
+ * @param fieldName The string name of the field.
+ * @param data The data to pass to the field's setter.
+ *
+ * @return True if successful, fails if failed.
+ */
+ private static boolean setField(
+ BaseRecord record, String fieldName, Object data) {
+
+ Method m = locateSetter(record, fieldName);
+ if (m != null) {
+ try {
+ m.invoke(record, data);
+ } catch (Exception e) {
+ LOG.error("Cannot set field " + fieldName + " on object "
+ + record.getClass().getName() + " to data " + data + " of type "
+ + data.getClass(), e);
+ return false;
+ }
+ }
+ return true;
+ }
+
+ /**
+ * Finds the appropriate setter for a field name.
+ *
+ * @param fieldName The legacy name of the field.
+ * @return The matching setter or null if not found.
+ */
+ private static Method locateSetter(BaseRecord record, String fieldName) {
+ for (Method m : record.getClass().getMethods()) {
+ if (m.getName().equalsIgnoreCase("set" + fieldName)) {
+ return m;
+ }
+ }
+ return null;
+ }
+
+ /**
+ * Returns all serializable fields in the object.
+ *
+ * @return Map with the fields.
+ */
+ private static Map<String, Class<?>> getFields(BaseRecord record) {
+ Map<String, Class<?>> getters = new HashMap<>();
+ for (Method m : record.getClass().getDeclaredMethods()) {
+ if (m.getName().startsWith("get")) {
+ try {
+ Class<?> type = m.getReturnType();
+ char[] c = m.getName().substring(3).toCharArray();
+ c[0] = Character.toLowerCase(c[0]);
+ String key = new String(c);
+ getters.put(key, type);
+ } catch (Exception e) {
+ LOG.error("Cannot execute getter " + m.getName()
+ + " on object " + record);
+ }
+ }
+ }
+ return getters;
+ }
+
+ /**
+ * Get the type of a field.
+ *
+ * @param fieldName
+ * @return Field type
+ */
+ private static Class<?> getFieldType(BaseRecord record, String fieldName) {
+ Method m = locateGetter(record, fieldName);
+ return m.getReturnType();
+ }
+
+ /**
+ * Fetches the value for a field name.
+ *
+ * @param fieldName the legacy name of the field.
+ * @return The field data or null if not found.
+ */
+ private static Object getField(BaseRecord record, String fieldName) {
+ Object result = null;
+ Method m = locateGetter(record, fieldName);
+ if (m != null) {
+ try {
+ result = m.invoke(record);
+ } catch (Exception e) {
+ LOG.error("Cannot get field " + fieldName + " on object " + record);
+ }
+ }
+ return result;
+ }
+
+ /**
+ * Finds the appropriate getter for a field name.
+ *
+ * @param fieldName The legacy name of the field.
+ * @return The matching getter or null if not found.
+ */
+ private static Method locateGetter(BaseRecord record, String fieldName) {
+ for (Method m : record.getClass().getMethods()) {
+ if (m.getName().equalsIgnoreCase("get" + fieldName)) {
+ return m;
+ }
+ }
+ return null;
+ }
+
+ /**
+ * Expands a data object from the store into an record object. Default store
+ * data type is a String. Override if additional serialization is required.
+ *
+ * @param data Object containing the serialized data. Only string is
+ * supported.
+ * @param clazz Target object class to hold the deserialized data.
+ * @return An instance of the target data object initialized with the
+ * deserialized data.
+ */
+ @Deprecated
+ @SuppressWarnings({ "unchecked", "rawtypes" })
+ private static <T> T fromString(String data, Class<T> clazz) {
+
+ if (data.equals("null")) {
+ return null;
+ } else if (clazz == String.class) {
+ return (T) data;
+ } else if (clazz == Long.class || clazz == long.class) {
+ return (T) Long.valueOf(data);
+ } else if (clazz == Integer.class || clazz == int.class) {
+ return (T) Integer.valueOf(data);
+ } else if (clazz == Double.class || clazz == double.class) {
+ return (T) Double.valueOf(data);
+ } else if (clazz == Float.class || clazz == float.class) {
+ return (T) Float.valueOf(data);
+ } else if (clazz == Boolean.class || clazz == boolean.class) {
+ return (T) Boolean.valueOf(data);
+ } else if (clazz.isEnum()) {
+ return (T) Enum.valueOf((Class<Enum>) clazz, data);
+ }
+ return null;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/87700d45/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/store/driver/TestStateStoreFile.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/store/driver/TestStateStoreFile.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/store/driver/TestStateStoreFile.java
new file mode 100644
index 0000000..a8a9020
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/store/driver/TestStateStoreFile.java
@@ -0,0 +1,76 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hdfs.server.federation.store.driver;
+
+import static org.apache.hadoop.hdfs.server.federation.store.FederationStateStoreTestUtils.getStateStoreConfiguration;
+
+import java.io.IOException;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdfs.server.federation.store.driver.impl.StateStoreFileImpl;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+/**
+ * Test the FileSystem (e.g., HDFS) implementation of the State Store driver.
+ */
+public class TestStateStoreFile extends TestStateStoreDriverBase {
+
+ @BeforeClass
+ public static void setupCluster() throws Exception {
+ Configuration conf = getStateStoreConfiguration(StateStoreFileImpl.class);
+ getStateStore(conf);
+ }
+
+ @Before
+ public void startup() throws IOException {
+ removeAll(getStateStoreDriver());
+ }
+
+ @Test
+ public void testInsert()
+ throws IllegalArgumentException, IllegalAccessException, IOException {
+ testInsert(getStateStoreDriver());
+ }
+
+ @Test
+ public void testUpdate()
+ throws IllegalArgumentException, ReflectiveOperationException,
+ IOException, SecurityException {
+ testPut(getStateStoreDriver());
+ }
+
+ @Test
+ public void testDelete()
+ throws IllegalArgumentException, IllegalAccessException, IOException {
+ testRemove(getStateStoreDriver());
+ }
+
+ @Test
+ public void testFetchErrors()
+ throws IllegalArgumentException, IllegalAccessException, IOException {
+ testFetchErrors(getStateStoreDriver());
+ }
+
+ @Test
+ public void testMetrics()
+ throws IllegalArgumentException, IllegalAccessException, IOException {
+ testMetrics(getStateStoreDriver());
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/87700d45/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/store/driver/TestStateStoreFileBase.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/store/driver/TestStateStoreFileBase.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/store/driver/TestStateStoreFileBase.java
new file mode 100644
index 0000000..9adfe33
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/store/driver/TestStateStoreFileBase.java
@@ -0,0 +1,47 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hdfs.server.federation.store.driver;
+
+import static org.apache.hadoop.hdfs.server.federation.store.driver.impl.StateStoreFileBaseImpl.isOldTempRecord;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import java.util.concurrent.TimeUnit;
+
+import org.apache.hadoop.util.Time;
+import org.junit.Test;
+
+/**
+ * Tests for the State Store file based implementation.
+ */
+public class TestStateStoreFileBase {
+
+ @Test
+ public void testTempOld() {
+ assertFalse(isOldTempRecord("test.txt"));
+ assertFalse(isOldTempRecord("testfolder/test.txt"));
+
+ long tnow = Time.now();
+ String tmpFile1 = "test." + tnow + ".tmp";
+ assertFalse(isOldTempRecord(tmpFile1));
+
+ long told = Time.now() - TimeUnit.MINUTES.toMillis(1);
+ String tmpFile2 = "test." + told + ".tmp";
+ assertTrue(isOldTempRecord(tmpFile2));
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/87700d45/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/store/driver/TestStateStoreFileSystem.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/store/driver/TestStateStoreFileSystem.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/store/driver/TestStateStoreFileSystem.java
new file mode 100644
index 0000000..8c4b188
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/store/driver/TestStateStoreFileSystem.java
@@ -0,0 +1,94 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hdfs.server.federation.store.driver;
+
+import java.io.IOException;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdfs.MiniDFSCluster;
+import org.apache.hadoop.hdfs.server.federation.store.FederationStateStoreTestUtils;
+import org.apache.hadoop.hdfs.server.federation.store.driver.impl.StateStoreFileSystemImpl;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+/**
+ * Test the FileSystem (e.g., HDFS) implementation of the State Store driver.
+ */
+public class TestStateStoreFileSystem extends TestStateStoreDriverBase {
+
+ private static MiniDFSCluster dfsCluster;
+
+ @BeforeClass
+ public static void setupCluster() throws Exception {
+ Configuration conf = FederationStateStoreTestUtils
+ .getStateStoreConfiguration(StateStoreFileSystemImpl.class);
+ conf.set(StateStoreFileSystemImpl.FEDERATION_STORE_FS_PATH,
+ "/hdfs-federation/");
+
+ // Create HDFS cluster to back the state tore
+ MiniDFSCluster.Builder builder = new MiniDFSCluster.Builder(conf);
+ builder.numDataNodes(1);
+ dfsCluster = builder.build();
+ dfsCluster.waitClusterUp();
+ getStateStore(conf);
+ }
+
+ @AfterClass
+ public static void tearDownCluster() {
+ if (dfsCluster != null) {
+ dfsCluster.shutdown();
+ }
+ }
+
+ @Before
+ public void startup() throws IOException {
+ removeAll(getStateStoreDriver());
+ }
+
+ @Test
+ public void testInsert()
+ throws IllegalArgumentException, IllegalAccessException, IOException {
+ testInsert(getStateStoreDriver());
+ }
+
+ @Test
+ public void testUpdate() throws IllegalArgumentException, IOException,
+ SecurityException, ReflectiveOperationException {
+ testPut(getStateStoreDriver());
+ }
+
+ @Test
+ public void testDelete()
+ throws IllegalArgumentException, IllegalAccessException, IOException {
+ testRemove(getStateStoreDriver());
+ }
+
+ @Test
+ public void testFetchErrors()
+ throws IllegalArgumentException, IllegalAccessException, IOException {
+ testFetchErrors(getStateStoreDriver());
+ }
+
+ @Test
+ public void testMetrics()
+ throws IllegalArgumentException, IllegalAccessException, IOException {
+ testMetrics(getStateStoreDriver());
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/87700d45/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/store/driver/TestStateStoreZK.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/store/driver/TestStateStoreZK.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/store/driver/TestStateStoreZK.java
new file mode 100644
index 0000000..3cf7c91
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/store/driver/TestStateStoreZK.java
@@ -0,0 +1,105 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hdfs.server.federation.store.driver;
+
+import static org.apache.hadoop.hdfs.server.federation.store.FederationStateStoreTestUtils.getStateStoreConfiguration;
+
+import java.io.IOException;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.retry.RetryNTimes;
+import org.apache.curator.test.TestingServer;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.CommonConfigurationKeys;
+import org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys;
+import org.apache.hadoop.hdfs.server.federation.store.driver.impl.StateStoreZooKeeperImpl;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+/**
+ * Test the ZooKeeper implementation of the State Store driver.
+ */
+public class TestStateStoreZK extends TestStateStoreDriverBase {
+
+ private static TestingServer curatorTestingServer;
+ private static CuratorFramework curatorFramework;
+
+ @BeforeClass
+ public static void setupCluster() throws Exception {
+ curatorTestingServer = new TestingServer();
+ curatorTestingServer.start();
+ String connectString = curatorTestingServer.getConnectString();
+ curatorFramework = CuratorFrameworkFactory.builder()
+ .connectString(connectString)
+ .retryPolicy(new RetryNTimes(100, 100))
+ .build();
+ curatorFramework.start();
+
+ // Create the ZK State Store
+ Configuration conf =
+ getStateStoreConfiguration(StateStoreZooKeeperImpl.class);
+ conf.set(CommonConfigurationKeys.ZK_ADDRESS, connectString);
+ // Disable auto-repair of connection
+ conf.setLong(RBFConfigKeys.FEDERATION_STORE_CONNECTION_TEST_MS,
+ TimeUnit.HOURS.toMillis(1));
+ getStateStore(conf);
+ }
+
+ @AfterClass
+ public static void tearDownCluster() {
+ curatorFramework.close();
+ try {
+ curatorTestingServer.stop();
+ } catch (IOException e) {
+ }
+ }
+
+ @Before
+ public void startup() throws IOException {
+ removeAll(getStateStoreDriver());
+ }
+
+ @Test
+ public void testInsert()
+ throws IllegalArgumentException, IllegalAccessException, IOException {
+ testInsert(getStateStoreDriver());
+ }
+
+ @Test
+ public void testUpdate()
+ throws IllegalArgumentException, ReflectiveOperationException,
+ IOException, SecurityException {
+ testPut(getStateStoreDriver());
+ }
+
+ @Test
+ public void testDelete()
+ throws IllegalArgumentException, IllegalAccessException, IOException {
+ testRemove(getStateStoreDriver());
+ }
+
+ @Test
+ public void testFetchErrors()
+ throws IllegalArgumentException, IllegalAccessException, IOException {
+ testFetchErrors(getStateStoreDriver());
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/87700d45/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/store/records/TestMembershipState.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/store/records/TestMembershipState.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/store/records/TestMembershipState.java
new file mode 100644
index 0000000..d922414
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/store/records/TestMembershipState.java
@@ -0,0 +1,129 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hdfs.server.federation.store.records;
+
+import static org.junit.Assert.assertEquals;
+
+import java.io.IOException;
+
+import org.apache.hadoop.hdfs.server.federation.resolver.FederationNamenodeServiceState;
+import org.apache.hadoop.hdfs.server.federation.store.driver.StateStoreSerializer;
+import org.junit.Test;
+
+/**
+ * Test the Membership State records.
+ */
+public class TestMembershipState {
+
+ private static final String ROUTER = "router";
+ private static final String NAMESERVICE = "nameservice";
+ private static final String NAMENODE = "namenode";
+ private static final String CLUSTER_ID = "cluster";
+ private static final String BLOCKPOOL_ID = "blockpool";
+ private static final String RPC_ADDRESS = "rpcaddress";
+ private static final String SERVICE_ADDRESS = "serviceaddress";
+ private static final String LIFELINE_ADDRESS = "lifelineaddress";
+ private static final String WEB_ADDRESS = "webaddress";
+ private static final boolean SAFE_MODE = false;
+
+ private static final long DATE_CREATED = 100;
+ private static final long DATE_MODIFIED = 200;
+
+ private static final long NUM_BLOCKS = 300;
+ private static final long NUM_FILES = 400;
+ private static final int NUM_DEAD = 500;
+ private static final int NUM_ACTIVE = 600;
+ private static final int NUM_DECOM = 700;
+ private static final int NUM_DECOM_ACTIVE = 800;
+ private static final int NUM_DECOM_DEAD = 900;
+ private static final long NUM_BLOCK_MISSING = 1000;
+
+ private static final long TOTAL_SPACE = 1100;
+ private static final long AVAILABLE_SPACE = 1200;
+
+ private static final FederationNamenodeServiceState STATE =
+ FederationNamenodeServiceState.ACTIVE;
+
+ private MembershipState createRecord() throws IOException {
+
+ MembershipState record = MembershipState.newInstance(
+ ROUTER, NAMESERVICE, NAMENODE, CLUSTER_ID,
+ BLOCKPOOL_ID, RPC_ADDRESS, SERVICE_ADDRESS, LIFELINE_ADDRESS,
+ WEB_ADDRESS, STATE, SAFE_MODE);
+ record.setDateCreated(DATE_CREATED);
+ record.setDateModified(DATE_MODIFIED);
+
+ MembershipStats stats = MembershipStats.newInstance();
+ stats.setNumOfBlocks(NUM_BLOCKS);
+ stats.setNumOfFiles(NUM_FILES);
+ stats.setNumOfActiveDatanodes(NUM_ACTIVE);
+ stats.setNumOfDeadDatanodes(NUM_DEAD);
+ stats.setNumOfDecommissioningDatanodes(NUM_DECOM);
+ stats.setNumOfDecomActiveDatanodes(NUM_DECOM_ACTIVE);
+ stats.setNumOfDecomDeadDatanodes(NUM_DECOM_DEAD);
+ stats.setNumOfBlocksMissing(NUM_BLOCK_MISSING);
+ stats.setTotalSpace(TOTAL_SPACE);
+ stats.setAvailableSpace(AVAILABLE_SPACE);
+ record.setStats(stats);
+ return record;
+ }
+
+ private void validateRecord(MembershipState record) throws IOException {
+
+ assertEquals(ROUTER, record.getRouterId());
+ assertEquals(NAMESERVICE, record.getNameserviceId());
+ assertEquals(CLUSTER_ID, record.getClusterId());
+ assertEquals(BLOCKPOOL_ID, record.getBlockPoolId());
+ assertEquals(RPC_ADDRESS, record.getRpcAddress());
+ assertEquals(WEB_ADDRESS, record.getWebAddress());
+ assertEquals(STATE, record.getState());
+ assertEquals(SAFE_MODE, record.getIsSafeMode());
+ assertEquals(DATE_CREATED, record.getDateCreated());
+ assertEquals(DATE_MODIFIED, record.getDateModified());
+
+ MembershipStats stats = record.getStats();
+ assertEquals(NUM_BLOCKS, stats.getNumOfBlocks());
+ assertEquals(NUM_FILES, stats.getNumOfFiles());
+ assertEquals(NUM_ACTIVE, stats.getNumOfActiveDatanodes());
+ assertEquals(NUM_DEAD, stats.getNumOfDeadDatanodes());
+ assertEquals(NUM_DECOM, stats.getNumOfDecommissioningDatanodes());
+ assertEquals(NUM_DECOM_ACTIVE, stats.getNumOfDecomActiveDatanodes());
+ assertEquals(NUM_DECOM_DEAD, stats.getNumOfDecomDeadDatanodes());
+ assertEquals(TOTAL_SPACE, stats.getTotalSpace());
+ assertEquals(AVAILABLE_SPACE, stats.getAvailableSpace());
+ }
+
+ @Test
+ public void testGetterSetter() throws IOException {
+ MembershipState record = createRecord();
+ validateRecord(record);
+ }
+
+ @Test
+ public void testSerialization() throws IOException {
+
+ MembershipState record = createRecord();
+
+ StateStoreSerializer serializer = StateStoreSerializer.getSerializer();
+ String serializedString = serializer.serializeString(record);
+ MembershipState newRecord =
+ serializer.deserialize(serializedString, MembershipState.class);
+
+ validateRecord(newRecord);
+ }
+}
\ No newline at end of file
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org