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 we...@apache.org on 2018/03/20 06:26:28 UTC
[11/38] hadoop git commit: HDFS-13215. RBF: Move Router to its own
module.
http://git-wip-us.apache.org/repos/asf/hadoop/blob/4aa34324/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreFileBaseImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreFileBaseImpl.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreFileBaseImpl.java
deleted file mode 100644
index 6638d1c..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreFileBaseImpl.java
+++ /dev/null
@@ -1,462 +0,0 @@
-/**
- * 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.impl;
-
-import static org.apache.hadoop.hdfs.server.federation.store.StateStoreUtils.filterMultiple;
-import static org.apache.hadoop.util.Time.monotonicNow;
-import static org.apache.hadoop.util.Time.now;
-
-import java.io.BufferedReader;
-import java.io.BufferedWriter;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.concurrent.TimeUnit;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
-
-import org.apache.hadoop.hdfs.server.federation.metrics.StateStoreMetrics;
-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.driver.StateStoreDriver;
-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.util.Time;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.google.common.annotations.VisibleForTesting;
-
-/**
- * {@link StateStoreDriver} implementation based on files. In this approach, we
- * use temporary files for the writes and renaming "atomically" to the final
- * value. Instead of writing to the final location, it will go to a temporary
- * one and then rename to the final destination.
- */
-public abstract class StateStoreFileBaseImpl
- extends StateStoreSerializableImpl {
-
- private static final Logger LOG =
- LoggerFactory.getLogger(StateStoreFileBaseImpl.class);
-
- /** File extension for temporary files. */
- private static final String TMP_MARK = ".tmp";
- /** We remove temporary files older than 10 seconds. */
- private static final long OLD_TMP_RECORD_MS = TimeUnit.SECONDS.toMillis(10);
- /** File pattern for temporary records: file.XYZ.tmp. */
- private static final Pattern OLD_TMP_RECORD_PATTERN =
- Pattern.compile(".+\\.(\\d+)\\.tmp");
-
- /** If it is initialized. */
- private boolean initialized = false;
-
-
- /**
- * Get the reader of a record for the file system.
- *
- * @param path Path of the record to read.
- * @return Reader for the record.
- */
- protected abstract <T extends BaseRecord> BufferedReader getReader(
- String path);
-
- /**
- * Get the writer of a record for the file system.
- *
- * @param path Path of the record to write.
- * @return Writer for the record.
- */
- protected abstract <T extends BaseRecord> BufferedWriter getWriter(
- String path);
-
- /**
- * Check if a path exists.
- *
- * @param path Path to check.
- * @return If the path exists.
- */
- protected abstract boolean exists(String path);
-
- /**
- * Make a directory.
- *
- * @param path Path of the directory to create.
- * @return If the directory was created.
- */
- protected abstract boolean mkdir(String path);
-
- /**
- * Rename a file. This should be atomic.
- *
- * @param src Source name.
- * @param dst Destination name.
- * @return If the rename was successful.
- */
- protected abstract boolean rename(String src, String dst);
-
- /**
- * Remove a file.
- *
- * @param path Path for the file to remove
- * @return If the file was removed.
- */
- protected abstract boolean remove(String path);
-
- /**
- * Get the children for a path.
- *
- * @param path Path to check.
- * @return List of children.
- */
- protected abstract List<String> getChildren(String path);
-
- /**
- * Get root directory.
- *
- * @return Root directory.
- */
- protected abstract String getRootDir();
-
- /**
- * Set the driver as initialized.
- *
- * @param ini If the driver is initialized.
- */
- public void setInitialized(boolean ini) {
- this.initialized = ini;
- }
-
- @Override
- public boolean initDriver() {
- String rootDir = getRootDir();
- try {
- if (rootDir == null) {
- LOG.error("Invalid root directory, unable to initialize driver.");
- return false;
- }
-
- // Check root path
- if (!exists(rootDir)) {
- if (!mkdir(rootDir)) {
- LOG.error("Cannot create State Store root directory {}", rootDir);
- return false;
- }
- }
- } catch (Exception ex) {
- LOG.error(
- "Cannot initialize filesystem using root directory {}", rootDir, ex);
- return false;
- }
- setInitialized(true);
- return true;
- }
-
- @Override
- public <T extends BaseRecord> boolean initRecordStorage(
- String className, Class<T> recordClass) {
-
- String dataDirPath = getRootDir() + "/" + className;
- try {
- // Create data directories for files
- if (!exists(dataDirPath)) {
- LOG.info("{} data directory doesn't exist, creating it", dataDirPath);
- if (!mkdir(dataDirPath)) {
- LOG.error("Cannot create data directory {}", dataDirPath);
- return false;
- }
- }
- } catch (Exception ex) {
- LOG.error("Cannot create data directory {}", dataDirPath, ex);
- return false;
- }
- return true;
- }
-
- @Override
- public <T extends BaseRecord> QueryResult<T> get(Class<T> clazz)
- throws IOException {
- verifyDriverReady();
- long start = monotonicNow();
- StateStoreMetrics metrics = getMetrics();
- List<T> ret = new ArrayList<>();
- try {
- String path = getPathForClass(clazz);
- List<String> children = getChildren(path);
- for (String child : children) {
- String pathRecord = path + "/" + child;
- if (child.endsWith(TMP_MARK)) {
- LOG.debug("There is a temporary file {} in {}", child, path);
- if (isOldTempRecord(child)) {
- LOG.warn("Removing {} as it's an old temporary record", child);
- remove(pathRecord);
- }
- } else {
- T record = getRecord(pathRecord, clazz);
- ret.add(record);
- }
- }
- } catch (Exception e) {
- if (metrics != null) {
- metrics.addFailure(monotonicNow() - start);
- }
- String msg = "Cannot fetch records for " + clazz.getSimpleName();
- LOG.error(msg, e);
- throw new IOException(msg, e);
- }
-
- if (metrics != null) {
- metrics.addRead(monotonicNow() - start);
- }
- return new QueryResult<T>(ret, getTime());
- }
-
- /**
- * Check if a record is temporary and old.
- *
- * @param pathRecord Path for the record to check.
- * @return If the record is temporary and old.
- */
- @VisibleForTesting
- public static boolean isOldTempRecord(final String pathRecord) {
- if (!pathRecord.endsWith(TMP_MARK)) {
- return false;
- }
- // Extract temporary record creation time
- Matcher m = OLD_TMP_RECORD_PATTERN.matcher(pathRecord);
- if (m.find()) {
- long time = Long.parseLong(m.group(1));
- return now() - time > OLD_TMP_RECORD_MS;
- }
- return false;
- }
-
- /**
- * Read a record from a file.
- *
- * @param path Path to the file containing the record.
- * @param clazz Class of the record.
- * @return Record read from the file.
- * @throws IOException If the file cannot be read.
- */
- private <T extends BaseRecord> T getRecord(
- final String path, final Class<T> clazz) throws IOException {
- BufferedReader reader = getReader(path);
- try {
- String line;
- while ((line = reader.readLine()) != null) {
- if (!line.startsWith("#") && line.length() > 0) {
- try {
- T record = newRecord(line, clazz, false);
- return record;
- } catch (Exception ex) {
- LOG.error("Cannot parse line {} in file {}", line, path, ex);
- }
- }
- }
- } finally {
- if (reader != null) {
- reader.close();
- }
- }
- throw new IOException("Cannot read " + path + " for record " +
- clazz.getSimpleName());
- }
-
- /**
- * Get the path for a record class.
- * @param clazz Class of the record.
- * @return Path for this record class.
- */
- private <T extends BaseRecord> String getPathForClass(final Class<T> clazz) {
- String className = StateStoreUtils.getRecordName(clazz);
- StringBuilder sb = new StringBuilder();
- sb.append(getRootDir());
- if (sb.charAt(sb.length() - 1) != '/') {
- sb.append("/");
- }
- sb.append(className);
- return sb.toString();
- }
-
- @Override
- public boolean isDriverReady() {
- return this.initialized;
- }
-
- @Override
- public <T extends BaseRecord> boolean putAll(
- List<T> records, boolean allowUpdate, boolean errorIfExists)
- throws StateStoreUnavailableException {
- verifyDriverReady();
- if (records.isEmpty()) {
- return true;
- }
-
- long start = monotonicNow();
- StateStoreMetrics metrics = getMetrics();
-
- // Check if any record exists
- Map<String, T> toWrite = new HashMap<>();
- for (T record : records) {
- Class<? extends BaseRecord> recordClass = record.getClass();
- String path = getPathForClass(recordClass);
- String primaryKey = getPrimaryKey(record);
- String recordPath = path + "/" + primaryKey;
-
- if (exists(recordPath)) {
- if (allowUpdate) {
- // Update the mod time stamp. Many backends will use their
- // own timestamp for the mod time.
- record.setDateModified(this.getTime());
- toWrite.put(recordPath, record);
- } else if (errorIfExists) {
- LOG.error("Attempt to insert record {} that already exists",
- recordPath);
- if (metrics != null) {
- metrics.addFailure(monotonicNow() - start);
- }
- return false;
- } else {
- LOG.debug("Not updating {}", record);
- }
- } else {
- toWrite.put(recordPath, record);
- }
- }
-
- // Write the records
- boolean success = true;
- for (Entry<String, T> entry : toWrite.entrySet()) {
- String recordPath = entry.getKey();
- String recordPathTemp = recordPath + "." + now() + TMP_MARK;
- BufferedWriter writer = getWriter(recordPathTemp);
- try {
- T record = entry.getValue();
- String line = serializeString(record);
- writer.write(line);
- } catch (IOException e) {
- LOG.error("Cannot write {}", recordPathTemp, e);
- success = false;
- } finally {
- if (writer != null) {
- try {
- writer.close();
- } catch (IOException e) {
- LOG.error("Cannot close the writer for {}", recordPathTemp);
- }
- }
- }
- // Commit
- if (!rename(recordPathTemp, recordPath)) {
- LOG.error("Failed committing record into {}", recordPath);
- success = false;
- }
- }
-
- long end = monotonicNow();
- if (metrics != null) {
- if (success) {
- metrics.addWrite(end - start);
- } else {
- metrics.addFailure(end - start);
- }
- }
- return success;
- }
-
- @Override
- public <T extends BaseRecord> int remove(Class<T> clazz, Query<T> query)
- throws StateStoreUnavailableException {
- verifyDriverReady();
-
- if (query == null) {
- return 0;
- }
-
- long start = Time.monotonicNow();
- StateStoreMetrics metrics = getMetrics();
- int removed = 0;
- // Get the current records
- try {
- final QueryResult<T> result = get(clazz);
- final List<T> existingRecords = result.getRecords();
- // Write all of the existing records except those to be removed
- final List<T> recordsToRemove = filterMultiple(query, existingRecords);
- boolean success = true;
- for (T recordToRemove : recordsToRemove) {
- String path = getPathForClass(clazz);
- String primaryKey = getPrimaryKey(recordToRemove);
- String recordToRemovePath = path + "/" + primaryKey;
- if (remove(recordToRemovePath)) {
- removed++;
- } else {
- LOG.error("Cannot remove record {}", recordToRemovePath);
- success = false;
- }
- }
- if (!success) {
- LOG.error("Cannot remove records {} query {}", clazz, query);
- if (metrics != null) {
- metrics.addFailure(monotonicNow() - start);
- }
- }
- } catch (IOException e) {
- LOG.error("Cannot remove records {} query {}", clazz, query, e);
- if (metrics != null) {
- metrics.addFailure(monotonicNow() - start);
- }
- }
-
- if (removed > 0 && metrics != null) {
- metrics.addRemove(monotonicNow() - start);
- }
- return removed;
- }
-
- @Override
- public <T extends BaseRecord> boolean removeAll(Class<T> clazz)
- throws StateStoreUnavailableException {
- verifyDriverReady();
- long start = Time.monotonicNow();
- StateStoreMetrics metrics = getMetrics();
-
- boolean success = true;
- String path = getPathForClass(clazz);
- List<String> children = getChildren(path);
- for (String child : children) {
- String pathRecord = path + "/" + child;
- if (!remove(pathRecord)) {
- success = false;
- }
- }
-
- if (metrics != null) {
- long time = Time.monotonicNow() - start;
- if (success) {
- metrics.addRemove(time);
- } else {
- metrics.addFailure(time);
- }
- }
- return success;
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/4aa34324/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreFileImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreFileImpl.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreFileImpl.java
deleted file mode 100644
index c585a23..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreFileImpl.java
+++ /dev/null
@@ -1,148 +0,0 @@
-/**
- * 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.impl;
-
-import java.io.BufferedReader;
-import java.io.BufferedWriter;
-import java.io.File;
-import java.io.FileInputStream;
-import java.io.FileOutputStream;
-import java.io.IOException;
-import java.io.InputStreamReader;
-import java.io.OutputStreamWriter;
-import java.nio.charset.StandardCharsets;
-import java.util.LinkedList;
-import java.util.List;
-
-import org.apache.hadoop.hdfs.DFSConfigKeys;
-import org.apache.hadoop.hdfs.server.federation.store.records.BaseRecord;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.google.common.io.Files;
-
-/**
- * StateStoreDriver implementation based on a local file.
- */
-public class StateStoreFileImpl extends StateStoreFileBaseImpl {
-
- private static final Logger LOG =
- LoggerFactory.getLogger(StateStoreFileImpl.class);
-
- /** Configuration keys. */
- public static final String FEDERATION_STORE_FILE_DIRECTORY =
- DFSConfigKeys.FEDERATION_STORE_PREFIX + "driver.file.directory";
-
- /** Root directory for the state store. */
- private String rootDirectory;
-
-
- @Override
- protected boolean exists(String path) {
- File test = new File(path);
- return test.exists();
- }
-
- @Override
- protected boolean mkdir(String path) {
- File dir = new File(path);
- return dir.mkdirs();
- }
-
- @Override
- protected boolean rename(String src, String dst) {
- try {
- Files.move(new File(src), new File(dst));
- return true;
- } catch (IOException e) {
- LOG.error("Cannot rename {} to {}", src, dst, e);
- return false;
- }
- }
-
- @Override
- protected boolean remove(String path) {
- File file = new File(path);
- return file.delete();
- }
-
- @Override
- protected String getRootDir() {
- if (this.rootDirectory == null) {
- String dir = getConf().get(FEDERATION_STORE_FILE_DIRECTORY);
- if (dir == null) {
- File tempDir = Files.createTempDir();
- dir = tempDir.getAbsolutePath();
- LOG.warn("The root directory is not available, using {}", dir);
- }
- this.rootDirectory = dir;
- }
- return this.rootDirectory;
- }
-
- @Override
- protected <T extends BaseRecord> BufferedReader getReader(String filename) {
- BufferedReader reader = null;
- try {
- LOG.debug("Loading file: {}", filename);
- File file = new File(filename);
- FileInputStream fis = new FileInputStream(file);
- InputStreamReader isr =
- new InputStreamReader(fis, StandardCharsets.UTF_8);
- reader = new BufferedReader(isr);
- } catch (Exception ex) {
- LOG.error("Cannot open read stream for record {}", filename, ex);
- }
- return reader;
- }
-
- @Override
- protected <T extends BaseRecord> BufferedWriter getWriter(String filename) {
- BufferedWriter writer = null;
- try {
- LOG.debug("Writing file: {}", filename);
- File file = new File(filename);
- FileOutputStream fos = new FileOutputStream(file, false);
- OutputStreamWriter osw =
- new OutputStreamWriter(fos, StandardCharsets.UTF_8);
- writer = new BufferedWriter(osw);
- } catch (IOException e) {
- LOG.error("Cannot open write stream for record {}", filename, e);
- }
- return writer;
- }
-
- @Override
- public void close() throws Exception {
- setInitialized(false);
- }
-
- @Override
- protected List<String> getChildren(String path) {
- List<String> ret = new LinkedList<>();
- File dir = new File(path);
- File[] files = dir.listFiles();
- if (files != null) {
- for (File file : files) {
- String filename = file.getName();
- ret.add(filename);
- }
- }
- return ret;
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/4aa34324/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreFileSystemImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreFileSystemImpl.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreFileSystemImpl.java
deleted file mode 100644
index 8d6c626..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreFileSystemImpl.java
+++ /dev/null
@@ -1,178 +0,0 @@
-/**
- * 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.impl;
-
-import java.io.BufferedReader;
-import java.io.BufferedWriter;
-import java.io.IOException;
-import java.io.InputStreamReader;
-import java.io.OutputStreamWriter;
-import java.net.URI;
-import java.nio.charset.StandardCharsets;
-import java.util.LinkedList;
-import java.util.List;
-
-import org.apache.hadoop.fs.FSDataInputStream;
-import org.apache.hadoop.fs.FSDataOutputStream;
-import org.apache.hadoop.fs.FileStatus;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Options;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.hdfs.DFSConfigKeys;
-import org.apache.hadoop.hdfs.DistributedFileSystem;
-import org.apache.hadoop.hdfs.server.federation.store.records.BaseRecord;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * StateStoreDriver} implementation based on a filesystem. The most common uses
- * HDFS as a backend.
- */
-public class StateStoreFileSystemImpl extends StateStoreFileBaseImpl {
-
- private static final Logger LOG =
- LoggerFactory.getLogger(StateStoreFileSystemImpl.class);
-
-
- /** Configuration keys. */
- public static final String FEDERATION_STORE_FS_PATH =
- DFSConfigKeys.FEDERATION_STORE_PREFIX + "driver.fs.path";
-
- /** File system to back the State Store. */
- private FileSystem fs;
- /** Working path in the filesystem. */
- private String workPath;
-
- @Override
- protected boolean exists(String path) {
- try {
- return fs.exists(new Path(path));
- } catch (IOException e) {
- return false;
- }
- }
-
- @Override
- protected boolean mkdir(String path) {
- try {
- return fs.mkdirs(new Path(path));
- } catch (IOException e) {
- return false;
- }
- }
-
- @Override
- protected boolean rename(String src, String dst) {
- try {
- if (fs instanceof DistributedFileSystem) {
- DistributedFileSystem dfs = (DistributedFileSystem)fs;
- dfs.rename(new Path(src), new Path(dst), Options.Rename.OVERWRITE);
- return true;
- } else {
- // Replace should be atomic but not available
- if (fs.exists(new Path(dst))) {
- fs.delete(new Path(dst), true);
- }
- return fs.rename(new Path(src), new Path(dst));
- }
- } catch (Exception e) {
- LOG.error("Cannot rename {} to {}", src, dst, e);
- return false;
- }
- }
-
- @Override
- protected boolean remove(String path) {
- try {
- return fs.delete(new Path(path), true);
- } catch (Exception e) {
- LOG.error("Cannot remove {}", path, e);
- return false;
- }
- }
-
- @Override
- protected String getRootDir() {
- if (this.workPath == null) {
- String rootPath = getConf().get(FEDERATION_STORE_FS_PATH);
- URI workUri;
- try {
- workUri = new URI(rootPath);
- fs = FileSystem.get(workUri, getConf());
- } catch (Exception ex) {
- return null;
- }
- this.workPath = rootPath;
- }
- return this.workPath;
- }
-
- @Override
- public void close() throws Exception {
- if (fs != null) {
- fs.close();
- }
- }
-
- @Override
- protected <T extends BaseRecord> BufferedReader getReader(String pathName) {
- BufferedReader reader = null;
- Path path = new Path(pathName);
- try {
- FSDataInputStream fdis = fs.open(path);
- InputStreamReader isr =
- new InputStreamReader(fdis, StandardCharsets.UTF_8);
- reader = new BufferedReader(isr);
- } catch (IOException ex) {
- LOG.error("Cannot open read stream for {}", path);
- }
- return reader;
- }
-
- @Override
- protected <T extends BaseRecord> BufferedWriter getWriter(String pathName) {
- BufferedWriter writer = null;
- Path path = new Path(pathName);
- try {
- FSDataOutputStream fdos = fs.create(path, true);
- OutputStreamWriter osw =
- new OutputStreamWriter(fdos, StandardCharsets.UTF_8);
- writer = new BufferedWriter(osw);
- } catch (IOException ex) {
- LOG.error("Cannot open write stream for {}", path);
- }
- return writer;
- }
-
- @Override
- protected List<String> getChildren(String pathName) {
- List<String> ret = new LinkedList<>();
- Path path = new Path(workPath, pathName);
- try {
- FileStatus[] files = fs.listStatus(path);
- for (FileStatus file : files) {
- Path filePath = file.getPath();
- String fileName = filePath.getName();
- ret.add(fileName);
- }
- } catch (Exception e) {
- LOG.error("Cannot get children for {}", pathName, e);
- }
- return ret;
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/4aa34324/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreSerializableImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreSerializableImpl.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreSerializableImpl.java
deleted file mode 100644
index 7bc93de..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreSerializableImpl.java
+++ /dev/null
@@ -1,98 +0,0 @@
-/**
- * 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.impl;
-
-import java.io.IOException;
-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.driver.StateStoreSerializer;
-import org.apache.hadoop.hdfs.server.federation.store.records.BaseRecord;
-
-/**
- * State Store driver that stores a serialization of the records. The serializer
- * is pluggable.
- */
-public abstract class StateStoreSerializableImpl extends StateStoreBaseImpl {
-
- /** Mark for slashes in path names. */
- protected static final String SLASH_MARK = "0SLASH0";
- /** Mark for colon in path names. */
- protected static final String COLON_MARK = "_";
-
- /** Default serializer for this driver. */
- private StateStoreSerializer serializer;
-
-
- @Override
- public boolean init(final Configuration config, final String id,
- final Collection<Class<? extends BaseRecord>> records,
- final StateStoreMetrics metrics) {
- boolean ret = super.init(config, id, records, metrics);
-
- this.serializer = StateStoreSerializer.getSerializer(config);
-
- return ret;
- }
-
- /**
- * Serialize a record using the serializer.
- * @param record Record to serialize.
- * @return Byte array with the serialization of the record.
- */
- protected <T extends BaseRecord> byte[] serialize(T record) {
- return serializer.serialize(record);
- }
-
- /**
- * Serialize a record using the serializer.
- * @param record Record to serialize.
- * @return String with the serialization of the record.
- */
- protected <T extends BaseRecord> String serializeString(T record) {
- return serializer.serializeString(record);
- }
-
- /**
- * Creates a record from an input data string.
- * @param data Serialized text of the record.
- * @param clazz Record class.
- * @param includeDates If dateModified and dateCreated are serialized.
- * @return The created record.
- * @throws IOException
- */
- protected <T extends BaseRecord> T newRecord(
- String data, Class<T> clazz, boolean includeDates) throws IOException {
- return serializer.deserialize(data, clazz);
- }
-
- /**
- * Get the primary key for a record. If we don't want to store in folders, we
- * need to remove / from the name.
- *
- * @param record Record to get the primary key for.
- * @return Primary key for the record.
- */
- protected static String getPrimaryKey(BaseRecord record) {
- String primaryKey = record.getPrimaryKey();
- primaryKey = primaryKey.replaceAll("/", SLASH_MARK);
- primaryKey = primaryKey.replaceAll(":", COLON_MARK);
- return primaryKey;
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/4aa34324/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreSerializerPBImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreSerializerPBImpl.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreSerializerPBImpl.java
deleted file mode 100644
index 45c5dd6..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreSerializerPBImpl.java
+++ /dev/null
@@ -1,115 +0,0 @@
-/**
- * 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.impl;
-
-import java.io.IOException;
-
-import org.apache.commons.codec.binary.Base64;
-import org.apache.commons.codec.binary.StringUtils;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hdfs.server.federation.store.driver.StateStoreSerializer;
-import org.apache.hadoop.hdfs.server.federation.store.records.BaseRecord;
-import org.apache.hadoop.hdfs.server.federation.store.records.impl.pb.PBRecord;
-import org.apache.hadoop.util.ReflectionUtils;
-
-import com.google.protobuf.Message;
-
-/**
- * Protobuf implementation of the State Store serializer.
- */
-public final class StateStoreSerializerPBImpl extends StateStoreSerializer {
-
- private static final String PB_IMPL_PACKAGE_SUFFIX = "impl.pb";
- private static final String PB_IMPL_CLASS_SUFFIX = "PBImpl";
-
- private Configuration localConf = new Configuration();
-
-
- private StateStoreSerializerPBImpl() {
- }
-
- @Override
- @SuppressWarnings("unchecked")
- public <T> T newRecordInstance(Class<T> clazz) {
- try {
- String clazzPBImpl = getPBImplClassName(clazz);
- Class<?> pbClazz = localConf.getClassByName(clazzPBImpl);
- Object retObject = ReflectionUtils.newInstance(pbClazz, localConf);
- return (T)retObject;
- } catch (ClassNotFoundException e) {
- throw new RuntimeException(e);
- }
- }
-
- private String getPBImplClassName(Class<?> clazz) {
- String srcPackagePart = getPackageName(clazz);
- String srcClassName = getClassName(clazz);
- String destPackagePart = srcPackagePart + "." + PB_IMPL_PACKAGE_SUFFIX;
- String destClassPart = srcClassName + PB_IMPL_CLASS_SUFFIX;
- return destPackagePart + "." + destClassPart;
- }
-
- private String getClassName(Class<?> clazz) {
- String fqName = clazz.getName();
- return (fqName.substring(fqName.lastIndexOf(".") + 1, fqName.length()));
- }
-
- private String getPackageName(Class<?> clazz) {
- return clazz.getPackage().getName();
- }
-
- @Override
- public byte[] serialize(BaseRecord record) {
- byte[] byteArray64 = null;
- if (record instanceof PBRecord) {
- PBRecord recordPB = (PBRecord) record;
- Message msg = recordPB.getProto();
- byte[] byteArray = msg.toByteArray();
- byteArray64 = Base64.encodeBase64(byteArray, false);
- }
- return byteArray64;
- }
-
- @Override
- public String serializeString(BaseRecord record) {
- byte[] byteArray64 = serialize(record);
- String base64Encoded = StringUtils.newStringUtf8(byteArray64);
- return base64Encoded;
- }
-
- @Override
- public <T extends BaseRecord> T deserialize(
- byte[] byteArray, Class<T> clazz) throws IOException {
-
- T record = newRecord(clazz);
- if (record instanceof PBRecord) {
- PBRecord pbRecord = (PBRecord) record;
- byte[] byteArray64 = Base64.encodeBase64(byteArray, false);
- String base64Encoded = StringUtils.newStringUtf8(byteArray64);
- pbRecord.readInstance(base64Encoded);
- }
- return record;
- }
-
- @Override
- public <T extends BaseRecord> T deserialize(String data, Class<T> clazz)
- throws IOException {
- byte[] byteArray64 = Base64.decodeBase64(data);
- return deserialize(byteArray64, clazz);
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/4aa34324/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreZooKeeperImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreZooKeeperImpl.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreZooKeeperImpl.java
deleted file mode 100644
index 7f98171..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreZooKeeperImpl.java
+++ /dev/null
@@ -1,328 +0,0 @@
-/**
- * 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.impl;
-
-import static org.apache.hadoop.hdfs.server.federation.store.StateStoreUtils.filterMultiple;
-import static org.apache.hadoop.hdfs.server.federation.store.StateStoreUtils.getRecordName;
-import static org.apache.hadoop.util.curator.ZKCuratorManager.getNodePath;
-import static org.apache.hadoop.util.Time.monotonicNow;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
-
-import org.apache.curator.framework.CuratorFramework;
-import org.apache.curator.framework.imps.CuratorFrameworkState;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hdfs.DFSConfigKeys;
-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.Query;
-import org.apache.hadoop.hdfs.server.federation.store.records.QueryResult;
-import org.apache.hadoop.util.curator.ZKCuratorManager;
-import org.apache.zookeeper.data.Stat;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * {@link StateStoreDriver} driver implementation that uses ZooKeeper as a
- * backend.
- * <p>
- * The structure of the znodes in the ensemble is:
- * PARENT_PATH
- * |--- MOUNT
- * |--- MEMBERSHIP
- * |--- REBALANCER
- * |--- ROUTERS
- */
-public class StateStoreZooKeeperImpl extends StateStoreSerializableImpl {
-
- private static final Logger LOG =
- LoggerFactory.getLogger(StateStoreZooKeeperImpl.class);
-
-
- /** Configuration keys. */
- public static final String FEDERATION_STORE_ZK_DRIVER_PREFIX =
- DFSConfigKeys.FEDERATION_STORE_PREFIX + "driver.zk.";
- public static final String FEDERATION_STORE_ZK_PARENT_PATH =
- FEDERATION_STORE_ZK_DRIVER_PREFIX + "parent-path";
- public static final String FEDERATION_STORE_ZK_PARENT_PATH_DEFAULT =
- "/hdfs-federation";
-
-
- /** Directory to store the state store data. */
- private String baseZNode;
-
- /** Interface to ZooKeeper. */
- private ZKCuratorManager zkManager;
-
-
- @Override
- public boolean initDriver() {
- LOG.info("Initializing ZooKeeper connection");
-
- Configuration conf = getConf();
- baseZNode = conf.get(
- FEDERATION_STORE_ZK_PARENT_PATH,
- FEDERATION_STORE_ZK_PARENT_PATH_DEFAULT);
- try {
- this.zkManager = new ZKCuratorManager(conf);
- this.zkManager.start();
- } catch (IOException e) {
- LOG.error("Cannot initialize the ZK connection", e);
- return false;
- }
- return true;
- }
-
- @Override
- public <T extends BaseRecord> boolean initRecordStorage(
- String className, Class<T> clazz) {
- try {
- String checkPath = getNodePath(baseZNode, className);
- zkManager.createRootDirRecursively(checkPath);
- return true;
- } catch (Exception e) {
- LOG.error("Cannot initialize ZK node for {}: {}",
- className, e.getMessage());
- return false;
- }
- }
-
- @Override
- public void close() throws Exception {
- if (zkManager != null) {
- zkManager.close();
- }
- }
-
- @Override
- public boolean isDriverReady() {
- if (zkManager == null) {
- return false;
- }
- CuratorFramework curator = zkManager.getCurator();
- if (curator == null) {
- return false;
- }
- return curator.getState() == CuratorFrameworkState.STARTED;
- }
-
- @Override
- public <T extends BaseRecord> QueryResult<T> get(Class<T> clazz)
- throws IOException {
- verifyDriverReady();
- long start = monotonicNow();
- List<T> ret = new ArrayList<>();
- String znode = getZNodeForClass(clazz);
- try {
- List<String> children = zkManager.getChildren(znode);
- for (String child : children) {
- try {
- String path = getNodePath(znode, child);
- Stat stat = new Stat();
- String data = zkManager.getStringData(path, stat);
- boolean corrupted = false;
- if (data == null || data.equals("")) {
- // All records should have data, otherwise this is corrupted
- corrupted = true;
- } else {
- try {
- T record = createRecord(data, stat, clazz);
- ret.add(record);
- } catch (IOException e) {
- LOG.error("Cannot create record type \"{}\" from \"{}\": {}",
- clazz.getSimpleName(), data, e.getMessage());
- corrupted = true;
- }
- }
-
- if (corrupted) {
- LOG.error("Cannot get data for {} at {}, cleaning corrupted data",
- child, path);
- zkManager.delete(path);
- }
- } catch (Exception e) {
- LOG.error("Cannot get data for {}: {}", child, e.getMessage());
- }
- }
- } catch (Exception e) {
- getMetrics().addFailure(monotonicNow() - start);
- String msg = "Cannot get children for \"" + znode + "\": " +
- e.getMessage();
- LOG.error(msg);
- throw new IOException(msg);
- }
- long end = monotonicNow();
- getMetrics().addRead(end - start);
- return new QueryResult<T>(ret, getTime());
- }
-
- @Override
- public <T extends BaseRecord> boolean putAll(
- List<T> records, boolean update, boolean error) throws IOException {
- verifyDriverReady();
- if (records.isEmpty()) {
- return true;
- }
-
- // All records should be the same
- T record0 = records.get(0);
- Class<? extends BaseRecord> recordClass = record0.getClass();
- String znode = getZNodeForClass(recordClass);
-
- long start = monotonicNow();
- boolean status = true;
- for (T record : records) {
- String primaryKey = getPrimaryKey(record);
- String recordZNode = getNodePath(znode, primaryKey);
- byte[] data = serialize(record);
- if (!writeNode(recordZNode, data, update, error)){
- status = false;
- }
- }
- long end = monotonicNow();
- if (status) {
- getMetrics().addWrite(end - start);
- } else {
- getMetrics().addFailure(end - start);
- }
- return status;
- }
-
- @Override
- public <T extends BaseRecord> int remove(
- Class<T> clazz, Query<T> query) throws IOException {
- verifyDriverReady();
- if (query == null) {
- return 0;
- }
-
- // Read the current data
- long start = monotonicNow();
- List<T> records = null;
- try {
- QueryResult<T> result = get(clazz);
- records = result.getRecords();
- } catch (IOException ex) {
- LOG.error("Cannot get existing records", ex);
- getMetrics().addFailure(monotonicNow() - start);
- return 0;
- }
-
- // Check the records to remove
- String znode = getZNodeForClass(clazz);
- List<T> recordsToRemove = filterMultiple(query, records);
-
- // Remove the records
- int removed = 0;
- for (T existingRecord : recordsToRemove) {
- LOG.info("Removing \"{}\"", existingRecord);
- try {
- String primaryKey = getPrimaryKey(existingRecord);
- String path = getNodePath(znode, primaryKey);
- if (zkManager.delete(path)) {
- removed++;
- } else {
- LOG.error("Did not remove \"{}\"", existingRecord);
- }
- } catch (Exception e) {
- LOG.error("Cannot remove \"{}\"", existingRecord, e);
- getMetrics().addFailure(monotonicNow() - start);
- }
- }
- long end = monotonicNow();
- if (removed > 0) {
- getMetrics().addRemove(end - start);
- }
- return removed;
- }
-
- @Override
- public <T extends BaseRecord> boolean removeAll(Class<T> clazz)
- throws IOException {
- long start = monotonicNow();
- boolean status = true;
- String znode = getZNodeForClass(clazz);
- LOG.info("Deleting all children under {}", znode);
- try {
- List<String> children = zkManager.getChildren(znode);
- for (String child : children) {
- String path = getNodePath(znode, child);
- LOG.info("Deleting {}", path);
- zkManager.delete(path);
- }
- } catch (Exception e) {
- LOG.error("Cannot remove {}: {}", znode, e.getMessage());
- status = false;
- }
- long time = monotonicNow() - start;
- if (status) {
- getMetrics().addRemove(time);
- } else {
- getMetrics().addFailure(time);
- }
- return status;
- }
-
- private boolean writeNode(
- String znode, byte[] bytes, boolean update, boolean error) {
- try {
- boolean created = zkManager.create(znode);
- if (!update && !created && error) {
- LOG.info("Cannot write record \"{}\", it already exists", znode);
- return false;
- }
-
- // Write data
- zkManager.setData(znode, bytes, -1);
- return true;
- } catch (Exception e) {
- LOG.error("Cannot write record \"{}\": {}", znode, e.getMessage());
- }
- return false;
- }
-
- /**
- * Get the ZNode for a class.
- *
- * @param clazz Record class to evaluate.
- * @return The ZNode for the class.
- */
- private <T extends BaseRecord> String getZNodeForClass(Class<T> clazz) {
- String className = getRecordName(clazz);
- return getNodePath(baseZNode, className);
- }
-
- /**
- * Creates a record from a string returned by ZooKeeper.
- *
- * @param data The data to write.
- * @param stat Stat of the data record to create.
- * @param clazz The data record type to create.
- * @return The created record.
- * @throws IOException
- */
- private <T extends BaseRecord> T createRecord(
- String data, Stat stat, Class<T> clazz) throws IOException {
- T record = newRecord(data, clazz, false);
- record.setDateCreated(stat.getCtime());
- record.setDateModified(stat.getMtime());
- return record;
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/4aa34324/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/package-info.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/package-info.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/package-info.java
deleted file mode 100644
index a18433e..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/package-info.java
+++ /dev/null
@@ -1,39 +0,0 @@
-/**
- * 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.
- */
-
-/**
- * Implementations of state store data providers/drivers. Each driver is
- * responsible for maintaining, querying, updating and deleting persistent data
- * records. Data records are defined as subclasses of
- * {@link org.apache.hadoop.hdfs.server.federation.store.records.BaseRecord}.
- * Each driver implements the
- * {@link org.apache.hadoop.hdfs.server.federation.store.driver.StateStoreDriver
- * StateStoreDriver} interface.
- * <p>
- * Currently supported drivers:
- * <ul>
- * <li>{@link StateStoreFileImpl} A file-based data storage backend.
- * <li>{@link StateStoreZooKeeperImpl} A zookeeper based data storage backend.
- * </ul>
- */
-@InterfaceAudience.Private
-@InterfaceStability.Evolving
-package org.apache.hadoop.hdfs.server.federation.store.driver.impl;
-
-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/4aa34324/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/package-info.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/package-info.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/package-info.java
deleted file mode 100644
index da998b5..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/package-info.java
+++ /dev/null
@@ -1,37 +0,0 @@
-/**
- * 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 state store uses modular data storage classes derived from
- * StateStoreDriver to handle querying, updating and deleting data records. The
- * data storage driver is initialized and maintained by the StateStoreService.
- * The state store supports fetching all records of a type, filtering by column
- * values or fetching a single record by its primary key.
- * <p>
- * Each data storage backend is required to implement the methods contained in
- * the StateStoreDriver interface. These methods allow the querying, updating,
- * inserting and deleting of data records into the state store.
- */
-
-@InterfaceAudience.Private
-@InterfaceStability.Evolving
-
-package org.apache.hadoop.hdfs.server.federation.store.driver;
-
-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/4aa34324/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/impl/MembershipStoreImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/impl/MembershipStoreImpl.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/impl/MembershipStoreImpl.java
deleted file mode 100644
index 57b7b61..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/impl/MembershipStoreImpl.java
+++ /dev/null
@@ -1,310 +0,0 @@
-/**
- * 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.impl;
-
-import static org.apache.hadoop.hdfs.server.federation.store.StateStoreUtils.filterMultiple;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.TreeSet;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReadWriteLock;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
-
-import org.apache.hadoop.classification.InterfaceAudience;
-import org.apache.hadoop.classification.InterfaceStability;
-import org.apache.hadoop.hdfs.server.federation.resolver.FederationNamenodeServiceState;
-import org.apache.hadoop.hdfs.server.federation.resolver.FederationNamespaceInfo;
-import org.apache.hadoop.hdfs.server.federation.store.MembershipStore;
-import org.apache.hadoop.hdfs.server.federation.store.StateStoreCache;
-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;
-import org.apache.hadoop.hdfs.server.federation.store.records.Query;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Implementation of the {@link MembershipStore} State Store API.
- */
-@InterfaceAudience.Private
-@InterfaceStability.Evolving
-public class MembershipStoreImpl
- extends MembershipStore implements StateStoreCache {
-
- private static final Logger LOG =
- LoggerFactory.getLogger(MembershipStoreImpl.class);
-
-
- /** Reported namespaces that are not decommissioned. */
- private final Set<FederationNamespaceInfo> activeNamespaces;
-
- /** Namenodes (after evaluating the quorum) that are active in the cluster. */
- private final Map<String, MembershipState> activeRegistrations;
- /** Namenode status reports (raw) that were discarded for being too old. */
- private final Map<String, MembershipState> expiredRegistrations;
-
- /** Lock to access the local memory cache. */
- private final ReadWriteLock cacheReadWriteLock =
- new ReentrantReadWriteLock();
- private final Lock cacheReadLock = cacheReadWriteLock.readLock();
- private final Lock cacheWriteLock = cacheReadWriteLock.writeLock();
-
-
- public MembershipStoreImpl(StateStoreDriver driver) {
- super(driver);
-
- this.activeRegistrations = new HashMap<>();
- this.expiredRegistrations = new HashMap<>();
- this.activeNamespaces = new TreeSet<>();
- }
-
- @Override
- public GetNamenodeRegistrationsResponse getExpiredNamenodeRegistrations(
- GetNamenodeRegistrationsRequest request) throws IOException {
-
- GetNamenodeRegistrationsResponse response =
- GetNamenodeRegistrationsResponse.newInstance();
- cacheReadLock.lock();
- try {
- Collection<MembershipState> vals = this.expiredRegistrations.values();
- List<MembershipState> copyVals = new ArrayList<>(vals);
- response.setNamenodeMemberships(copyVals);
- } finally {
- cacheReadLock.unlock();
- }
- return response;
- }
-
- @Override
- public GetNamespaceInfoResponse getNamespaceInfo(
- GetNamespaceInfoRequest request) throws IOException {
-
- Set<FederationNamespaceInfo> namespaces = new HashSet<>();
- try {
- cacheReadLock.lock();
- namespaces.addAll(activeNamespaces);
- } finally {
- cacheReadLock.unlock();
- }
-
- GetNamespaceInfoResponse response =
- GetNamespaceInfoResponse.newInstance(namespaces);
- return response;
- }
-
- @Override
- public GetNamenodeRegistrationsResponse getNamenodeRegistrations(
- final GetNamenodeRegistrationsRequest request) throws IOException {
-
- // TODO Cache some common queries and sorts
- List<MembershipState> ret = null;
-
- cacheReadLock.lock();
- try {
- Collection<MembershipState> registrations = activeRegistrations.values();
- MembershipState partialMembership = request.getPartialMembership();
- if (partialMembership == null) {
- ret = new ArrayList<>(registrations);
- } else {
- Query<MembershipState> query = new Query<>(partialMembership);
- ret = filterMultiple(query, registrations);
- }
- } finally {
- cacheReadLock.unlock();
- }
- // Sort in ascending update date order
- Collections.sort(ret);
-
- GetNamenodeRegistrationsResponse response =
- GetNamenodeRegistrationsResponse.newInstance(ret);
- return response;
- }
-
- @Override
- public NamenodeHeartbeatResponse namenodeHeartbeat(
- NamenodeHeartbeatRequest request) throws IOException {
-
- MembershipState record = request.getNamenodeMembership();
- String nnId = record.getNamenodeKey();
- MembershipState existingEntry = null;
- cacheReadLock.lock();
- try {
- existingEntry = this.activeRegistrations.get(nnId);
- } finally {
- cacheReadLock.unlock();
- }
-
- if (existingEntry != null) {
- if (existingEntry.getState() != record.getState()) {
- LOG.info("NN registration state has changed: {} -> {}",
- existingEntry, record);
- } else {
- LOG.debug("Updating NN registration: {} -> {}", existingEntry, record);
- }
- } else {
- LOG.info("Inserting new NN registration: {}", record);
- }
-
- boolean status = getDriver().put(record, true, false);
-
- NamenodeHeartbeatResponse response =
- NamenodeHeartbeatResponse.newInstance(status);
- return response;
- }
-
- @Override
- public boolean loadCache(boolean force) throws IOException {
- super.loadCache(force);
-
- // Update local cache atomically
- cacheWriteLock.lock();
- try {
- this.activeRegistrations.clear();
- this.expiredRegistrations.clear();
- this.activeNamespaces.clear();
-
- // Build list of NN registrations: nnId -> registration list
- Map<String, List<MembershipState>> nnRegistrations = new HashMap<>();
- List<MembershipState> cachedRecords = getCachedRecords();
- for (MembershipState membership : cachedRecords) {
- String nnId = membership.getNamenodeKey();
- if (membership.getState() == FederationNamenodeServiceState.EXPIRED) {
- // Expired, RPC service does not use these
- String key = membership.getPrimaryKey();
- this.expiredRegistrations.put(key, membership);
- } else {
- // This is a valid NN registration, build a list of all registrations
- // using the NN id to use for the quorum calculation.
- List<MembershipState> nnRegistration =
- nnRegistrations.get(nnId);
- if (nnRegistration == null) {
- nnRegistration = new LinkedList<>();
- nnRegistrations.put(nnId, nnRegistration);
- }
- nnRegistration.add(membership);
- String bpId = membership.getBlockPoolId();
- String cId = membership.getClusterId();
- String nsId = membership.getNameserviceId();
- FederationNamespaceInfo nsInfo =
- new FederationNamespaceInfo(bpId, cId, nsId);
- this.activeNamespaces.add(nsInfo);
- }
- }
-
- // Calculate most representative entry for each active NN id
- for (List<MembershipState> nnRegistration : nnRegistrations.values()) {
- // Run quorum based on NN state
- MembershipState representativeRecord =
- getRepresentativeQuorum(nnRegistration);
- String nnKey = representativeRecord.getNamenodeKey();
- this.activeRegistrations.put(nnKey, representativeRecord);
- }
- LOG.debug("Refreshed {} NN registrations from State Store",
- cachedRecords.size());
- } finally {
- cacheWriteLock.unlock();
- }
- return true;
- }
-
- @Override
- public UpdateNamenodeRegistrationResponse updateNamenodeRegistration(
- UpdateNamenodeRegistrationRequest request) throws IOException {
-
- boolean status = false;
- cacheWriteLock.lock();
- try {
- String namenode = MembershipState.getNamenodeKey(
- request.getNameserviceId(), request.getNamenodeId());
- MembershipState member = this.activeRegistrations.get(namenode);
- if (member != null) {
- member.setState(request.getState());
- status = true;
- }
- } finally {
- cacheWriteLock.unlock();
- }
- UpdateNamenodeRegistrationResponse response =
- UpdateNamenodeRegistrationResponse.newInstance(status);
- return response;
- }
-
- /**
- * Picks the most recent entry in the subset that is most agreeable on the
- * specified field. 1) If a majority of the collection has the same value for
- * the field, the first sorted entry within the subset the matches the
- * majority value 2) Otherwise the first sorted entry in the set of all
- * entries
- *
- * @param records - Collection of state store record objects of the same type
- * @return record that is most representative of the field name
- */
- private MembershipState getRepresentativeQuorum(
- Collection<MembershipState> records) {
-
- // Collate objects by field value: field value -> order set of records
- Map<FederationNamenodeServiceState, TreeSet<MembershipState>> occurenceMap =
- new HashMap<>();
- for (MembershipState record : records) {
- FederationNamenodeServiceState state = record.getState();
- TreeSet<MembershipState> matchingSet = occurenceMap.get(state);
- if (matchingSet == null) {
- // TreeSet orders elements by descending date via comparators
- matchingSet = new TreeSet<>();
- occurenceMap.put(state, matchingSet);
- }
- matchingSet.add(record);
- }
-
- // Select largest group
- TreeSet<MembershipState> largestSet = new TreeSet<>();
- for (TreeSet<MembershipState> matchingSet : occurenceMap.values()) {
- if (largestSet.size() < matchingSet.size()) {
- largestSet = matchingSet;
- }
- }
-
- // If quorum, use the newest element here
- if (largestSet.size() > records.size() / 2) {
- return largestSet.first();
- // Otherwise, return most recent by class comparator
- } else if (records.size() > 0) {
- TreeSet<MembershipState> sortedList = new TreeSet<>(records);
- LOG.debug("Quorum failed, using most recent: {}", sortedList.first());
- return sortedList.first();
- } else {
- return null;
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/4aa34324/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/impl/MountTableStoreImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/impl/MountTableStoreImpl.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/impl/MountTableStoreImpl.java
deleted file mode 100644
index eb117d6..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/impl/MountTableStoreImpl.java
+++ /dev/null
@@ -1,154 +0,0 @@
-/**
- * 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.impl;
-
-import java.io.IOException;
-import java.util.Collections;
-import java.util.Iterator;
-import java.util.List;
-
-import org.apache.hadoop.classification.InterfaceAudience;
-import org.apache.hadoop.classification.InterfaceStability;
-import org.apache.hadoop.fs.permission.FsAction;
-import org.apache.hadoop.hdfs.server.federation.router.RouterAdminServer;
-import org.apache.hadoop.hdfs.server.federation.router.RouterPermissionChecker;
-import org.apache.hadoop.hdfs.server.federation.store.MountTableStore;
-import org.apache.hadoop.hdfs.server.federation.store.driver.StateStoreDriver;
-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.RemoveMountTableEntryResponse;
-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.Query;
-import org.apache.hadoop.security.AccessControlException;
-import org.apache.hadoop.util.Time;
-
-/**
- * Implementation of the {@link MountTableStore} state store API.
- */
-@InterfaceAudience.Private
-@InterfaceStability.Evolving
-public class MountTableStoreImpl extends MountTableStore {
-
- public MountTableStoreImpl(StateStoreDriver driver) {
- super(driver);
- }
-
- @Override
- public AddMountTableEntryResponse addMountTableEntry(
- AddMountTableEntryRequest request) throws IOException {
- MountTable mountTable = request.getEntry();
- if (mountTable != null) {
- RouterPermissionChecker pc = RouterAdminServer.getPermissionChecker();
- if (pc != null) {
- pc.checkPermission(mountTable, FsAction.WRITE);
- }
- }
-
- boolean status = getDriver().put(mountTable, false, true);
- AddMountTableEntryResponse response =
- AddMountTableEntryResponse.newInstance();
- response.setStatus(status);
- return response;
- }
-
- @Override
- public UpdateMountTableEntryResponse updateMountTableEntry(
- UpdateMountTableEntryRequest request) throws IOException {
- MountTable mountTable = request.getEntry();
- if (mountTable != null) {
- RouterPermissionChecker pc = RouterAdminServer.getPermissionChecker();
- if (pc != null) {
- pc.checkPermission(mountTable, FsAction.WRITE);
- }
- }
-
- boolean status = getDriver().put(mountTable, true, true);
- UpdateMountTableEntryResponse response =
- UpdateMountTableEntryResponse.newInstance();
- response.setStatus(status);
- return response;
- }
-
- @Override
- public RemoveMountTableEntryResponse removeMountTableEntry(
- RemoveMountTableEntryRequest request) throws IOException {
- final String srcPath = request.getSrcPath();
- final MountTable partial = MountTable.newInstance();
- partial.setSourcePath(srcPath);
- final Query<MountTable> query = new Query<>(partial);
- final MountTable deleteEntry = getDriver().get(getRecordClass(), query);
-
- boolean status = false;
- if (deleteEntry != null) {
- RouterPermissionChecker pc = RouterAdminServer.getPermissionChecker();
- if (pc != null) {
- pc.checkPermission(deleteEntry, FsAction.WRITE);
- }
- status = getDriver().remove(deleteEntry);
- }
-
- RemoveMountTableEntryResponse response =
- RemoveMountTableEntryResponse.newInstance();
- response.setStatus(status);
- return response;
- }
-
- @Override
- public GetMountTableEntriesResponse getMountTableEntries(
- GetMountTableEntriesRequest request) throws IOException {
- RouterPermissionChecker pc =
- RouterAdminServer.getPermissionChecker();
- // Get all values from the cache
- List<MountTable> records = getCachedRecords();
-
- // Sort and filter
- Collections.sort(records, MountTable.SOURCE_COMPARATOR);
- String reqSrcPath = request.getSrcPath();
- if (reqSrcPath != null && !reqSrcPath.isEmpty()) {
- // Return only entries beneath this path
- Iterator<MountTable> it = records.iterator();
- while (it.hasNext()) {
- MountTable record = it.next();
- String srcPath = record.getSourcePath();
- if (!srcPath.startsWith(reqSrcPath)) {
- it.remove();
- } else if (pc != null) {
- // do the READ permission check
- try {
- pc.checkPermission(record, FsAction.READ);
- } catch (AccessControlException ignored) {
- // Remove this mount table entry if it cannot
- // be accessed by current user.
- it.remove();
- }
- }
- }
- }
-
- GetMountTableEntriesResponse response =
- GetMountTableEntriesResponse.newInstance();
- response.setEntries(records);
- response.setTimestamp(Time.now());
- return response;
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/4aa34324/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/impl/RouterStoreImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/impl/RouterStoreImpl.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/impl/RouterStoreImpl.java
deleted file mode 100644
index d58c288..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/impl/RouterStoreImpl.java
+++ /dev/null
@@ -1,93 +0,0 @@
-/**
- * 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.impl;
-
-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.RouterStore;
-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.Query;
-import org.apache.hadoop.hdfs.server.federation.store.records.QueryResult;
-import org.apache.hadoop.hdfs.server.federation.store.records.RouterState;
-
-/**
- * Implementation of the {@link RouterStore} state store API.
- */
-@InterfaceAudience.Private
-@InterfaceStability.Evolving
-public class RouterStoreImpl extends RouterStore {
-
- public RouterStoreImpl(StateStoreDriver driver) {
- super(driver);
- }
-
- @Override
- public GetRouterRegistrationResponse getRouterRegistration(
- GetRouterRegistrationRequest request) throws IOException {
-
- final RouterState partial = RouterState.newInstance();
- partial.setAddress(request.getRouterId());
- final Query<RouterState> query = new Query<RouterState>(partial);
- RouterState record = getDriver().get(getRecordClass(), query);
- if (record != null) {
- overrideExpiredRecord(record);
- }
- GetRouterRegistrationResponse response =
- GetRouterRegistrationResponse.newInstance();
- response.setRouter(record);
- return response;
- }
-
- @Override
- public GetRouterRegistrationsResponse getRouterRegistrations(
- GetRouterRegistrationsRequest request) throws IOException {
-
- // Get all values from the cache
- QueryResult<RouterState> recordsAndTimeStamp =
- getCachedRecordsAndTimeStamp();
- List<RouterState> records = recordsAndTimeStamp.getRecords();
- long timestamp = recordsAndTimeStamp.getTimestamp();
-
- // Generate response
- GetRouterRegistrationsResponse response =
- GetRouterRegistrationsResponse.newInstance();
- response.setRouters(records);
- response.setTimestamp(timestamp);
- return response;
- }
-
- @Override
- public RouterHeartbeatResponse routerHeartbeat(RouterHeartbeatRequest request)
- throws IOException {
-
- RouterState record = request.getRouter();
- boolean status = getDriver().put(record, true, false);
- RouterHeartbeatResponse response =
- RouterHeartbeatResponse.newInstance(status);
- return response;
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/4aa34324/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/impl/package-info.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/impl/package-info.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/impl/package-info.java
deleted file mode 100644
index 1a50d15..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/impl/package-info.java
+++ /dev/null
@@ -1,31 +0,0 @@
-/**
- * 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.
- */
-
-/**
- * Contains implementations of the state store API interfaces. All classes
- * derive from {@link
- * org.apache.hadoop.hdfs.server.federation.store.RecordStore}. The API
- * definitions are contained in the
- * org.apache.hadoop.hdfs.server.federation.store package.
- */
-@InterfaceAudience.Private
-@InterfaceStability.Evolving
-package org.apache.hadoop.hdfs.server.federation.store.impl;
-
-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/4aa34324/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/package-info.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/package-info.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/package-info.java
deleted file mode 100644
index 949ec7c..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/package-info.java
+++ /dev/null
@@ -1,62 +0,0 @@
-/**
- * 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 federation state store tracks persistent values that are shared between
- * multiple routers.
- * <p>
- * Data is stored in data records that inherit from a common class. Data records
- * are serialized when written to the data store using a modular serialization
- * implementation. The default is profobuf serialization. Data is stored as rows
- * of records of the same type with each data member in a record representing a
- * column.
- * <p>
- * The state store uses a modular data storage
- * {@link org.apache.hadoop.hdfs.server.federation.store.driver.StateStoreDriver
- * StateStoreDriver} to handle querying, updating and deleting data records. The
- * data storage driver is initialized and maintained by the
- * {@link org.apache.hadoop.hdfs.server.federation.store.
- * StateStoreService FederationStateStoreService}. The state store
- * supports fetching all records of a type, filtering by column values or
- * fetching a single record by its primary key.
- * <p>
- * The state store contains several API interfaces, one for each data records
- * type.
- * <p>
- * <ul>
- * <li>FederationMembershipStateStore: state of all Namenodes in the federation.
- * Uses the MembershipState record.
- * <li>FederationMountTableStore: Mount table mapping paths in the global
- * namespace to individual subcluster paths. Uses the MountTable record.
- * <li>RouterStateStore: State of all routers in the federation. Uses the
- * RouterState record.
- * </ul>
- * <p>
- * Each API is defined in a separate interface. The implementations of these
- * interfaces are responsible for accessing the
- * {@link org.apache.hadoop.hdfs.server.federation.store.driver.StateStoreDriver
- * StateStoreDriver} to query, update and delete data records.
- */
-
-@InterfaceAudience.Private
-@InterfaceStability.Evolving
-
-package org.apache.hadoop.hdfs.server.federation.store;
-
-import org.apache.hadoop.classification.InterfaceAudience;
-import org.apache.hadoop.classification.InterfaceStability;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/4aa34324/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/protocol/AddMountTableEntryRequest.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/protocol/AddMountTableEntryRequest.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/protocol/AddMountTableEntryRequest.java
deleted file mode 100644
index 2d9f102..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/protocol/AddMountTableEntryRequest.java
+++ /dev/null
@@ -1,47 +0,0 @@
-/**
- * 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.protocol;
-
-import org.apache.hadoop.classification.InterfaceAudience.Public;
-import org.apache.hadoop.classification.InterfaceStability.Unstable;
-import org.apache.hadoop.hdfs.server.federation.store.driver.StateStoreSerializer;
-import org.apache.hadoop.hdfs.server.federation.store.records.MountTable;
-
-/**
- * API request for adding a mount table entry to the state store.
- */
-public abstract class AddMountTableEntryRequest {
-
- public static AddMountTableEntryRequest newInstance() {
- return StateStoreSerializer.newRecord(AddMountTableEntryRequest.class);
- }
-
- public static AddMountTableEntryRequest newInstance(MountTable newEntry) {
- AddMountTableEntryRequest request = newInstance();
- request.setEntry(newEntry);
- return request;
- }
-
- @Public
- @Unstable
- public abstract MountTable getEntry();
-
- @Public
- @Unstable
- public abstract void setEntry(MountTable mount);
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/4aa34324/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/protocol/AddMountTableEntryResponse.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/protocol/AddMountTableEntryResponse.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/protocol/AddMountTableEntryResponse.java
deleted file mode 100644
index 9bc7f92..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/protocol/AddMountTableEntryResponse.java
+++ /dev/null
@@ -1,42 +0,0 @@
-/**
- * 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.protocol;
-
-import java.io.IOException;
-
-import org.apache.hadoop.classification.InterfaceAudience.Public;
-import org.apache.hadoop.classification.InterfaceStability.Unstable;
-import org.apache.hadoop.hdfs.server.federation.store.driver.StateStoreSerializer;
-
-/**
- * API response for adding a mount table entry to the state store.
- */
-public abstract class AddMountTableEntryResponse {
-
- public static AddMountTableEntryResponse newInstance() throws IOException {
- return StateStoreSerializer.newRecord(AddMountTableEntryResponse.class);
- }
-
- @Public
- @Unstable
- public abstract boolean getStatus();
-
- @Public
- @Unstable
- public abstract void setStatus(boolean result);
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/4aa34324/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/protocol/EnterSafeModeRequest.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/protocol/EnterSafeModeRequest.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/protocol/EnterSafeModeRequest.java
deleted file mode 100644
index 89df1bb..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/store/protocol/EnterSafeModeRequest.java
+++ /dev/null
@@ -1,32 +0,0 @@
-/**
- * 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.protocol;
-
-import java.io.IOException;
-
-import org.apache.hadoop.hdfs.server.federation.store.driver.StateStoreSerializer;
-
-/**
- * API request for the Router entering safe mode state and updating
- * its state in State Store.
- */
-public abstract class EnterSafeModeRequest {
- public static EnterSafeModeRequest newInstance() throws IOException {
- return StateStoreSerializer.newRecord(EnterSafeModeRequest.class);
- }
-}
\ 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