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:12 UTC
[36/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/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterSafemodeService.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterSafemodeService.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterSafemodeService.java
new file mode 100644
index 0000000..5dfb356
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterSafemodeService.java
@@ -0,0 +1,149 @@
+/**
+ * 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.router;
+
+import static org.apache.hadoop.util.Time.now;
+
+import java.util.concurrent.TimeUnit;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdfs.server.federation.store.StateStoreService;
+import org.apache.hadoop.util.Time;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Service to periodically check if the {@link org.apache.hadoop.hdfs.server.
+ * federation.store.StateStoreService StateStoreService} cached information in
+ * the {@link Router} is up to date. This is for performance and removes the
+ * {@link org.apache.hadoop.hdfs.server.federation.store.StateStoreService
+ * StateStoreService} from the critical path in common operations.
+ */
+public class RouterSafemodeService extends PeriodicService {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(RouterSafemodeService.class);
+
+ /** Router to manage safe mode. */
+ private final Router router;
+
+ /** Interval in ms to wait post startup before allowing RPC requests. */
+ private long startupInterval;
+ /** Interval in ms after which the State Store cache is too stale. */
+ private long staleInterval;
+ /** Start time in ms of this service. */
+ private long startupTime;
+
+ /** The time the Router enters safe mode in milliseconds. */
+ private long enterSafeModeTime = now();
+
+
+ /**
+ * Create a new Cache update service.
+ *
+ * @param router Router containing the cache.
+ */
+ public RouterSafemodeService(Router router) {
+ super(RouterSafemodeService.class.getSimpleName());
+ this.router = router;
+ }
+
+ /**
+ * Enter safe mode.
+ */
+ private void enter() {
+ LOG.info("Entering safe mode");
+ enterSafeModeTime = now();
+ RouterRpcServer rpcServer = router.getRpcServer();
+ rpcServer.setSafeMode(true);
+ router.updateRouterState(RouterServiceState.SAFEMODE);
+ }
+
+ /**
+ * Leave safe mode.
+ */
+ private void leave() {
+ // Cache recently updated, leave safemode
+ long timeInSafemode = now() - enterSafeModeTime;
+ LOG.info("Leaving safe mode after {} milliseconds", timeInSafemode);
+ RouterMetrics routerMetrics = router.getRouterMetrics();
+ if (routerMetrics == null) {
+ LOG.error("The Router metrics are not enabled");
+ } else {
+ routerMetrics.setSafeModeTime(timeInSafemode);
+ }
+ RouterRpcServer rpcServer = router.getRpcServer();
+ rpcServer.setSafeMode(false);
+ router.updateRouterState(RouterServiceState.RUNNING);
+ }
+
+ @Override
+ protected void serviceInit(Configuration conf) throws Exception {
+
+ // Use same interval as cache update service
+ this.setIntervalMs(conf.getTimeDuration(
+ RBFConfigKeys.DFS_ROUTER_CACHE_TIME_TO_LIVE_MS,
+ RBFConfigKeys.DFS_ROUTER_CACHE_TIME_TO_LIVE_MS_DEFAULT,
+ TimeUnit.MILLISECONDS));
+
+ this.startupInterval = conf.getTimeDuration(
+ RBFConfigKeys.DFS_ROUTER_SAFEMODE_EXTENSION,
+ RBFConfigKeys.DFS_ROUTER_SAFEMODE_EXTENSION_DEFAULT,
+ TimeUnit.MILLISECONDS);
+ LOG.info("Leave startup safe mode after {} ms", this.startupInterval);
+
+ this.staleInterval = conf.getTimeDuration(
+ RBFConfigKeys.DFS_ROUTER_SAFEMODE_EXPIRATION,
+ RBFConfigKeys.DFS_ROUTER_SAFEMODE_EXPIRATION_DEFAULT,
+ TimeUnit.MILLISECONDS);
+ LOG.info("Enter safe mode after {} ms without reaching the State Store",
+ this.staleInterval);
+
+ this.startupTime = Time.now();
+
+ // Initializing the RPC server in safe mode, it will disable it later
+ enter();
+
+ super.serviceInit(conf);
+ }
+
+ @Override
+ public void periodicInvoke() {
+ long now = Time.now();
+ long delta = now - startupTime;
+ if (delta < startupInterval) {
+ LOG.info("Delaying safemode exit for {} milliseconds...",
+ this.startupInterval - delta);
+ return;
+ }
+ RouterRpcServer rpcServer = router.getRpcServer();
+ StateStoreService stateStore = router.getStateStore();
+ long cacheUpdateTime = stateStore.getCacheUpdateTime();
+ boolean isCacheStale = (now - cacheUpdateTime) > this.staleInterval;
+
+ // Always update to indicate our cache was updated
+ if (isCacheStale) {
+ if (!rpcServer.isInSafeMode()) {
+ enter();
+ }
+ } else if (rpcServer.isInSafeMode()) {
+ // Cache recently updated, leave safe mode
+ leave();
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/87700d45/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterServiceState.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterServiceState.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterServiceState.java
new file mode 100644
index 0000000..3accbe9
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterServiceState.java
@@ -0,0 +1,31 @@
+/**
+ * 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.router;
+
+/**
+ * States of the Router.
+ */
+public enum RouterServiceState {
+ UNINITIALIZED,
+ INITIALIZING,
+ SAFEMODE,
+ RUNNING,
+ STOPPING,
+ SHUTDOWN,
+ EXPIRED;
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/87700d45/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterStateManager.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterStateManager.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterStateManager.java
new file mode 100644
index 0000000..527600c
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterStateManager.java
@@ -0,0 +1,50 @@
+/**
+ * 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.router;
+
+import java.io.IOException;
+
+import org.apache.hadoop.hdfs.server.federation.store.protocol.EnterSafeModeRequest;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.EnterSafeModeResponse;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.GetSafeModeRequest;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.GetSafeModeResponse;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.LeaveSafeModeRequest;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.LeaveSafeModeResponse;
+
+/**
+ * Interface of managing the Router state.
+ */
+public interface RouterStateManager {
+ /**
+ * Enter safe mode and change Router state to RouterServiceState#SAFEMODE.
+ */
+ EnterSafeModeResponse enterSafeMode(EnterSafeModeRequest request)
+ throws IOException;
+
+ /**
+ * Leave safe mode and change Router state to RouterServiceState#RUNNING.
+ */
+ LeaveSafeModeResponse leaveSafeMode(LeaveSafeModeRequest request)
+ throws IOException;
+
+ /**
+ * Verify if current Router state is safe mode.
+ */
+ GetSafeModeResponse getSafeMode(GetSafeModeRequest request)
+ throws IOException;
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/87700d45/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/package-info.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/package-info.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/package-info.java
new file mode 100644
index 0000000..327f39b
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/package-info.java
@@ -0,0 +1,31 @@
+/**
+ * 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.
+ */
+
+/**
+ * The router package includes the core services for a HDFS federation router.
+ * The {@link Router} acts as a transparent proxy in front of a cluster of
+ * multiple NameNodes and nameservices. The {@link RouterRpcServer} exposes the
+ * NameNode clientProtocol and is the primary contact point for DFS clients in a
+ * federated cluster.
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Evolving
+package org.apache.hadoop.hdfs.server.federation.router;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/87700d45/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/CachedRecordStore.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/CachedRecordStore.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/CachedRecordStore.java
new file mode 100644
index 0000000..cdd4449
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/CachedRecordStore.java
@@ -0,0 +1,242 @@
+/**
+ * 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 java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+import org.apache.hadoop.hdfs.server.federation.metrics.StateStoreMetrics;
+import org.apache.hadoop.hdfs.server.federation.store.driver.StateStoreDriver;
+import org.apache.hadoop.hdfs.server.federation.store.records.BaseRecord;
+import org.apache.hadoop.hdfs.server.federation.store.records.QueryResult;
+import org.apache.hadoop.util.Time;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Record store that takes care of caching the records in memory.
+ *
+ * @param <R> Record to store by this interface.
+ */
+public abstract class CachedRecordStore<R extends BaseRecord>
+ extends RecordStore<R> implements StateStoreCache {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(CachedRecordStore.class);
+
+
+ /** Prevent loading the cache more than once every 500 ms. */
+ private static final long MIN_UPDATE_MS = 500;
+
+
+ /** Cached entries. */
+ private List<R> records = new ArrayList<>();
+
+ /** Time stamp of the cached entries. */
+ private long timestamp = -1;
+
+ /** If the cache is initialized. */
+ private boolean initialized = false;
+
+ /** Last time the cache was updated. */
+ private long lastUpdate = -1;
+
+ /** Lock to access the memory cache. */
+ private final ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
+ private final Lock readLock = readWriteLock.readLock();
+ private final Lock writeLock = readWriteLock.writeLock();
+
+ /** If it should override the expired values when loading the cache. */
+ private boolean override = false;
+
+
+ /**
+ * Create a new cached record store.
+ *
+ * @param clazz Class of the record to store.
+ * @param driver State Store driver.
+ */
+ protected CachedRecordStore(Class<R> clazz, StateStoreDriver driver) {
+ this(clazz, driver, false);
+ }
+
+ /**
+ * Create a new cached record store.
+ *
+ * @param clazz Class of the record to store.
+ * @param driver State Store driver.
+ * @param over If the entries should be override if they expire
+ */
+ protected CachedRecordStore(
+ Class<R> clazz, StateStoreDriver driver, boolean over) {
+ super(clazz, driver);
+
+ this.override = over;
+ }
+
+ /**
+ * Check that the cache of the State Store information is available.
+ *
+ * @throws StateStoreUnavailableException If the cache is not initialized.
+ */
+ private void checkCacheAvailable() throws StateStoreUnavailableException {
+ if (!this.initialized) {
+ throw new StateStoreUnavailableException(
+ "Cached State Store not initialized, " +
+ getRecordClass().getSimpleName() + " records not valid");
+ }
+ }
+
+ @Override
+ public boolean loadCache(boolean force) throws IOException {
+ // Prevent loading the cache too frequently
+ if (force || isUpdateTime()) {
+ List<R> newRecords = null;
+ long t = -1;
+ try {
+ QueryResult<R> result = getDriver().get(getRecordClass());
+ newRecords = result.getRecords();
+ t = result.getTimestamp();
+
+ // If we have any expired record, update the State Store
+ if (this.override) {
+ overrideExpiredRecords(result);
+ }
+ } catch (IOException e) {
+ LOG.error("Cannot get \"{}\" records from the State Store",
+ getRecordClass().getSimpleName());
+ this.initialized = false;
+ return false;
+ }
+
+ // Update cache atomically
+ writeLock.lock();
+ try {
+ this.records.clear();
+ this.records.addAll(newRecords);
+ this.timestamp = t;
+ this.initialized = true;
+ } finally {
+ writeLock.unlock();
+ }
+
+ // Update the metrics for the cache State Store size
+ StateStoreMetrics metrics = getDriver().getMetrics();
+ if (metrics != null) {
+ String recordName = getRecordClass().getSimpleName();
+ metrics.setCacheSize(recordName, this.records.size());
+ }
+
+ lastUpdate = Time.monotonicNow();
+ }
+ return true;
+ }
+
+ /**
+ * Check if it's time to update the cache. Update it it was never updated.
+ *
+ * @return If it's time to update this cache.
+ */
+ private boolean isUpdateTime() {
+ return Time.monotonicNow() - lastUpdate > MIN_UPDATE_MS;
+ }
+
+ /**
+ * Updates the state store with any record overrides we detected, such as an
+ * expired state.
+ *
+ * @param query RecordQueryResult containing the data to be inspected.
+ * @throws IOException
+ */
+ public void overrideExpiredRecords(QueryResult<R> query) throws IOException {
+ List<R> commitRecords = new ArrayList<>();
+ List<R> newRecords = query.getRecords();
+ long currentDriverTime = query.getTimestamp();
+ if (newRecords == null || currentDriverTime <= 0) {
+ LOG.error("Cannot check overrides for record");
+ return;
+ }
+ for (R record : newRecords) {
+ if (record.checkExpired(currentDriverTime)) {
+ String recordName = StateStoreUtils.getRecordName(record.getClass());
+ LOG.info("Override State Store record {}: {}", recordName, record);
+ commitRecords.add(record);
+ }
+ }
+ if (commitRecords.size() > 0) {
+ getDriver().putAll(commitRecords, true, false);
+ }
+ }
+
+ /**
+ * Updates the state store with any record overrides we detected, such as an
+ * expired state.
+ *
+ * @param record Record record to be updated.
+ * @throws IOException
+ */
+ public void overrideExpiredRecord(R record) throws IOException {
+ List<R> newRecords = Collections.singletonList(record);
+ long time = getDriver().getTime();
+ QueryResult<R> query = new QueryResult<>(newRecords, time);
+ overrideExpiredRecords(query);
+ }
+
+ /**
+ * Get all the cached records.
+ *
+ * @return Copy of the cached records.
+ * @throws StateStoreUnavailableException If the State store is not available.
+ */
+ public List<R> getCachedRecords() throws StateStoreUnavailableException {
+ checkCacheAvailable();
+
+ List<R> ret = new LinkedList<R>();
+ this.readLock.lock();
+ try {
+ ret.addAll(this.records);
+ } finally {
+ this.readLock.unlock();
+ }
+ return ret;
+ }
+
+ /**
+ * Get all the cached records and the time stamp of the cache.
+ *
+ * @return Copy of the cached records and the time stamp.
+ * @throws StateStoreUnavailableException If the State store is not available.
+ */
+ protected QueryResult<R> getCachedRecordsAndTimeStamp()
+ throws StateStoreUnavailableException {
+ checkCacheAvailable();
+
+ this.readLock.lock();
+ try {
+ return new QueryResult<R>(this.records, this.timestamp);
+ } finally {
+ this.readLock.unlock();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/87700d45/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/MembershipStore.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/MembershipStore.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/MembershipStore.java
new file mode 100644
index 0000000..3e8ba6b
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/MembershipStore.java
@@ -0,0 +1,126 @@
+/**
+ * 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 java.io.IOException;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.hdfs.server.federation.store.driver.StateStoreDriver;
+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.GetNamespaceInfoRequest;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.GetNamespaceInfoResponse;
+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.protocol.UpdateNamenodeRegistrationResponse;
+import org.apache.hadoop.hdfs.server.federation.store.records.MembershipState;
+
+/**
+ * Management API for NameNode registrations stored in
+ * {@link org.apache.hadoop.hdfs.server.federation.store.records.MembershipState
+ * MembershipState} records. The {@link org.apache.hadoop.hdfs.server.
+ * federation.router.RouterHeartbeatService RouterHeartbeatService} periodically
+ * polls each NN to update the NameNode metadata(addresses, operational) and HA
+ * state(active, standby). Each NameNode may be polled by multiple
+ * {@link org.apache.hadoop.hdfs.server.federation.router.Router Router}
+ * instances.
+ * <p>
+ * Once fetched from the
+ * {@link org.apache.hadoop.hdfs.server.federation.store.driver.StateStoreDriver
+ * StateStoreDriver}, NameNode registrations are cached until the next query.
+ * The fetched registration data is aggregated using a quorum to determine the
+ * best/most accurate state for each NameNode. The cache is periodically updated
+ * by the @{link StateStoreCacheUpdateService}.
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Evolving
+public abstract class MembershipStore
+ extends CachedRecordStore<MembershipState> {
+
+ protected MembershipStore(StateStoreDriver driver) {
+ super(MembershipState.class, driver, true);
+ }
+
+ /**
+ * Inserts or updates a namenode membership entry into the table.
+ *
+ * @param request Fully populated NamenodeHeartbeatRequest request.
+ * @return True if successful, false otherwise.
+ * @throws StateStoreUnavailableException Throws exception if the data store
+ * is not initialized.
+ * @throws IOException if the data store could not be queried or the query is
+ * invalid.
+ */
+ public abstract NamenodeHeartbeatResponse namenodeHeartbeat(
+ NamenodeHeartbeatRequest request) throws IOException;
+
+ /**
+ * Queries for a single cached registration entry matching the given
+ * parameters. Possible keys are the names of data structure elements Possible
+ * values are matching SQL "LIKE" targets.
+ *
+ * @param request Fully populated GetNamenodeRegistrationsRequest request.
+ * @return Single matching FederationMembershipStateEntry or null if not found
+ * or more than one entry matches.
+ * @throws StateStoreUnavailableException Throws exception if the data store
+ * is not initialized.
+ * @throws IOException if the data store could not be queried or the query is
+ * invalid.
+ */
+ public abstract GetNamenodeRegistrationsResponse getNamenodeRegistrations(
+ GetNamenodeRegistrationsRequest request) throws IOException;
+
+ /**
+ * Get the expired registrations from the registration cache.
+ *
+ * @return Expired registrations or zero-length list if none are found.
+ * @throws StateStoreUnavailableException Throws exception if the data store
+ * is not initialized.
+ * @throws IOException if the data store could not be queried or the query is
+ * invalid.
+ */
+ public abstract GetNamenodeRegistrationsResponse
+ getExpiredNamenodeRegistrations(GetNamenodeRegistrationsRequest request)
+ throws IOException;
+
+ /**
+ * Retrieves a list of registered nameservices and their associated info.
+ *
+ * @param request
+ * @return Collection of information for each registered nameservice.
+ * @throws IOException if the data store could not be queried or the query is
+ * invalid.
+ */
+ public abstract GetNamespaceInfoResponse getNamespaceInfo(
+ GetNamespaceInfoRequest request) throws IOException;
+
+ /**
+ * Overrides a cached namenode state with an updated state.
+ *
+ * @param request Fully populated OverrideNamenodeRegistrationRequest request.
+ * @return OverrideNamenodeRegistrationResponse
+ * @throws StateStoreUnavailableException if the data store is not
+ * initialized.
+ * @throws IOException if the data store could not be queried or the query is
+ * invalid.
+ */
+ public abstract UpdateNamenodeRegistrationResponse updateNamenodeRegistration(
+ UpdateNamenodeRegistrationRequest request) throws IOException;
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/87700d45/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/MountTableStore.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/MountTableStore.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/MountTableStore.java
new file mode 100644
index 0000000..b439659
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/MountTableStore.java
@@ -0,0 +1,49 @@
+/**
+ * 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 org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.hdfs.server.federation.resolver.MountTableManager;
+import org.apache.hadoop.hdfs.server.federation.store.driver.StateStoreDriver;
+import org.apache.hadoop.hdfs.server.federation.store.records.MountTable;
+
+/**
+ * Management API for the HDFS mount table information stored in
+ * {@link org.apache.hadoop.hdfs.server.federation.store.records.MountTable
+ * MountTable} records. The mount table contains entries that map a particular
+ * global namespace path one or more HDFS nameservices (NN) + target path. It is
+ * possible to map mount locations for root folders, directories or individual
+ * files.
+ * <p>
+ * Once fetched from the
+ * {@link org.apache.hadoop.hdfs.server.federation.store.driver.StateStoreDriver
+ * StateStoreDriver}, MountTable records are cached in a tree for faster access.
+ * Each path in the global namespace is mapped to a nameserivce ID and local
+ * path upon request. The cache is periodically updated by the @{link
+ * StateStoreCacheUpdateService}.
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Evolving
+public abstract class MountTableStore extends CachedRecordStore<MountTable>
+ implements MountTableManager {
+
+ public MountTableStore(StateStoreDriver driver) {
+ super(MountTable.class, driver);
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/87700d45/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/RecordStore.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/RecordStore.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/RecordStore.java
new file mode 100644
index 0000000..53a8b82
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/RecordStore.java
@@ -0,0 +1,100 @@
+/**
+ * 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 java.lang.reflect.Constructor;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.hdfs.server.federation.store.driver.StateStoreDriver;
+import org.apache.hadoop.hdfs.server.federation.store.records.BaseRecord;
+
+/**
+ * Store records in the State Store. Subclasses provide interfaces to operate on
+ * those records.
+ *
+ * @param <R> Record to store by this interface.
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Evolving
+public abstract class RecordStore<R extends BaseRecord> {
+
+ private static final Log LOG = LogFactory.getLog(RecordStore.class);
+
+
+ /** Class of the record stored in this State Store. */
+ private final Class<R> recordClass;
+
+ /** State store driver backed by persistent storage. */
+ private final StateStoreDriver driver;
+
+
+ /**
+ * Create a new store for records.
+ *
+ * @param clazz Class of the record to store.
+ * @param stateStoreDriver Driver for the State Store.
+ */
+ protected RecordStore(Class<R> clazz, StateStoreDriver stateStoreDriver) {
+ this.recordClass = clazz;
+ this.driver = stateStoreDriver;
+ }
+
+ /**
+ * Report a required record to the data store. The data store uses this to
+ * create/maintain storage for the record.
+ *
+ * @return The class of the required record or null if no record is required
+ * for this interface.
+ */
+ public Class<R> getRecordClass() {
+ return this.recordClass;
+ }
+
+ /**
+ * Get the State Store driver.
+ *
+ * @return State Store driver.
+ */
+ protected StateStoreDriver getDriver() {
+ return this.driver;
+ }
+
+ /**
+ * Build a state store API implementation interface.
+ *
+ * @param clazz The specific interface implementation to create
+ * @param driver The {@link StateStoreDriver} implementation in use.
+ * @return An initialized instance of the specified state store API
+ * implementation.
+ */
+ public static <T extends RecordStore<?>> T newInstance(
+ final Class<T> clazz, final StateStoreDriver driver) {
+
+ try {
+ Constructor<T> constructor = clazz.getConstructor(StateStoreDriver.class);
+ T recordStore = constructor.newInstance(driver);
+ return recordStore;
+ } catch (Exception e) {
+ LOG.error("Cannot create new instance for " + clazz, e);
+ return null;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/87700d45/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/RouterStore.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/RouterStore.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/RouterStore.java
new file mode 100644
index 0000000..c6a0dad
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/RouterStore.java
@@ -0,0 +1,78 @@
+/**
+ * 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 java.io.IOException;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.hdfs.server.federation.store.driver.StateStoreDriver;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.GetRouterRegistrationRequest;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.GetRouterRegistrationResponse;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.GetRouterRegistrationsRequest;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.GetRouterRegistrationsResponse;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.RouterHeartbeatRequest;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.RouterHeartbeatResponse;
+import org.apache.hadoop.hdfs.server.federation.store.records.RouterState;
+
+/**
+ * Management API for
+ * {@link org.apache.hadoop.hdfs.server.federation.store.records.RouterState
+ * RouterState} records in the state store. Accesses the data store via the
+ * {@link org.apache.hadoop.hdfs.server.federation.store.driver.
+ * StateStoreDriver StateStoreDriver} interface. No data is cached.
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Evolving
+public abstract class RouterStore extends CachedRecordStore<RouterState> {
+
+ public RouterStore(StateStoreDriver driver) {
+ super(RouterState.class, driver, true);
+ }
+
+ /**
+ * Fetches the current router state object.
+ *
+ * @param request Fully populated request object.
+ * @return The matching router record or null if none exists.
+ * @throws IOException Throws exception if unable to query the data store or
+ * if more than one matching record is found.
+ */
+ public abstract GetRouterRegistrationResponse getRouterRegistration(
+ GetRouterRegistrationRequest request) throws IOException;
+
+ /**
+ * Fetches all router status objects.
+ *
+ * @param request Fully populated request object.
+ * @return List of Router records present in the data store.
+ * @throws IOException Throws exception if unable to query the data store
+ */
+ public abstract GetRouterRegistrationsResponse getRouterRegistrations(
+ GetRouterRegistrationsRequest request) throws IOException;
+
+ /**
+ * Update the state of this router in the State Store.
+ *
+ * @param request Fully populated request object.
+ * @return True if the update was successfully recorded, false otherwise.
+ * @throws IOException Throws exception if unable to query the data store
+ */
+ public abstract RouterHeartbeatResponse routerHeartbeat(
+ RouterHeartbeatRequest request) throws IOException;
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/87700d45/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/StateStoreCache.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/StateStoreCache.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/StateStoreCache.java
new file mode 100644
index 0000000..83fc501
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/StateStoreCache.java
@@ -0,0 +1,36 @@
+/**
+ * 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 java.io.IOException;
+
+/**
+ * Interface for a cached copy of the State Store.
+ */
+public interface StateStoreCache {
+
+ /**
+ * Load the cache from the State Store. Called by the cache update service
+ * when the data has been reloaded.
+ *
+ * @param force If we force the load.
+ * @return If the cache was loaded successfully.
+ * @throws IOException If there was an error loading the cache.
+ */
+ boolean loadCache(boolean force) throws IOException;
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/87700d45/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/StateStoreCacheUpdateService.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/StateStoreCacheUpdateService.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/StateStoreCacheUpdateService.java
new file mode 100644
index 0000000..6a4dc76
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/StateStoreCacheUpdateService.java
@@ -0,0 +1,70 @@
+/**
+ * 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 java.util.concurrent.TimeUnit;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdfs.server.federation.router.PeriodicService;
+import org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Service to periodically update the {@link StateStoreService}
+ * cached information in the
+ * {@link org.apache.hadoop.hdfs.server.federation.router.Router Router}.
+ * This is for performance and removes the State Store from the critical path
+ * in common operations.
+ */
+public class StateStoreCacheUpdateService extends PeriodicService {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(StateStoreCacheUpdateService.class);
+
+ /** The service that manages the State Store connection. */
+ private final StateStoreService stateStore;
+
+
+ /**
+ * Create a new Cache update service.
+ *
+ * @param stateStore Implementation of the state store
+ */
+ public StateStoreCacheUpdateService(StateStoreService stateStore) {
+ super(StateStoreCacheUpdateService.class.getSimpleName());
+ this.stateStore = stateStore;
+ }
+
+ @Override
+ protected void serviceInit(Configuration conf) throws Exception {
+
+ this.setIntervalMs(conf.getTimeDuration(
+ RBFConfigKeys.DFS_ROUTER_CACHE_TIME_TO_LIVE_MS,
+ RBFConfigKeys.DFS_ROUTER_CACHE_TIME_TO_LIVE_MS_DEFAULT,
+ TimeUnit.MILLISECONDS));
+
+ super.serviceInit(conf);
+ }
+
+ @Override
+ public void periodicInvoke() {
+ LOG.debug("Updating State Store cache");
+ stateStore.refreshCaches();
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/87700d45/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/StateStoreConnectionMonitorService.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/StateStoreConnectionMonitorService.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/StateStoreConnectionMonitorService.java
new file mode 100644
index 0000000..8ebf4b8
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/StateStoreConnectionMonitorService.java
@@ -0,0 +1,67 @@
+/**
+ * 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 org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdfs.server.federation.router.PeriodicService;
+import org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Service to periodically monitor the connection of the StateStore
+ * {@link StateStoreService} data store and to re-open the connection
+ * to the data store if required.
+ */
+public class StateStoreConnectionMonitorService extends PeriodicService {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(StateStoreConnectionMonitorService.class);
+
+ /** Service that maintains the State Store connection. */
+ private final StateStoreService stateStore;
+
+
+ /**
+ * Create a new service to monitor the connectivity of the state store driver.
+ *
+ * @param store Instance of the state store to be monitored.
+ */
+ public StateStoreConnectionMonitorService(StateStoreService store) {
+ super(StateStoreConnectionMonitorService.class.getSimpleName());
+ this.stateStore = store;
+ }
+
+ @Override
+ protected void serviceInit(Configuration conf) throws Exception {
+ this.setIntervalMs(conf.getLong(
+ RBFConfigKeys.FEDERATION_STORE_CONNECTION_TEST_MS,
+ RBFConfigKeys.FEDERATION_STORE_CONNECTION_TEST_MS_DEFAULT));
+
+ super.serviceInit(conf);
+ }
+
+ @Override
+ public void periodicInvoke() {
+ LOG.debug("Checking state store connection");
+ if (!stateStore.isDriverReady()) {
+ LOG.info("Attempting to open state store driver.");
+ stateStore.loadDriver();
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/87700d45/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/StateStoreService.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/StateStoreService.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/StateStoreService.java
new file mode 100644
index 0000000..ccbde09
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/StateStoreService.java
@@ -0,0 +1,450 @@
+/**
+ * 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 java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import javax.management.NotCompliantMBeanException;
+import javax.management.ObjectName;
+import javax.management.StandardMBean;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdfs.server.federation.metrics.StateStoreMBean;
+import org.apache.hadoop.hdfs.server.federation.metrics.StateStoreMetrics;
+import org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys;
+import org.apache.hadoop.hdfs.server.federation.store.driver.StateStoreDriver;
+import org.apache.hadoop.hdfs.server.federation.store.impl.MembershipStoreImpl;
+import org.apache.hadoop.hdfs.server.federation.store.impl.MountTableStoreImpl;
+import org.apache.hadoop.hdfs.server.federation.store.impl.RouterStoreImpl;
+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.RouterState;
+import org.apache.hadoop.metrics2.MetricsException;
+import org.apache.hadoop.metrics2.util.MBeans;
+import org.apache.hadoop.service.CompositeService;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.hadoop.util.Time;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.annotations.VisibleForTesting;
+
+/**
+ * A service to initialize a
+ * {@link org.apache.hadoop.hdfs.server.federation.store.driver.StateStoreDriver
+ * StateStoreDriver} and maintain the connection to the data store. There are
+ * multiple state store driver connections supported:
+ * <ul>
+ * <li>File
+ * {@link org.apache.hadoop.hdfs.server.federation.store.driver.impl.
+ * StateStoreFileImpl StateStoreFileImpl}
+ * <li>ZooKeeper
+ * {@link org.apache.hadoop.hdfs.server.federation.store.driver.impl.
+ * StateStoreZooKeeperImpl StateStoreZooKeeperImpl}
+ * </ul>
+ * <p>
+ * The service also supports the dynamic registration of record stores like:
+ * <ul>
+ * <li>{@link MembershipStore}: state of the Namenodes in the
+ * federation.
+ * <li>{@link MountTableStore}: Mount table between to subclusters.
+ * See {@link org.apache.hadoop.fs.viewfs.ViewFs ViewFs}.
+ * <li>{@link RebalancerStore}: Log of the rebalancing operations.
+ * <li>{@link RouterStore}: Router state in the federation.
+ * <li>{@link TokenStore}: Tokens in the federation.
+ * </ul>
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Evolving
+public class StateStoreService extends CompositeService {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(StateStoreService.class);
+
+
+ /** State Store configuration. */
+ private Configuration conf;
+
+ /** Identifier for the service. */
+ private String identifier;
+
+ /** Driver for the back end connection. */
+ private StateStoreDriver driver;
+
+ /** Service to maintain data store connection. */
+ private StateStoreConnectionMonitorService monitorService;
+
+ /** StateStore metrics. */
+ private StateStoreMetrics metrics;
+
+ /** Supported record stores. */
+ private final Map<
+ Class<? extends BaseRecord>, RecordStore<? extends BaseRecord>>
+ recordStores;
+
+ /** Service to maintain State Store caches. */
+ private StateStoreCacheUpdateService cacheUpdater;
+ /** Time the cache was last successfully updated. */
+ private long cacheLastUpdateTime;
+ /** List of internal caches to update. */
+ private final List<StateStoreCache> cachesToUpdateInternal;
+ /** List of external caches to update. */
+ private final List<StateStoreCache> cachesToUpdateExternal;
+
+
+ public StateStoreService() {
+ super(StateStoreService.class.getName());
+
+ // Records and stores supported by this implementation
+ this.recordStores = new HashMap<>();
+
+ // Caches to maintain
+ this.cachesToUpdateInternal = new ArrayList<>();
+ this.cachesToUpdateExternal = new ArrayList<>();
+ }
+
+ /**
+ * Initialize the State Store and the connection to the backend.
+ *
+ * @param config Configuration for the State Store.
+ * @throws IOException
+ */
+ @Override
+ protected void serviceInit(Configuration config) throws Exception {
+ this.conf = config;
+
+ // Create implementation of State Store
+ Class<? extends StateStoreDriver> driverClass = this.conf.getClass(
+ RBFConfigKeys.FEDERATION_STORE_DRIVER_CLASS,
+ RBFConfigKeys.FEDERATION_STORE_DRIVER_CLASS_DEFAULT,
+ StateStoreDriver.class);
+ this.driver = ReflectionUtils.newInstance(driverClass, this.conf);
+
+ if (this.driver == null) {
+ throw new IOException("Cannot create driver for the State Store");
+ }
+
+ // Add supported record stores
+ addRecordStore(MembershipStoreImpl.class);
+ addRecordStore(MountTableStoreImpl.class);
+ addRecordStore(RouterStoreImpl.class);
+
+ // Check the connection to the State Store periodically
+ this.monitorService = new StateStoreConnectionMonitorService(this);
+ this.addService(monitorService);
+
+ // Set expirations intervals for each record
+ MembershipState.setExpirationMs(conf.getLong(
+ RBFConfigKeys.FEDERATION_STORE_MEMBERSHIP_EXPIRATION_MS,
+ RBFConfigKeys.FEDERATION_STORE_MEMBERSHIP_EXPIRATION_MS_DEFAULT));
+
+ RouterState.setExpirationMs(conf.getTimeDuration(
+ RBFConfigKeys.FEDERATION_STORE_ROUTER_EXPIRATION_MS,
+ RBFConfigKeys.FEDERATION_STORE_ROUTER_EXPIRATION_MS_DEFAULT,
+ TimeUnit.MILLISECONDS));
+
+ // Cache update service
+ this.cacheUpdater = new StateStoreCacheUpdateService(this);
+ addService(this.cacheUpdater);
+
+ // Create metrics for the State Store
+ this.metrics = StateStoreMetrics.create(conf);
+
+ // Adding JMX interface
+ try {
+ StandardMBean bean = new StandardMBean(metrics, StateStoreMBean.class);
+ ObjectName registeredObject =
+ MBeans.register("Router", "StateStore", bean);
+ LOG.info("Registered StateStoreMBean: {}", registeredObject);
+ } catch (NotCompliantMBeanException e) {
+ throw new RuntimeException("Bad StateStoreMBean setup", e);
+ } catch (MetricsException e) {
+ LOG.info("Failed to register State Store bean {}", e.getMessage());
+ }
+
+ super.serviceInit(this.conf);
+ }
+
+ @Override
+ protected void serviceStart() throws Exception {
+ loadDriver();
+ super.serviceStart();
+ }
+
+ @Override
+ protected void serviceStop() throws Exception {
+ closeDriver();
+
+ if (metrics != null) {
+ metrics.shutdown();
+ metrics = null;
+ }
+
+ super.serviceStop();
+ }
+
+ /**
+ * Add a record store to the State Store. It includes adding the store, the
+ * supported record and the cache management.
+ *
+ * @param clazz Class of the record store to track.
+ * @return New record store.
+ * @throws ReflectiveOperationException
+ */
+ private <T extends RecordStore<?>> void addRecordStore(
+ final Class<T> clazz) throws ReflectiveOperationException {
+
+ assert this.getServiceState() == STATE.INITED :
+ "Cannot add record to the State Store once started";
+
+ T recordStore = RecordStore.newInstance(clazz, this.getDriver());
+ Class<? extends BaseRecord> recordClass = recordStore.getRecordClass();
+ this.recordStores.put(recordClass, recordStore);
+
+ // Subscribe for cache updates
+ if (recordStore instanceof StateStoreCache) {
+ StateStoreCache cachedRecordStore = (StateStoreCache) recordStore;
+ this.cachesToUpdateInternal.add(cachedRecordStore);
+ }
+ }
+
+ /**
+ * Get the record store in this State Store for a given interface.
+ *
+ * @param recordStoreClass Class of the record store.
+ * @return Registered record store or null if not found.
+ */
+ public <T extends RecordStore<?>> T getRegisteredRecordStore(
+ final Class<T> recordStoreClass) {
+ for (RecordStore<? extends BaseRecord> recordStore :
+ this.recordStores.values()) {
+ if (recordStoreClass.isInstance(recordStore)) {
+ @SuppressWarnings("unchecked")
+ T recordStoreChecked = (T) recordStore;
+ return recordStoreChecked;
+ }
+ }
+ return null;
+ }
+
+ /**
+ * List of records supported by this State Store.
+ *
+ * @return List of supported record classes.
+ */
+ public Collection<Class<? extends BaseRecord>> getSupportedRecords() {
+ return this.recordStores.keySet();
+ }
+
+ /**
+ * Load the State Store driver. If successful, refresh cached data tables.
+ */
+ public void loadDriver() {
+ synchronized (this.driver) {
+ if (!isDriverReady()) {
+ String driverName = this.driver.getClass().getSimpleName();
+ if (this.driver.init(
+ conf, getIdentifier(), getSupportedRecords(), metrics)) {
+ LOG.info("Connection to the State Store driver {} is open and ready",
+ driverName);
+ this.refreshCaches();
+ } else {
+ LOG.error("Cannot initialize State Store driver {}", driverName);
+ }
+ }
+ }
+ }
+
+ /**
+ * Check if the driver is ready to be used.
+ *
+ * @return If the driver is ready.
+ */
+ public boolean isDriverReady() {
+ return this.driver.isDriverReady();
+ }
+
+ /**
+ * Manually shuts down the driver.
+ *
+ * @throws Exception If the driver cannot be closed.
+ */
+ @VisibleForTesting
+ public void closeDriver() throws Exception {
+ if (this.driver != null) {
+ this.driver.close();
+ }
+ }
+
+ /**
+ * Get the state store driver.
+ *
+ * @return State store driver.
+ */
+ public StateStoreDriver getDriver() {
+ return this.driver;
+ }
+
+ /**
+ * Fetch a unique identifier for this state store instance. Typically it is
+ * the address of the router.
+ *
+ * @return Unique identifier for this store.
+ */
+ public String getIdentifier() {
+ return this.identifier;
+ }
+
+ /**
+ * Set a unique synchronization identifier for this store.
+ *
+ * @param id Unique identifier, typically the router's RPC address.
+ */
+ public void setIdentifier(String id) {
+ this.identifier = id;
+ }
+
+ //
+ // Cached state store data
+ //
+ /**
+ * The last time the state store cache was fully updated.
+ *
+ * @return Timestamp.
+ */
+ public long getCacheUpdateTime() {
+ return this.cacheLastUpdateTime;
+ }
+
+ /**
+ * Stops the cache update service.
+ */
+ @VisibleForTesting
+ public void stopCacheUpdateService() {
+ if (this.cacheUpdater != null) {
+ this.cacheUpdater.stop();
+ removeService(this.cacheUpdater);
+ this.cacheUpdater = null;
+ }
+ }
+
+ /**
+ * Register a cached record store for automatic periodic cache updates.
+ *
+ * @param client Client to the state store.
+ */
+ public void registerCacheExternal(StateStoreCache client) {
+ this.cachesToUpdateExternal.add(client);
+ }
+
+ /**
+ * Refresh the cache with information from the State Store. Called
+ * periodically by the CacheUpdateService to maintain data caches and
+ * versions.
+ */
+ public void refreshCaches() {
+ refreshCaches(false);
+ }
+
+ /**
+ * Refresh the cache with information from the State Store. Called
+ * periodically by the CacheUpdateService to maintain data caches and
+ * versions.
+ * @param force If we force the refresh.
+ */
+ public void refreshCaches(boolean force) {
+ boolean success = true;
+ if (isDriverReady()) {
+ List<StateStoreCache> cachesToUpdate = new LinkedList<>();
+ cachesToUpdate.addAll(cachesToUpdateInternal);
+ cachesToUpdate.addAll(cachesToUpdateExternal);
+ for (StateStoreCache cachedStore : cachesToUpdate) {
+ String cacheName = cachedStore.getClass().getSimpleName();
+ boolean result = false;
+ try {
+ result = cachedStore.loadCache(force);
+ } catch (IOException e) {
+ LOG.error("Error updating cache for {}", cacheName, e);
+ result = false;
+ }
+ if (!result) {
+ success = false;
+ LOG.error("Cache update failed for cache {}", cacheName);
+ }
+ }
+ } else {
+ success = false;
+ LOG.info("Skipping State Store cache update, driver is not ready.");
+ }
+ if (success) {
+ // Uses local time, not driver time.
+ this.cacheLastUpdateTime = Time.now();
+ }
+ }
+
+ /**
+ * Update the cache for a specific record store.
+ *
+ * @param clazz Class of the record store.
+ * @return If the cached was loaded.
+ * @throws IOException if the cache update failed.
+ */
+ public boolean loadCache(final Class<?> clazz) throws IOException {
+ return loadCache(clazz, false);
+ }
+
+ /**
+ * Update the cache for a specific record store.
+ *
+ * @param clazz Class of the record store.
+ * @param force Force the update ignoring cached periods.
+ * @return If the cached was loaded.
+ * @throws IOException if the cache update failed.
+ */
+ public boolean loadCache(Class<?> clazz, boolean force) throws IOException {
+ List<StateStoreCache> cachesToUpdate =
+ new LinkedList<StateStoreCache>();
+ cachesToUpdate.addAll(this.cachesToUpdateInternal);
+ cachesToUpdate.addAll(this.cachesToUpdateExternal);
+ for (StateStoreCache cachedStore : cachesToUpdate) {
+ if (clazz.isInstance(cachedStore)) {
+ return cachedStore.loadCache(force);
+ }
+ }
+ throw new IOException("Registered cache was not found for " + clazz);
+ }
+
+ /**
+ * Get the metrics for the State Store.
+ *
+ * @return State Store metrics.
+ */
+ public StateStoreMetrics getMetrics() {
+ return metrics;
+ }
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/87700d45/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/StateStoreUnavailableException.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/StateStoreUnavailableException.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/StateStoreUnavailableException.java
new file mode 100644
index 0000000..4e6f8c8
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/StateStoreUnavailableException.java
@@ -0,0 +1,33 @@
+/**
+ * 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 java.io.IOException;
+
+/**
+ * Thrown when the state store is not reachable or available. Cached APIs and
+ * queries may succeed. Client should retry again later.
+ */
+public class StateStoreUnavailableException extends IOException {
+
+ private static final long serialVersionUID = 1L;
+
+ public StateStoreUnavailableException(String msg) {
+ super(msg);
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/87700d45/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/StateStoreUtils.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/StateStoreUtils.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/StateStoreUtils.java
new file mode 100644
index 0000000..0a36619
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/StateStoreUtils.java
@@ -0,0 +1,109 @@
+/**
+ * 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 java.util.ArrayList;
+import java.util.List;
+
+import org.apache.hadoop.hdfs.server.federation.store.records.BaseRecord;
+import org.apache.hadoop.hdfs.server.federation.store.records.Query;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Set of utility functions used to work with the State Store.
+ */
+public final class StateStoreUtils {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(StateStoreUtils.class);
+
+
+ private StateStoreUtils() {
+ // Utility class
+ }
+
+ /**
+ * Get the base class for a record class. If we get an implementation of a
+ * record we will return the real parent record class.
+ *
+ * @param clazz Class of the data record to check.
+ * @return Base class for the record.
+ */
+ @SuppressWarnings("unchecked")
+ public static <T extends BaseRecord>
+ Class<? extends BaseRecord> getRecordClass(final Class<T> clazz) {
+
+ // We ignore the Impl classes and go to the super class
+ Class<? extends BaseRecord> actualClazz = clazz;
+ while (actualClazz.getSimpleName().endsWith("Impl")) {
+ actualClazz = (Class<? extends BaseRecord>) actualClazz.getSuperclass();
+ }
+
+ // Check if we went too far
+ if (actualClazz.equals(BaseRecord.class)) {
+ LOG.error("We went too far ({}) with {}", actualClazz, clazz);
+ actualClazz = clazz;
+ }
+ return actualClazz;
+ }
+
+ /**
+ * Get the base class for a record. If we get an implementation of a record we
+ * will return the real parent record class.
+ *
+ * @param record Record to check its main class.
+ * @return Base class for the record.
+ */
+ public static <T extends BaseRecord>
+ Class<? extends BaseRecord> getRecordClass(final T record) {
+ return getRecordClass(record.getClass());
+ }
+
+ /**
+ * Get the base class name for a record. If we get an implementation of a
+ * record we will return the real parent record class.
+ *
+ * @param clazz Class of the data record to check.
+ * @return Name of the base class for the record.
+ */
+ public static <T extends BaseRecord> String getRecordName(
+ final Class<T> clazz) {
+ return getRecordClass(clazz).getSimpleName();
+ }
+
+ /**
+ * Filters a list of records to find all records matching the query.
+ *
+ * @param query Map of field names and objects to use to filter results.
+ * @param records List of data records to filter.
+ * @return List of all records matching the query (or empty list if none
+ * match), null if the data set could not be filtered.
+ */
+ public static <T extends BaseRecord> List<T> filterMultiple(
+ final Query<T> query, final Iterable<T> records) {
+
+ List<T> matchingList = new ArrayList<>();
+ for (T record : records) {
+ if (query.matches(record)) {
+ matchingList.add(record);
+ }
+ }
+ return matchingList;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/87700d45/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/StateStoreDriver.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/StateStoreDriver.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/StateStoreDriver.java
new file mode 100644
index 0000000..d595a97
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/StateStoreDriver.java
@@ -0,0 +1,203 @@
+/**
+ * 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.net.InetAddress;
+import java.util.Collection;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdfs.server.federation.metrics.StateStoreMetrics;
+import org.apache.hadoop.hdfs.server.federation.store.StateStoreService;
+import org.apache.hadoop.hdfs.server.federation.store.StateStoreUnavailableException;
+import org.apache.hadoop.hdfs.server.federation.store.StateStoreUtils;
+import org.apache.hadoop.hdfs.server.federation.store.records.BaseRecord;
+import org.apache.hadoop.util.Time;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Driver class for an implementation of a {@link StateStoreService}
+ * provider. Driver implementations will extend this class and implement some of
+ * the default methods.
+ */
+public abstract class StateStoreDriver implements StateStoreRecordOperations {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(StateStoreDriver.class);
+
+
+ /** State Store configuration. */
+ private Configuration conf;
+
+ /** Identifier for the driver. */
+ private String identifier;
+
+ /** State Store metrics. */
+ private StateStoreMetrics metrics;
+
+
+ /**
+ * Initialize the state store connection.
+ *
+ * @param config Configuration for the driver.
+ * @param id Identifier for the driver.
+ * @param records Records that are supported.
+ * @return If initialized and ready, false if failed to initialize driver.
+ */
+ public boolean init(final Configuration config, final String id,
+ final Collection<Class<? extends BaseRecord>> records,
+ final StateStoreMetrics stateStoreMetrics) {
+
+ this.conf = config;
+ this.identifier = id;
+ this.metrics = stateStoreMetrics;
+
+ if (this.identifier == null) {
+ LOG.warn("The identifier for the State Store connection is not set");
+ }
+
+ boolean success = initDriver();
+ if (!success) {
+ LOG.error("Cannot initialize driver for {}", getDriverName());
+ return false;
+ }
+
+ for (Class<? extends BaseRecord> cls : records) {
+ String recordString = StateStoreUtils.getRecordName(cls);
+ if (!initRecordStorage(recordString, cls)) {
+ LOG.error("Cannot initialize record store for {}", cls.getSimpleName());
+ return false;
+ }
+ }
+ return true;
+ }
+
+ /**
+ * Get the State Store configuration.
+ *
+ * @return Configuration for the State Store.
+ */
+ protected Configuration getConf() {
+ return this.conf;
+ }
+
+ /**
+ * Gets a unique identifier for the running task/process. Typically the
+ * router address.
+ *
+ * @return Unique identifier for the running task.
+ */
+ public String getIdentifier() {
+ return this.identifier;
+ }
+
+ /**
+ * Get the metrics for the State Store.
+ *
+ * @return State Store metrics.
+ */
+ public StateStoreMetrics getMetrics() {
+ return this.metrics;
+ }
+
+ /**
+ * Prepare the driver to access data storage.
+ *
+ * @return True if the driver was successfully initialized. If false is
+ * returned, the state store will periodically attempt to
+ * re-initialize the driver and the router will remain in safe mode
+ * until the driver is initialized.
+ */
+ public abstract boolean initDriver();
+
+ /**
+ * Initialize storage for a single record class.
+ *
+ * @param className String reference of the record class to initialize,
+ * used to construct paths and file names for the record.
+ * Determined by configuration settings for the specific
+ * driver.
+ * @param clazz Record type corresponding to the provided name.
+ * @return True if successful, false otherwise.
+ */
+ public abstract <T extends BaseRecord> boolean initRecordStorage(
+ String className, Class<T> clazz);
+
+ /**
+ * Check if the driver is currently running and the data store connection is
+ * valid.
+ *
+ * @return True if the driver is initialized and the data store is ready.
+ */
+ public abstract boolean isDriverReady();
+
+ /**
+ * Check if the driver is ready to be used and throw an exception otherwise.
+ *
+ * @throws StateStoreUnavailableException If the driver is not ready.
+ */
+ public void verifyDriverReady() throws StateStoreUnavailableException {
+ if (!isDriverReady()) {
+ String driverName = getDriverName();
+ String hostname = getHostname();
+ throw new StateStoreUnavailableException("State Store driver " +
+ driverName + " in " + hostname + " is not ready.");
+ }
+ }
+
+ /**
+ * Close the State Store driver connection.
+ */
+ public abstract void close() throws Exception;
+
+ /**
+ * Returns the current time synchronization from the underlying store.
+ * Override for stores that supply a current date. The data store driver is
+ * responsible for maintaining the official synchronization time/date for all
+ * distributed components.
+ *
+ * @return Current time stamp, used for all synchronization dates.
+ */
+ public long getTime() {
+ return Time.now();
+ }
+
+ /**
+ * Get the name of the driver implementation for debugging.
+ *
+ * @return Name of the driver implementation.
+ */
+ private String getDriverName() {
+ return this.getClass().getSimpleName();
+ }
+
+ /**
+ * Get the host name of the machine running the driver for debugging.
+ *
+ * @return Host name of the machine running the driver.
+ */
+ private String getHostname() {
+ String hostname = "Unknown";
+ try {
+ hostname = InetAddress.getLocalHost().getHostName();
+ } catch (Exception e) {
+ LOG.error("Cannot get local address", e);
+ }
+ return hostname;
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/87700d45/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/StateStoreRecordOperations.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/StateStoreRecordOperations.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/StateStoreRecordOperations.java
new file mode 100644
index 0000000..443d46e
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/StateStoreRecordOperations.java
@@ -0,0 +1,148 @@
+/**
+ * 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 java.util.List;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.hdfs.server.federation.store.records.BaseRecord;
+import org.apache.hadoop.hdfs.server.federation.store.records.Query;
+import org.apache.hadoop.hdfs.server.federation.store.records.QueryResult;
+import org.apache.hadoop.io.retry.AtMostOnce;
+import org.apache.hadoop.io.retry.Idempotent;
+
+/**
+ * Operations for a driver to manage records in the State Store.
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Evolving
+public interface StateStoreRecordOperations {
+
+ /**
+ * Get all records of the requested record class from the data store. To use
+ * the default implementations in this class, getAll must return new instances
+ * of the records on each call. It is recommended to override the default
+ * implementations for better performance.
+ *
+ * @param clazz Class of record to fetch.
+ * @return List of all records that match the clazz.
+ * @throws IOException Throws exception if unable to query the data store.
+ */
+ @Idempotent
+ <T extends BaseRecord> QueryResult<T> get(Class<T> clazz) throws IOException;
+
+ /**
+ * Get a single record from the store that matches the query.
+ *
+ * @param clazz Class of record to fetch.
+ * @param query Query to filter results.
+ * @return A single record matching the query. Null if there are no matching
+ * records or more than one matching record in the store.
+ * @throws IOException If multiple records match or if the data store cannot
+ * be queried.
+ */
+ @Idempotent
+ <T extends BaseRecord> T get(Class<T> clazz, Query<T> query)
+ throws IOException;
+
+ /**
+ * Get multiple records from the store that match a query. This method
+ * assumes the underlying driver does not support filtering. If the driver
+ * supports filtering it should overwrite this method.
+ *
+ * @param clazz Class of record to fetch.
+ * @param query Query to filter results.
+ * @return Records of type clazz that match the query or empty list if none
+ * are found.
+ * @throws IOException Throws exception if unable to query the data store.
+ */
+ @Idempotent
+ <T extends BaseRecord> List<T> getMultiple(
+ Class<T> clazz, Query<T> query) throws IOException;
+
+ /**
+ * Creates a single record. Optionally updates an existing record with same
+ * primary key.
+ *
+ * @param record The record to insert or update.
+ * @param allowUpdate True if update of exiting record is allowed.
+ * @param errorIfExists True if an error should be returned when inserting
+ * an existing record. Only used if allowUpdate = false.
+ * @return True if the operation was successful.
+ *
+ * @throws IOException Throws exception if unable to query the data store.
+ */
+ @AtMostOnce
+ <T extends BaseRecord> boolean put(
+ T record, boolean allowUpdate, boolean errorIfExists) throws IOException;
+
+ /**
+ * Creates multiple records. Optionally updates existing records that have
+ * the same primary key.
+ *
+ * @param records List of data records to update or create. All records must
+ * be of class clazz.
+ * @param clazz Record class of records.
+ * @param allowUpdate True if update of exiting record is allowed.
+ * @param errorIfExists True if an error should be returned when inserting
+ * an existing record. Only used if allowUpdate = false.
+ * @return true if all operations were successful.
+ *
+ * @throws IOException Throws exception if unable to query the data store.
+ */
+ @AtMostOnce
+ <T extends BaseRecord> boolean putAll(
+ List<T> records, boolean allowUpdate, boolean errorIfExists)
+ throws IOException;
+
+ /**
+ * Remove a single record.
+ *
+ * @param record Record to be removed.
+ * @return true If the record was successfully removed. False if the record
+ * could not be removed or not stored.
+ * @throws IOException Throws exception if unable to query the data store.
+ */
+ @AtMostOnce
+ <T extends BaseRecord> boolean remove(T record) throws IOException;
+
+ /**
+ * Remove all records of this class from the store.
+ *
+ * @param clazz Class of records to remove.
+ * @return True if successful.
+ * @throws IOException Throws exception if unable to query the data store.
+ */
+ @AtMostOnce
+ <T extends BaseRecord> boolean removeAll(Class<T> clazz) throws IOException;
+
+ /**
+ * Remove multiple records of a specific class that match a query. Requires
+ * the getAll implementation to fetch fresh records on each call.
+ *
+ * @param query Query to filter what to remove.
+ * @return The number of records removed.
+ * @throws IOException Throws exception if unable to query the data store.
+ */
+ @AtMostOnce
+ <T extends BaseRecord> int remove(Class<T> clazz, Query<T> query)
+ throws IOException;
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/87700d45/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/StateStoreSerializer.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/StateStoreSerializer.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/StateStoreSerializer.java
new file mode 100644
index 0000000..666712f
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/StateStoreSerializer.java
@@ -0,0 +1,119 @@
+/**
+ * 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.server.federation.router.RBFConfigKeys;
+import org.apache.hadoop.hdfs.server.federation.store.records.BaseRecord;
+import org.apache.hadoop.util.ReflectionUtils;
+
+/**
+ * Serializer to store and retrieve data in the State Store.
+ */
+public abstract class StateStoreSerializer {
+
+ /** Singleton for the serializer instance. */
+ private static StateStoreSerializer defaultSerializer;
+
+ /**
+ * Get the default serializer based.
+ * @return Singleton serializer.
+ */
+ public static StateStoreSerializer getSerializer() {
+ return getSerializer(null);
+ }
+
+ /**
+ * Get a serializer based on the provided configuration.
+ * @param conf Configuration. Default if null.
+ * @return Singleton serializer.
+ */
+ public static StateStoreSerializer getSerializer(Configuration conf) {
+ if (conf == null) {
+ synchronized (StateStoreSerializer.class) {
+ if (defaultSerializer == null) {
+ conf = new Configuration();
+ defaultSerializer = newSerializer(conf);
+ }
+ }
+ return defaultSerializer;
+ } else {
+ return newSerializer(conf);
+ }
+ }
+
+ private static StateStoreSerializer newSerializer(final Configuration conf) {
+ Class<? extends StateStoreSerializer> serializerName = conf.getClass(
+ RBFConfigKeys.FEDERATION_STORE_SERIALIZER_CLASS,
+ RBFConfigKeys.FEDERATION_STORE_SERIALIZER_CLASS_DEFAULT,
+ StateStoreSerializer.class);
+ return ReflectionUtils.newInstance(serializerName, conf);
+ }
+
+ /**
+ * Create a new record.
+ * @param clazz Class of the new record.
+ * @return New record.
+ */
+ public static <T> T newRecord(Class<T> clazz) {
+ return getSerializer(null).newRecordInstance(clazz);
+ }
+
+ /**
+ * Create a new record.
+ * @param clazz Class of the new record.
+ * @return New record.
+ */
+ public abstract <T> T newRecordInstance(Class<T> clazz);
+
+ /**
+ * Serialize a record into a byte array.
+ * @param record Record to serialize.
+ * @return Byte array with the serialized record.
+ */
+ public abstract byte[] serialize(BaseRecord record);
+
+ /**
+ * Serialize a record into a string.
+ * @param record Record to serialize.
+ * @return String with the serialized record.
+ */
+ public abstract String serializeString(BaseRecord record);
+
+ /**
+ * Deserialize a bytes array into a record.
+ * @param byteArray Byte array to deserialize.
+ * @param clazz Class of the record.
+ * @return New record.
+ * @throws IOException If it cannot deserialize the record.
+ */
+ public abstract <T extends BaseRecord> T deserialize(
+ byte[] byteArray, Class<T> clazz) throws IOException;
+
+ /**
+ * Deserialize a string into a record.
+ * @param data String with the data to deserialize.
+ * @param clazz Class of the record.
+ * @return New record.
+ * @throws IOException If it cannot deserialize the record.
+ */
+ public abstract <T extends BaseRecord> T deserialize(
+ String data, Class<T> clazz) throws IOException;
+}
\ 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