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 aw...@apache.org on 2016/08/27 00:17:32 UTC
[1/8] hadoop git commit: HDFS-10768. Optimize mkdir ops. Contributed
by Daryn Sharp. [Forced Update!]
Repository: hadoop
Updated Branches:
refs/heads/HADOOP-13341 e3cc43ac4 -> cec31024b (forced update)
HDFS-10768. Optimize mkdir ops. Contributed by Daryn Sharp.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/8b7adf4d
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/8b7adf4d
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/8b7adf4d
Branch: refs/heads/HADOOP-13341
Commit: 8b7adf4ddf420a93c586c4b2eac27dd0f649682e
Parents: cde3a00
Author: Kihwal Lee <ki...@apache.org>
Authored: Fri Aug 26 15:39:21 2016 -0500
Committer: Kihwal Lee <ki...@apache.org>
Committed: Fri Aug 26 15:39:56 2016 -0500
----------------------------------------------------------------------
.../hdfs/server/namenode/FSDirMkdirOp.java | 91 +++++++++-----------
.../hdfs/server/namenode/FSDirSymlinkOp.java | 15 ++--
.../hdfs/server/namenode/FSDirWriteFileOp.java | 16 ++--
.../hdfs/server/namenode/INodesInPath.java | 35 ++------
.../server/namenode/TestSnapshotPathINodes.java | 5 ++
5 files changed, 68 insertions(+), 94 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8b7adf4d/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirMkdirOp.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirMkdirOp.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirMkdirOp.java
index 8aac1f8..bf5ff00 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirMkdirOp.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirMkdirOp.java
@@ -32,10 +32,7 @@ import org.apache.hadoop.hdfs.protocol.QuotaExceededException;
import org.apache.hadoop.hdfs.server.namenode.snapshot.Snapshot;
import java.io.IOException;
-import java.util.AbstractMap;
import java.util.List;
-import java.util.Map;
-
import static org.apache.hadoop.util.Time.now;
class FSDirMkdirOp {
@@ -63,7 +60,6 @@ class FSDirMkdirOp {
throw new FileAlreadyExistsException("Path is not a directory: " + src);
}
- INodesInPath existing = lastINode != null ? iip : iip.getExistingINodes();
if (lastINode == null) {
if (fsd.isPermissionEnabled()) {
fsd.checkAncestorAccess(pc, iip, FsAction.WRITE);
@@ -78,26 +74,20 @@ class FSDirMkdirOp {
// create multiple inodes.
fsn.checkFsObjectLimit();
- List<String> nonExisting = iip.getPath(existing.length(),
- iip.length() - existing.length());
- int length = nonExisting.size();
- if (length > 1) {
- List<String> ancestors = nonExisting.subList(0, length - 1);
- // Ensure that the user can traversal the path by adding implicit
- // u+wx permission to all ancestor directories
- existing = createChildrenDirectories(fsd, existing, ancestors,
- addImplicitUwx(permissions, permissions));
- if (existing == null) {
- throw new IOException("Failed to create directory: " + src);
- }
+ // Ensure that the user can traversal the path by adding implicit
+ // u+wx permission to all ancestor directories.
+ INodesInPath existing =
+ createParentDirectories(fsd, iip, permissions, false);
+ if (existing != null) {
+ existing = createSingleDirectory(
+ fsd, existing, iip.getLastLocalName(), permissions);
}
-
- if ((existing = createChildrenDirectories(fsd, existing,
- nonExisting.subList(length - 1, length), permissions)) == null) {
+ if (existing == null) {
throw new IOException("Failed to create directory: " + src);
}
+ iip = existing;
}
- return fsd.getAuditFileInfo(existing);
+ return fsd.getAuditFileInfo(iip);
} finally {
fsd.writeUnlock();
}
@@ -112,35 +102,18 @@ class FSDirMkdirOp {
* For example, path="/foo/bar/spam", "/foo" is an existing directory,
* "/foo/bar" is not existing yet, the function will create directory bar.
*
- * @return a tuple which contains both the new INodesInPath (with all the
- * existing and newly created directories) and the last component in the
- * relative path. Or return null if there are errors.
+ * @return a INodesInPath with all the existing and newly created
+ * ancestor directories created.
+ * Or return null if there are errors.
*/
- static Map.Entry<INodesInPath, String> createAncestorDirectories(
+ static INodesInPath createAncestorDirectories(
FSDirectory fsd, INodesInPath iip, PermissionStatus permission)
throws IOException {
- final String last = DFSUtil.bytes2String(iip.getLastLocalName());
- INodesInPath existing = iip.getExistingINodes();
- List<String> children = iip.getPath(existing.length(),
- iip.length() - existing.length());
- int size = children.size();
- if (size > 1) { // otherwise all ancestors have been created
- List<String> directories = children.subList(0, size - 1);
- INode parentINode = existing.getLastINode();
- // Ensure that the user can traversal the path by adding implicit
- // u+wx permission to all ancestor directories
- existing = createChildrenDirectories(fsd, existing, directories,
- addImplicitUwx(parentINode.getPermissionStatus(), permission));
- if (existing == null) {
- return null;
- }
- }
- return new AbstractMap.SimpleImmutableEntry<>(existing, last);
+ return createParentDirectories(fsd, iip, permission, true);
}
/**
- * Create the directory {@code parent} / {@code children} and all ancestors
- * along the path.
+ * Create all ancestor directories and return the parent inodes.
*
* @param fsd FSDirectory
* @param existing The INodesInPath instance containing all the existing
@@ -149,21 +122,35 @@ class FSDirMkdirOp {
* starting with "/"
* @param perm the permission of the directory. Note that all ancestors
* created along the path has implicit {@code u+wx} permissions.
+ * @param inheritPerms if the ancestor directories should inherit permissions
+ * or use the specified permissions.
*
* @return {@link INodesInPath} which contains all inodes to the
* target directory, After the execution parentPath points to the path of
* the returned INodesInPath. The function return null if the operation has
* failed.
*/
- private static INodesInPath createChildrenDirectories(FSDirectory fsd,
- INodesInPath existing, List<String> children, PermissionStatus perm)
+ private static INodesInPath createParentDirectories(FSDirectory fsd,
+ INodesInPath iip, PermissionStatus perm, boolean inheritPerms)
throws IOException {
assert fsd.hasWriteLock();
-
- for (String component : children) {
- existing = createSingleDirectory(fsd, existing, component, perm);
- if (existing == null) {
- return null;
+ // this is the desired parent iip if the subsequent delta is 1.
+ INodesInPath existing = iip.getExistingINodes();
+ int missing = iip.length() - existing.length();
+ if (missing == 0) { // full path exists, return parents.
+ existing = iip.getParentINodesInPath();
+ } else if (missing > 1) { // need to create at least one ancestor dir.
+ // Ensure that the user can traversal the path by adding implicit
+ // u+wx permission to all ancestor directories.
+ PermissionStatus basePerm = inheritPerms
+ ? existing.getLastINode().getPermissionStatus()
+ : perm;
+ perm = addImplicitUwx(basePerm, perm);
+ // create all the missing directories.
+ final int last = iip.length() - 2;
+ for (int i = existing.length(); existing != null && i <= last; i++) {
+ byte[] component = iip.getPathComponent(i);
+ existing = createSingleDirectory(fsd, existing, component, perm);
}
}
return existing;
@@ -183,11 +170,11 @@ class FSDirMkdirOp {
}
private static INodesInPath createSingleDirectory(FSDirectory fsd,
- INodesInPath existing, String localName, PermissionStatus perm)
+ INodesInPath existing, byte[] localName, PermissionStatus perm)
throws IOException {
assert fsd.hasWriteLock();
existing = unprotectedMkdir(fsd, fsd.allocateNewInodeId(), existing,
- DFSUtil.string2Bytes(localName), perm, null, now());
+ localName, perm, null, now());
if (existing == null) {
return null;
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8b7adf4d/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirSymlinkOp.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirSymlinkOp.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirSymlinkOp.java
index 4d32993..ec93952 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirSymlinkOp.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirSymlinkOp.java
@@ -27,7 +27,6 @@ import org.apache.hadoop.hdfs.protocol.HdfsFileStatus;
import org.apache.hadoop.hdfs.protocol.QuotaExceededException;
import java.io.IOException;
-import java.util.Map;
import static org.apache.hadoop.util.Time.now;
@@ -99,21 +98,21 @@ class FSDirSymlinkOp {
INodesInPath iip, String target, PermissionStatus dirPerms,
boolean createParent, boolean logRetryCache) throws IOException {
final long mtime = now();
- final byte[] localName = iip.getLastLocalName();
+ final INodesInPath parent;
if (createParent) {
- Map.Entry<INodesInPath, String> e = FSDirMkdirOp
- .createAncestorDirectories(fsd, iip, dirPerms);
- if (e == null) {
+ parent = FSDirMkdirOp.createAncestorDirectories(fsd, iip, dirPerms);
+ if (parent == null) {
return null;
}
- iip = INodesInPath.append(e.getKey(), null, localName);
+ } else {
+ parent = iip.getParentINodesInPath();
}
final String userName = dirPerms.getUserName();
long id = fsd.allocateNewInodeId();
PermissionStatus perm = new PermissionStatus(
userName, null, FsPermission.getDefault());
- INodeSymlink newNode = unprotectedAddSymlink(fsd, iip.getExistingINodes(),
- localName, id, target, mtime, mtime, perm);
+ INodeSymlink newNode = unprotectedAddSymlink(fsd, parent,
+ iip.getLastLocalName(), id, target, mtime, mtime, perm);
if (newNode == null) {
NameNode.stateChangeLog.info("addSymlink: failed to add " + path);
return null;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8b7adf4d/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirWriteFileOp.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirWriteFileOp.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirWriteFileOp.java
index 030d8cc..cb639b1 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirWriteFileOp.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirWriteFileOp.java
@@ -65,7 +65,6 @@ import java.util.Collections;
import java.util.EnumSet;
import java.util.HashSet;
import java.util.List;
-import java.util.Map;
import java.util.Set;
import static org.apache.hadoop.hdfs.server.namenode.snapshot.Snapshot.CURRENT_STATE_ID;
@@ -419,10 +418,10 @@ class FSDirWriteFileOp {
}
fsn.checkFsObjectLimit();
INodeFile newNode = null;
- Map.Entry<INodesInPath, String> parent = FSDirMkdirOp
- .createAncestorDirectories(fsd, iip, permissions);
+ INodesInPath parent =
+ FSDirMkdirOp.createAncestorDirectories(fsd, iip, permissions);
if (parent != null) {
- iip = addFile(fsd, parent.getKey(), parent.getValue(), permissions,
+ iip = addFile(fsd, parent, iip.getLastLocalName(), permissions,
replication, blockSize, holder, clientMachine);
newNode = iip != null ? iip.getLastINode().asFile() : null;
}
@@ -572,7 +571,7 @@ class FSDirWriteFileOp {
* @return the new INodesInPath instance that contains the new INode
*/
private static INodesInPath addFile(
- FSDirectory fsd, INodesInPath existing, String localName,
+ FSDirectory fsd, INodesInPath existing, byte[] localName,
PermissionStatus permissions, short replication, long preferredBlockSize,
String clientName, String clientMachine)
throws IOException {
@@ -589,7 +588,7 @@ class FSDirWriteFileOp {
}
INodeFile newNode = newINodeFile(fsd.allocateNewInodeId(), permissions,
modTime, modTime, replication, preferredBlockSize, ecPolicy != null);
- newNode.setLocalName(DFSUtil.string2Bytes(localName));
+ newNode.setLocalName(localName);
newNode.toUnderConstruction(clientName, clientMachine);
newiip = fsd.addINode(existing, newNode);
} finally {
@@ -597,12 +596,13 @@ class FSDirWriteFileOp {
}
if (newiip == null) {
NameNode.stateChangeLog.info("DIR* addFile: failed to add " +
- existing.getPath() + "/" + localName);
+ existing.getPath() + "/" + DFSUtil.bytes2String(localName));
return null;
}
if(NameNode.stateChangeLog.isDebugEnabled()) {
- NameNode.stateChangeLog.debug("DIR* addFile: " + localName + " is added");
+ NameNode.stateChangeLog.debug("DIR* addFile: " +
+ DFSUtil.bytes2String(localName) + " is added");
}
return newiip;
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8b7adf4d/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/INodesInPath.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/INodesInPath.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/INodesInPath.java
index af8998f..8f65ff8 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/INodesInPath.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/INodesInPath.java
@@ -22,7 +22,6 @@ import java.util.Collections;
import java.util.List;
import java.util.NoSuchElementException;
-import com.google.common.collect.ImmutableList;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.Path;
@@ -361,6 +360,10 @@ public class INodesInPath {
return path;
}
+ public byte[] getPathComponent(int i) {
+ return path[i];
+ }
+
/** @return the full path in string form */
public String getPath() {
return DFSUtil.byteArray2PathString(path);
@@ -374,21 +377,6 @@ public class INodesInPath {
return DFSUtil.byteArray2PathString(path, 0, pos + 1); // it's a length...
}
- /**
- * @param offset start endpoint (inclusive)
- * @param length number of path components
- * @return sub-list of the path
- */
- public List<String> getPath(int offset, int length) {
- Preconditions.checkArgument(offset >= 0 && length >= 0 && offset + length
- <= path.length);
- ImmutableList.Builder<String> components = ImmutableList.builder();
- for (int i = offset; i < offset + length; i++) {
- components.add(DFSUtil.bytes2String(path[i]));
- }
- return components.build();
- }
-
public int length() {
return inodes.length;
}
@@ -429,22 +417,17 @@ public class INodesInPath {
}
/**
- * @return a new INodesInPath instance that only contains exisitng INodes.
+ * @return a new INodesInPath instance that only contains existing INodes.
* Note that this method only handles non-snapshot paths.
*/
public INodesInPath getExistingINodes() {
Preconditions.checkState(!isSnapshot());
- int i = 0;
- for (; i < inodes.length; i++) {
- if (inodes[i] == null) {
- break;
+ for (int i = inodes.length; i > 0; i--) {
+ if (inodes[i - 1] != null) {
+ return (i == inodes.length) ? this : getAncestorINodesInPath(i);
}
}
- INode[] existing = new INode[i];
- byte[][] existingPath = new byte[i][];
- System.arraycopy(inodes, 0, existing, 0, i);
- System.arraycopy(path, 0, existingPath, 0, i);
- return new INodesInPath(existing, existingPath, isRaw, false, snapshotId);
+ return null;
}
/**
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8b7adf4d/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestSnapshotPathINodes.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestSnapshotPathINodes.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestSnapshotPathINodes.java
index d022903..24ec1a2 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestSnapshotPathINodes.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestSnapshotPathINodes.java
@@ -146,6 +146,11 @@ public class TestSnapshotPathINodes {
// The returned nodesInPath should be non-snapshot
assertSnapshot(nodesInPath, false, null, -1);
+ // verify components are correct
+ for (int i=0; i < components.length; i++) {
+ assertEquals(components[i], nodesInPath.getPathComponent(i));
+ }
+
// The last INode should be associated with file1
assertTrue("file1=" + file1 + ", nodesInPath=" + nodesInPath,
nodesInPath.getINode(components.length - 1) != null);
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
[8/8] hadoop git commit: HADOOP-13355. Handle HADOOP_CLIENT_OPTS in a
function (aw)
Posted by aw...@apache.org.
HADOOP-13355. Handle HADOOP_CLIENT_OPTS in a function (aw)
Signed-off-by: Allen Wittenauer <aw...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/cec31024
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/cec31024
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/cec31024
Branch: refs/heads/HADOOP-13341
Commit: cec31024b8c4c0ce3d2833f6642ea7df692af5cb
Parents: defe759
Author: Allen Wittenauer <aw...@apache.org>
Authored: Fri Aug 26 14:33:33 2016 -0700
Committer: Allen Wittenauer <aw...@apache.org>
Committed: Fri Aug 26 17:17:29 2016 -0700
----------------------------------------------------------------------
.../hadoop-common/src/main/bin/hadoop | 6 +--
.../src/main/bin/hadoop-functions.sh | 21 ++++++++++
.../test/scripts/hadoop_add_client_opts.bats | 40 ++++++++++++++++++++
.../hadoop-hdfs/src/main/bin/hdfs | 14 +------
hadoop-mapreduce-project/bin/mapred | 12 +-----
.../main/shellprofile.d/hadoop-archive-logs.sh | 2 -
.../src/main/shellprofile.d/hadoop-distcp.sh | 4 --
.../src/main/shellprofile.d/hadoop-extras.sh | 2 -
.../src/main/shellprofile.d/hadoop-rumen.sh | 4 --
.../hadoop-sls/src/main/bin/rumen2sls.sh | 3 +-
hadoop-tools/hadoop-sls/src/main/bin/slsrun.sh | 3 +-
.../src/main/shellprofile.d/hadoop-streaming.sh | 4 --
hadoop-yarn-project/hadoop-yarn/bin/yarn | 34 ++++++-----------
13 files changed, 81 insertions(+), 68 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/cec31024/hadoop-common-project/hadoop-common/src/main/bin/hadoop
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/bin/hadoop b/hadoop-common-project/hadoop-common/src/main/bin/hadoop
index b57a4c1..4aae621 100755
--- a/hadoop-common-project/hadoop-common/src/main/bin/hadoop
+++ b/hadoop-common-project/hadoop-common/src/main/bin/hadoop
@@ -161,10 +161,6 @@ function hadoopcmd_case
fi
;;
esac
-
- # Always respect HADOOP_OPTS and HADOOP_CLIENT_OPTS
- hadoop_debug "Appending HADOOP_CLIENT_OPTS onto HADOOP_OPTS"
- HADOOP_OPTS="${HADOOP_OPTS} ${HADOOP_CLIENT_OPTS}"
}
# This script runs the hadoop core commands.
@@ -205,6 +201,8 @@ fi
hadoop_verify_user "${HADOOP_SUBCMD}"
+hadoop_add_client_opts
+
if [[ ${HADOOP_WORKER_MODE} = true ]]; then
hadoop_common_worker_mode_execute "${HADOOP_COMMON_HOME}/bin/hadoop" "${HADOOP_USER_PARAMS[@]}"
exit $?
http://git-wip-us.apache.org/repos/asf/hadoop/blob/cec31024/hadoop-common-project/hadoop-common/src/main/bin/hadoop-functions.sh
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/bin/hadoop-functions.sh b/hadoop-common-project/hadoop-common/src/main/bin/hadoop-functions.sh
index 6e58dca..9003913 100755
--- a/hadoop-common-project/hadoop-common/src/main/bin/hadoop-functions.sh
+++ b/hadoop-common-project/hadoop-common/src/main/bin/hadoop-functions.sh
@@ -306,6 +306,13 @@ function hadoop_bootstrap
HADOOP_TOOLS_DIR=${HADOOP_TOOLS_DIR:-"share/hadoop/tools"}
HADOOP_TOOLS_LIB_JARS_DIR=${HADOOP_TOOLS_LIB_JARS_DIR:-"${HADOOP_TOOLS_DIR}/lib"}
+ # by default, whatever we are about to run doesn't support
+ # daemonization
+ HADOOP_SUBCMD_SUPPORTDAEMONIZATION=false
+
+ # shellcheck disable=SC2034
+ HADOOP_SUBCMD_SECURESERVICE=false
+
# usage output set to zero
hadoop_reset_usage
@@ -1230,6 +1237,20 @@ function hadoop_translate_cygwin_path
fi
}
+## @description Adds the HADOOP_CLIENT_OPTS variable to
+## @description HADOOP_OPTS if HADOOP_SUBCMD_SUPPORTDAEMONIZATION is false
+## @audience public
+## @stability stable
+## @replaceable yes
+function hadoop_add_client_opts
+{
+ if [[ "${HADOOP_SUBCMD_SUPPORTDAEMONIZATION}" = false
+ || -z "${HADOOP_SUBCMD_SUPPORTDAEMONIZATION}" ]]; then
+ hadoop_debug "Appending HADOOP_CLIENT_OPTS onto HADOOP_OPTS"
+ HADOOP_OPTS="${HADOOP_OPTS} ${HADOOP_CLIENT_OPTS}"
+ fi
+}
+
## @description Finish configuring Hadoop specific system properties
## @description prior to executing Java
## @audience private
http://git-wip-us.apache.org/repos/asf/hadoop/blob/cec31024/hadoop-common-project/hadoop-common/src/test/scripts/hadoop_add_client_opts.bats
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/test/scripts/hadoop_add_client_opts.bats b/hadoop-common-project/hadoop-common/src/test/scripts/hadoop_add_client_opts.bats
new file mode 100644
index 0000000..bc5051d
--- /dev/null
+++ b/hadoop-common-project/hadoop-common/src/test/scripts/hadoop_add_client_opts.bats
@@ -0,0 +1,40 @@
+# 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.
+
+load hadoop-functions_test_helper
+
+@test "hadoop_subcommand_opts (daemonization false)" {
+ HADOOP_OPTS="1"
+ HADOOP_CLIENT_OPTS="2"
+ HADOOP_SUBCMD_SUPPORTDAEMONIZATION="false"
+ hadoop_add_client_opts
+ [ "${HADOOP_OPTS}" = "1 2" ]
+}
+
+@test "hadoop_subcommand_opts (daemonization true)" {
+ HADOOP_OPTS="1"
+ HADOOP_CLIENT_OPTS="2"
+ HADOOP_SUBCMD_SUPPORTDAEMONIZATION="true"
+ hadoop_add_client_opts
+ [ "${HADOOP_OPTS}" = "1" ]
+}
+
+@test "hadoop_subcommand_opts (daemonization empty)" {
+ HADOOP_OPTS="1"
+ HADOOP_CLIENT_OPTS="2"
+ unset HADOOP_SUBCMD_SUPPORTDAEMONIZATION
+ hadoop_add_client_opts
+ [ "${HADOOP_OPTS}" = "1 2" ]
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/cec31024/hadoop-hdfs-project/hadoop-hdfs/src/main/bin/hdfs
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/bin/hdfs b/hadoop-hdfs-project/hadoop-hdfs/src/main/bin/hdfs
index 5059528..41b3c12 100755
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/bin/hdfs
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/bin/hdfs
@@ -118,18 +118,12 @@ function hdfscmd_case
;;
dfs)
HADOOP_CLASSNAME=org.apache.hadoop.fs.FsShell
- hadoop_debug "Appending HADOOP_CLIENT_OPTS onto HADOOP_OPTS"
- HADOOP_OPTS="${HADOOP_OPTS} ${HADOOP_CLIENT_OPTS}"
;;
dfsadmin)
HADOOP_CLASSNAME=org.apache.hadoop.hdfs.tools.DFSAdmin
- hadoop_debug "Appending HADOOP_CLIENT_OPTS onto HADOOP_OPTS"
- HADOOP_OPTS="${HADOOP_OPTS} ${HADOOP_CLIENT_OPTS}"
;;
diskbalancer)
HADOOP_CLASSNAME=org.apache.hadoop.hdfs.tools.DiskBalancer
- hadoop_debug "Appending HADOOP_CLIENT_OPTS onto HADOOP_OPTS"
- HADOOP_OPTS="${HADOOP_OPTS} ${HADOOP_CLIENT_OPTS}"
;;
envvars)
echo "JAVA_HOME='${JAVA_HOME}'"
@@ -144,16 +138,12 @@ function hdfscmd_case
;;
erasurecode)
HADOOP_CLASSNAME=org.apache.hadoop.hdfs.tools.erasurecode.ECCli
- hadoop_debug "Appending HADOOP_CLIENT_OPTS onto HADOOP_OPTS"
- HADOOP_OPTS="${HADOOP_OPTS} ${HADOOP_CLIENT_OPTS}"
;;
fetchdt)
HADOOP_CLASSNAME=org.apache.hadoop.hdfs.tools.DelegationTokenFetcher
;;
fsck)
HADOOP_CLASSNAME=org.apache.hadoop.hdfs.tools.DFSck
- hadoop_debug "Appending HADOOP_CLIENT_OPTS onto HADOOP_OPTS"
- HADOOP_OPTS="${HADOOP_OPTS} ${HADOOP_CLIENT_OPTS}"
;;
getconf)
HADOOP_CLASSNAME=org.apache.hadoop.hdfs.tools.GetConf
@@ -163,8 +153,6 @@ function hdfscmd_case
;;
haadmin)
HADOOP_CLASSNAME=org.apache.hadoop.hdfs.tools.DFSHAAdmin
- hadoop_debug "Appending HADOOP_CLIENT_OPTS onto HADOOP_OPTS"
- HADOOP_OPTS="${HADOOP_OPTS} ${HADOOP_CLIENT_OPTS}"
;;
journalnode)
HADOOP_SUBCMD_SUPPORTDAEMONIZATION="true"
@@ -293,6 +281,8 @@ fi
hadoop_verify_user "${HADOOP_SUBCMD}"
+hadoop_add_client_opts
+
if [[ ${HADOOP_WORKER_MODE} = true ]]; then
hadoop_common_worker_mode_execute "${HADOOP_HDFS_HOME}/bin/hdfs" "${HADOOP_USER_PARAMS[@]}"
exit $?
http://git-wip-us.apache.org/repos/asf/hadoop/blob/cec31024/hadoop-mapreduce-project/bin/mapred
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/bin/mapred b/hadoop-mapreduce-project/bin/mapred
index 046d48c..3243d20 100755
--- a/hadoop-mapreduce-project/bin/mapred
+++ b/hadoop-mapreduce-project/bin/mapred
@@ -79,31 +79,21 @@ function mapredcmd_case
;;
hsadmin)
HADOOP_CLASSNAME=org.apache.hadoop.mapreduce.v2.hs.client.HSAdmin
- hadoop_debug "Appending HADOOP_CLIENT_OPTS onto HADOOP_OPTS"
- HADOOP_OPTS="${HADOOP_OPTS} ${HADOOP_CLIENT_OPTS}"
;;
job)
HADOOP_CLASSNAME=org.apache.hadoop.mapred.JobClient
- hadoop_debug "Appending HADOOP_CLIENT_OPTS onto HADOOP_OPTS"
- HADOOP_OPTS="${HADOOP_OPTS} ${HADOOP_CLIENT_OPTS}"
;;
pipes)
HADOOP_CLASSNAME=org.apache.hadoop.mapred.pipes.Submitter
- hadoop_debug "Appending HADOOP_CLIENT_OPTS onto HADOOP_OPTS"
- HADOOP_OPTS="${HADOOP_OPTS} ${HADOOP_CLIENT_OPTS}"
;;
queue)
HADOOP_CLASSNAME=org.apache.hadoop.mapred.JobQueueClient
;;
sampler)
HADOOP_CLASSNAME=org.apache.hadoop.mapred.lib.InputSampler
- hadoop_debug "Appending HADOOP_CLIENT_OPTS onto HADOOP_OPTS"
- HADOOP_OPTS="${HADOOP_OPTS} ${HADOOP_CLIENT_OPTS}"
;;
version)
HADOOP_CLASSNAME=org.apache.hadoop.util.VersionInfo
- hadoop_debug "Appending HADOOP_CLIENT_OPTS onto HADOOP_OPTS"
- HADOOP_OPTS="${HADOOP_OPTS} ${HADOOP_CLIENT_OPTS}"
;;
*)
HADOOP_CLASSNAME="${subcmd}"
@@ -152,6 +142,8 @@ fi
hadoop_verify_user "${HADOOP_SUBCMD}"
+hadoop_add_client_opts
+
if [[ ${HADOOP_SLAVE_MODE} = true ]]; then
hadoop_common_slave_mode_execute "${HADOOP_MAPRED_HOME}/bin/mapred" "${HADOOP_USER_PARAMS[@]}"
exit $?
http://git-wip-us.apache.org/repos/asf/hadoop/blob/cec31024/hadoop-tools/hadoop-archive-logs/src/main/shellprofile.d/hadoop-archive-logs.sh
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-archive-logs/src/main/shellprofile.d/hadoop-archive-logs.sh b/hadoop-tools/hadoop-archive-logs/src/main/shellprofile.d/hadoop-archive-logs.sh
index ae7b6c6..c889816 100755
--- a/hadoop-tools/hadoop-archive-logs/src/main/shellprofile.d/hadoop-archive-logs.sh
+++ b/hadoop-tools/hadoop-archive-logs/src/main/shellprofile.d/hadoop-archive-logs.sh
@@ -32,8 +32,6 @@ function mapred_subcommand_archive-logs
# shellcheck disable=SC2034
HADOOP_CLASSNAME=org.apache.hadoop.tools.HadoopArchiveLogs
hadoop_add_to_classpath_tools hadoop-archive-logs
- hadoop_debug "Appending HADOOP_CLIENT_OPTS onto HADOOP_OPTS"
- HADOOP_OPTS="${HADOOP_OPTS} ${HADOOP_CLIENT_OPTS}"
}
fi
http://git-wip-us.apache.org/repos/asf/hadoop/blob/cec31024/hadoop-tools/hadoop-distcp/src/main/shellprofile.d/hadoop-distcp.sh
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-distcp/src/main/shellprofile.d/hadoop-distcp.sh b/hadoop-tools/hadoop-distcp/src/main/shellprofile.d/hadoop-distcp.sh
index 0178c54..6e93ec1 100755
--- a/hadoop-tools/hadoop-distcp/src/main/shellprofile.d/hadoop-distcp.sh
+++ b/hadoop-tools/hadoop-distcp/src/main/shellprofile.d/hadoop-distcp.sh
@@ -32,8 +32,6 @@ function hadoop_subcommand_distcp
# shellcheck disable=SC2034
HADOOP_CLASSNAME=org.apache.hadoop.tools.DistCp
hadoop_add_to_classpath_tools hadoop-distcp
- hadoop_debug "Appending HADOOP_CLIENT_OPTS onto HADOOP_OPTS"
- HADOOP_OPTS="${HADOOP_OPTS} ${HADOOP_CLIENT_OPTS}"
}
fi
@@ -55,8 +53,6 @@ function mapred_subcommand_distcp
# shellcheck disable=SC2034
HADOOP_CLASSNAME=org.apache.hadoop.tools.DistCp
hadoop_add_to_classpath_tools hadoop-distcp
- hadoop_debug "Appending HADOOP_CLIENT_OPTS onto HADOOP_OPTS"
- HADOOP_OPTS="${HADOOP_OPTS} ${HADOOP_CLIENT_OPTS}"
}
fi
http://git-wip-us.apache.org/repos/asf/hadoop/blob/cec31024/hadoop-tools/hadoop-extras/src/main/shellprofile.d/hadoop-extras.sh
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-extras/src/main/shellprofile.d/hadoop-extras.sh b/hadoop-tools/hadoop-extras/src/main/shellprofile.d/hadoop-extras.sh
index 829d406..1ce9aee 100755
--- a/hadoop-tools/hadoop-extras/src/main/shellprofile.d/hadoop-extras.sh
+++ b/hadoop-tools/hadoop-extras/src/main/shellprofile.d/hadoop-extras.sh
@@ -32,8 +32,6 @@ function hadoop_subcommand_distch
# shellcheck disable=SC2034
HADOOP_CLASSNAME=org.apache.hadoop.tools.DistCh
hadoop_add_to_classpath_tools hadoop-extras
- hadoop_debug "Appending HADOOP_CLIENT_OPTS onto HADOOP_OPTS"
- HADOOP_OPTS="${HADOOP_OPTS} ${HADOOP_CLIENT_OPTS}"
}
fi
http://git-wip-us.apache.org/repos/asf/hadoop/blob/cec31024/hadoop-tools/hadoop-rumen/src/main/shellprofile.d/hadoop-rumen.sh
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-rumen/src/main/shellprofile.d/hadoop-rumen.sh b/hadoop-tools/hadoop-rumen/src/main/shellprofile.d/hadoop-rumen.sh
index d7d4022..77023ff 100755
--- a/hadoop-tools/hadoop-rumen/src/main/shellprofile.d/hadoop-rumen.sh
+++ b/hadoop-tools/hadoop-rumen/src/main/shellprofile.d/hadoop-rumen.sh
@@ -30,8 +30,6 @@ function hadoop_subcommand_rumenfolder
# shellcheck disable=SC2034
HADOOP_CLASSNAME=org.apache.hadoop.tools.rumen.Folder
hadoop_add_to_classpath_tools hadoop-rumen
- hadoop_debug "Appending HADOOP_CLIENT_OPTS onto HADOOP_OPTS"
- HADOOP_OPTS="${HADOOP_OPTS} ${HADOOP_CLIENT_OPTS}"
}
fi
@@ -51,8 +49,6 @@ function hadoop_subcommand_rumentrace
# shellcheck disable=SC2034
HADOOP_CLASSNAME=org.apache.hadoop.tools.rumen.TraceBuilder
hadoop_add_to_classpath_tools hadoop-rumen
- hadoop_debug "Appending HADOOP_CLIENT_OPTS onto HADOOP_OPTS"
- HADOOP_OPTS="${HADOOP_OPTS} ${HADOOP_CLIENT_OPTS}"
}
fi
http://git-wip-us.apache.org/repos/asf/hadoop/blob/cec31024/hadoop-tools/hadoop-sls/src/main/bin/rumen2sls.sh
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-sls/src/main/bin/rumen2sls.sh b/hadoop-tools/hadoop-sls/src/main/bin/rumen2sls.sh
index 0bd291b..565dfe6 100644
--- a/hadoop-tools/hadoop-sls/src/main/bin/rumen2sls.sh
+++ b/hadoop-tools/hadoop-sls/src/main/bin/rumen2sls.sh
@@ -68,8 +68,7 @@ function run_sls_generator()
hadoop_add_param args -outputJobs "-outputJobs ${outputdir}/${outputprefix}-jobs.json"
hadoop_add_param args -outputNodes "-outputNodes ${outputdir}/${outputprefix}-nodes.json"
- hadoop_debug "Appending HADOOP_CLIENT_OPTS onto HADOOP_OPTS"
- HADOOP_OPTS="${HADOOP_OPTS} ${HADOOP_CLIENT_OPTS}"
+ hadoop_add_client_opts
hadoop_finalize
# shellcheck disable=SC2086
http://git-wip-us.apache.org/repos/asf/hadoop/blob/cec31024/hadoop-tools/hadoop-sls/src/main/bin/slsrun.sh
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-sls/src/main/bin/slsrun.sh b/hadoop-tools/hadoop-sls/src/main/bin/slsrun.sh
index 403c4bb..218dee4 100644
--- a/hadoop-tools/hadoop-sls/src/main/bin/slsrun.sh
+++ b/hadoop-tools/hadoop-sls/src/main/bin/slsrun.sh
@@ -96,8 +96,7 @@ function run_simulation() {
hadoop_add_param args -printsimulation "-printsimulation"
fi
- hadoop_debug "Appending HADOOP_CLIENT_OPTS onto HADOOP_OPTS"
- HADOOP_OPTS="${HADOOP_OPTS} ${HADOOP_CLIENT_OPTS}"
+ hadoop_add_client_opts
hadoop_finalize
# shellcheck disable=SC2086
http://git-wip-us.apache.org/repos/asf/hadoop/blob/cec31024/hadoop-tools/hadoop-streaming/src/main/shellprofile.d/hadoop-streaming.sh
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-streaming/src/main/shellprofile.d/hadoop-streaming.sh b/hadoop-tools/hadoop-streaming/src/main/shellprofile.d/hadoop-streaming.sh
index cca016d..c3010ff 100755
--- a/hadoop-tools/hadoop-streaming/src/main/shellprofile.d/hadoop-streaming.sh
+++ b/hadoop-tools/hadoop-streaming/src/main/shellprofile.d/hadoop-streaming.sh
@@ -46,10 +46,6 @@ function mapred_subcommand_streaming
done
IFS=${oldifs}
-
- hadoop_debug "Appending HADOOP_CLIENT_OPTS onto HADOOP_OPTS"
- HADOOP_OPTS="${HADOOP_OPTS} ${HADOOP_CLIENT_OPTS}"
-
}
fi
http://git-wip-us.apache.org/repos/asf/hadoop/blob/cec31024/hadoop-yarn-project/hadoop-yarn/bin/yarn
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/bin/yarn b/hadoop-yarn-project/hadoop-yarn/bin/yarn
index bd91633..66a87b6 100755
--- a/hadoop-yarn-project/hadoop-yarn/bin/yarn
+++ b/hadoop-yarn-project/hadoop-yarn/bin/yarn
@@ -68,8 +68,6 @@ function yarncmd_case
case ${subcmd} in
application|applicationattempt|container)
HADOOP_CLASSNAME=org.apache.hadoop.yarn.client.cli.ApplicationCLI
- hadoop_debug "Append YARN_CLIENT_OPTS onto HADOOP_OPTS"
- HADOOP_OPTS="${HADOOP_OPTS} ${YARN_CLIENT_OPTS}"
set -- "${subcmd}" "$@"
HADOOP_SUBCMD_ARGS=("$@")
;;
@@ -78,13 +76,9 @@ function yarncmd_case
;;
cluster)
HADOOP_CLASSNAME=org.apache.hadoop.yarn.client.cli.ClusterCLI
- hadoop_debug "Append YARN_CLIENT_OPTS onto YARN_OPTS"
- YARN_OPTS="${YARN_OPTS} ${YARN_CLIENT_OPTS}"
;;
daemonlog)
HADOOP_CLASSNAME=org.apache.hadoop.log.LogLevel
- hadoop_debug "Append YARN_CLIENT_OPTS onto HADOOP_OPTS"
- HADOOP_OPTS="${HADOOP_OPTS} ${YARN_CLIENT_OPTS}"
;;
envvars)
echo "JAVA_HOME='${JAVA_HOME}'"
@@ -99,8 +93,6 @@ function yarncmd_case
;;
jar)
HADOOP_CLASSNAME=org.apache.hadoop.util.RunJar
- hadoop_debug "Append YARN_CLIENT_OPTS onto HADOOP_OPTS"
- HADOOP_OPTS="${HADOOP_OPTS} ${YARN_CLIENT_OPTS}"
;;
historyserver)
HADOOP_SUBCMD_SUPPORTDAEMONIZATION="true"
@@ -111,13 +103,9 @@ function yarncmd_case
;;
logs)
HADOOP_CLASSNAME=org.apache.hadoop.yarn.client.cli.LogsCLI
- hadoop_debug "Append YARN_CLIENT_OPTS onto HADOOP_OPTS"
- HADOOP_OPTS="${HADOOP_OPTS} ${YARN_CLIENT_OPTS}"
;;
node)
HADOOP_CLASSNAME=org.apache.hadoop.yarn.client.cli.NodeCLI
- hadoop_debug "Append YARN_CLIENT_OPTS onto HADOOP_OPTS"
- HADOOP_OPTS="${HADOOP_OPTS} ${YARN_CLIENT_OPTS}"
;;
nodemanager)
HADOOP_SUBCMD_SUPPORTDAEMONIZATION="true"
@@ -142,8 +130,6 @@ function yarncmd_case
;;
queue)
HADOOP_CLASSNAME=org.apache.hadoop.yarn.client.cli.QueueCLI
- hadoop_debug "Append YARN_CLIENT_OPTS onto HADOOP_OPTS"
- HADOOP_OPTS="${HADOOP_OPTS} ${YARN_CLIENT_OPTS}"
;;
resourcemanager)
HADOOP_SUBCMD_SUPPORTDAEMONIZATION="true"
@@ -158,13 +144,9 @@ function yarncmd_case
;;
rmadmin)
HADOOP_CLASSNAME='org.apache.hadoop.yarn.client.cli.RMAdminCLI'
- hadoop_debug "Append YARN_CLIENT_OPTS onto HADOOP_OPTS"
- HADOOP_OPTS="${HADOOP_OPTS} ${YARN_CLIENT_OPTS}"
;;
scmadmin)
HADOOP_CLASSNAME='org.apache.hadoop.yarn.client.SCMAdmin'
- hadoop_debug "Append YARN_CLIENT_OPTS onto HADOOP_OPTS"
- HADOOP_OPTS="${HADOOP_OPTS} ${YARN_CLIENT_OPTS}"
;;
sharedcachemanager)
HADOOP_SUBCMD_SUPPORTDAEMONIZATION="true"
@@ -191,8 +173,6 @@ function yarncmd_case
;;
version)
HADOOP_CLASSNAME=org.apache.hadoop.util.VersionInfo
- hadoop_debug "Append YARN_CLIENT_OPTS onto HADOOP_OPTS"
- HADOOP_OPTS="${HADOOP_OPTS} ${YARN_CLIENT_OPTS}"
;;
top)
doNotSetCols=0
@@ -222,8 +202,6 @@ function yarncmd_case
fi
fi
HADOOP_CLASSNAME=org.apache.hadoop.yarn.client.cli.TopCLI
- hadoop_debug "Append YARN_CLIENT_OPTS onto HADOOP_OPTS"
- HADOOP_OPTS="${HADOOP_OPTS} ${YARN_CLIENT_OPTS}"
HADOOP_SUBCMD_ARGS=("$@")
;;
*)
@@ -273,6 +251,18 @@ fi
hadoop_verify_user "${HADOOP_SUBCMD}"
+# It's unclear if YARN_CLIENT_OPTS is actually a useful
+# thing to have separate from HADOOP_CLIENT_OPTS. Someone
+# might use it, so let's not deprecate it and just override
+# HADOOP_CLIENT_OPTS instead before we (potentially) add it
+# to the command line
+if [[ -n "${YARN_CLIENT_OPTS}" ]]; then
+ # shellcheck disable=SC2034
+ HADOOP_CLIENT_OPTS=${YARN_CLIENT_OPTS}
+fi
+
+hadoop_add_client_opts
+
if [[ ${HADOOP_WORKER_MODE} = true ]]; then
hadoop_common_worker_mode_execute "${HADOOP_YARN_HOME}/bin/yarn" "${HADOOP_USER_PARAMS[@]}"
exit $?
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
[5/8] hadoop git commit: YARN-5327. API changes required to support
recurring reservations in the YARN ReservationSystem. (Sangeetha Abdu Jyothi
via Subru).
Posted by aw...@apache.org.
YARN-5327. API changes required to support recurring reservations in the YARN ReservationSystem. (Sangeetha Abdu Jyothi via Subru).
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/b930dc3e
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/b930dc3e
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/b930dc3e
Branch: refs/heads/HADOOP-13341
Commit: b930dc3ec06afa479a249490976e3e127d201706
Parents: e806db7
Author: Subru Krishnan <su...@apache.org>
Authored: Fri Aug 26 16:58:47 2016 -0700
Committer: Subru Krishnan <su...@apache.org>
Committed: Fri Aug 26 16:58:47 2016 -0700
----------------------------------------------------------------------
.../yarn/api/records/ReservationDefinition.java | 49 +++++++++++++++-
.../src/main/proto/yarn_protos.proto | 1 +
.../impl/pb/ReservationDefinitionPBImpl.java | 21 ++++++-
.../reservation/ReservationInputValidator.java | 14 +++++
.../TestReservationInputValidator.java | 61 ++++++++++++++++++++
5 files changed, 143 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/b930dc3e/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/ReservationDefinition.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/ReservationDefinition.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/ReservationDefinition.java
index 10f592a..8ef881b 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/ReservationDefinition.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/ReservationDefinition.java
@@ -37,13 +37,24 @@ public abstract class ReservationDefinition {
@Public
@Unstable
public static ReservationDefinition newInstance(long arrival, long deadline,
- ReservationRequests reservationRequests, String name) {
+ ReservationRequests reservationRequests, String name,
+ String recurrenceExpression) {
ReservationDefinition rDefinition =
Records.newRecord(ReservationDefinition.class);
rDefinition.setArrival(arrival);
rDefinition.setDeadline(deadline);
rDefinition.setReservationRequests(reservationRequests);
rDefinition.setReservationName(name);
+ rDefinition.setRecurrenceExpression(recurrenceExpression);
+ return rDefinition;
+ }
+
+ @Public
+ @Unstable
+ public static ReservationDefinition newInstance(long arrival, long deadline,
+ ReservationRequests reservationRequests, String name) {
+ ReservationDefinition rDefinition =
+ newInstance(arrival, deadline, reservationRequests, name, "0");
return rDefinition;
}
@@ -134,4 +145,40 @@ public abstract class ReservationDefinition {
@Evolving
public abstract void setReservationName(String name);
+ /**
+ * Get the recurrence of this reservation representing the time period of
+ * the periodic job. Currently, only long values are supported. Later,
+ * support for regular expressions denoting arbitrary recurrence patterns
+ * (e.g., every Tuesday and Thursday) will be added.
+ * Recurrence is represented in milliseconds for periodic jobs.
+ * Recurrence is 0 for non-periodic jobs. Periodic jobs are valid until they
+ * are explicitly cancelled and have higher priority than non-periodic jobs
+ * (during initial placement and replanning). Periodic job allocations are
+ * consistent across runs (flexibility in allocation is leveraged only during
+ * initial placement, allocations remain consistent thereafter).
+ *
+ * @return recurrence of this reservation
+ */
+ @Public
+ @Evolving
+ public abstract String getRecurrenceExpression();
+
+ /**
+ * Set the recurrence of this reservation representing the time period of
+ * the periodic job. Currently, only long values are supported. Later,
+ * support for regular expressions denoting arbitrary recurrence patterns
+ * (e.g., every Tuesday and Thursday) will be added.
+ * Recurrence is represented in milliseconds for periodic jobs.
+ * Recurrence is 0 for non-periodic jobs. Periodic jobs are valid until they
+ * are explicitly cancelled and have higher priority than non-periodic jobs
+ * (during initial placement and replanning). Periodic job allocations are
+ * consistent across runs (flexibility in allocation is leveraged only during
+ * initial placement, allocations remain consistent thereafter).
+ *
+ * @param recurrenceExpression recurrence interval of this reservation
+ */
+ @Public
+ @Evolving
+ public abstract void setRecurrenceExpression(String recurrenceExpression);
+
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/b930dc3e/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_protos.proto
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_protos.proto b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_protos.proto
index 8a6ae3f..2cc1784 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_protos.proto
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_protos.proto
@@ -477,6 +477,7 @@ message ReservationDefinitionProto {
optional int64 arrival = 2;
optional int64 deadline = 3;
optional string reservation_name = 4;
+ optional string recurrence_expression = 5 [default = "0"];
}
message ResourceAllocationRequestProto {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/b930dc3e/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/records/impl/pb/ReservationDefinitionPBImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/records/impl/pb/ReservationDefinitionPBImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/records/impl/pb/ReservationDefinitionPBImpl.java
index cbe7ccd..b30cd2a 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/records/impl/pb/ReservationDefinitionPBImpl.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/records/impl/pb/ReservationDefinitionPBImpl.java
@@ -162,8 +162,9 @@ public class ReservationDefinitionPBImpl extends ReservationDefinition {
@Override
public String toString() {
return "{Arrival: " + getArrival() + ", Deadline: " + getDeadline()
- + ", Reservation Name: " + getReservationName() + ", Resources: "
- + getReservationRequests() + "}";
+ + ", Reservation Name: " + getReservationName()
+ + ", Recurrence expression: " + getRecurrenceExpression()
+ + ", Resources: " + getReservationRequests() + "}";
}
@Override
@@ -181,4 +182,20 @@ public class ReservationDefinitionPBImpl extends ReservationDefinition {
return false;
}
+ @Override
+ public String getRecurrenceExpression() {
+ ReservationDefinitionProtoOrBuilder p = viaProto ? proto : builder;
+ if (p.hasRecurrenceExpression()) {
+ String recurrenceExpression = p.getRecurrenceExpression();
+ return recurrenceExpression;
+ } else {
+ return "0";
+ }
+ }
+
+ @Override
+ public void setRecurrenceExpression(String recurrenceExpression) {
+ builder.setRecurrenceExpression(recurrenceExpression);
+ }
+
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/b930dc3e/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/ReservationInputValidator.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/ReservationInputValidator.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/ReservationInputValidator.java
index 8ebde0a..068894f 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/ReservationInputValidator.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/ReservationInputValidator.java
@@ -155,6 +155,20 @@ public class ReservationInputValidator {
"validate reservation input definition", "ClientRMService", message);
throw RPCUtil.getRemoteException(message);
}
+ // check that the recurrence is a positive long value.
+ String recurrenceExpression = contract.getRecurrenceExpression();
+ try {
+ Long recurrence = Long.parseLong(recurrenceExpression);
+ if (recurrence < 0) {
+ message = "Negative Period : " + recurrenceExpression + ". Please try"
+ + " again with a non-negative long value as period";
+ throw RPCUtil.getRemoteException(message);
+ }
+ } catch (NumberFormatException e) {
+ message = "Invalid period " + recurrenceExpression + ". Please try"
+ + " again with a non-negative long value as period";
+ throw RPCUtil.getRemoteException(message);
+ }
}
private Plan getPlanFromQueue(ReservationSystem reservationSystem, String
http://git-wip-us.apache.org/repos/asf/hadoop/blob/b930dc3e/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/TestReservationInputValidator.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/TestReservationInputValidator.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/TestReservationInputValidator.java
index bd40ccd..031a767 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/TestReservationInputValidator.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/TestReservationInputValidator.java
@@ -270,6 +270,59 @@ public class TestReservationInputValidator {
}
@Test
+ public void testSubmitReservationValidRecurrenceExpression() {
+ ReservationSubmissionRequest request =
+ createSimpleReservationSubmissionRequest(1, 1, 1, 5, 3, "600000");
+ plan = null;
+ try {
+ plan =
+ rrValidator.validateReservationSubmissionRequest(rSystem, request,
+ ReservationSystemTestUtil.getNewReservationId());
+ } catch (YarnException e) {
+ Assert.fail(e.getMessage());
+ }
+ Assert.assertNotNull(plan);
+ }
+
+ @Test
+ public void testSubmitReservationNegativeRecurrenceExpression() {
+ ReservationSubmissionRequest request =
+ createSimpleReservationSubmissionRequest(1, 1, 1, 5, 3, "-1234");
+ plan = null;
+ try {
+ plan =
+ rrValidator.validateReservationSubmissionRequest(rSystem, request,
+ ReservationSystemTestUtil.getNewReservationId());
+ Assert.fail();
+ } catch (YarnException e) {
+ Assert.assertNull(plan);
+ String message = e.getMessage();
+ Assert.assertTrue(message
+ .startsWith("Negative Period : "));
+ LOG.info(message);
+ }
+ }
+
+ @Test
+ public void testSubmitReservationInvalidRecurrenceExpression() {
+ ReservationSubmissionRequest request =
+ createSimpleReservationSubmissionRequest(1, 1, 1, 5, 3, "123abc");
+ plan = null;
+ try {
+ plan =
+ rrValidator.validateReservationSubmissionRequest(rSystem, request,
+ ReservationSystemTestUtil.getNewReservationId());
+ Assert.fail();
+ } catch (YarnException e) {
+ Assert.assertNull(plan);
+ String message = e.getMessage();
+ Assert.assertTrue(message
+ .startsWith("Invalid period "));
+ LOG.info(message);
+ }
+ }
+
+ @Test
public void testUpdateReservationNormal() {
ReservationUpdateRequest request =
createSimpleReservationUpdateRequest(1, 1, 1, 5, 3);
@@ -625,12 +678,20 @@ public class TestReservationInputValidator {
private ReservationSubmissionRequest createSimpleReservationSubmissionRequest(
int numRequests, int numContainers, long arrival, long deadline,
long duration) {
+ return createSimpleReservationSubmissionRequest(numRequests, numContainers,
+ arrival, deadline, duration, "0");
+ }
+
+ private ReservationSubmissionRequest createSimpleReservationSubmissionRequest(
+ int numRequests, int numContainers, long arrival, long deadline,
+ long duration, String recurrence) {
// create a request with a single atomic ask
ReservationSubmissionRequest request =
new ReservationSubmissionRequestPBImpl();
ReservationDefinition rDef = new ReservationDefinitionPBImpl();
rDef.setArrival(arrival);
rDef.setDeadline(deadline);
+ rDef.setRecurrenceExpression(recurrence);
if (numRequests > 0) {
ReservationRequests reqs = new ReservationRequestsPBImpl();
rDef.setReservationRequests(reqs);
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
[4/8] hadoop git commit: HDFS-10584. Allow long-running Mover tool to
login with keytab. Contributed by Rakesh R.
Posted by aw...@apache.org.
HDFS-10584. Allow long-running Mover tool to login with keytab. Contributed by Rakesh R.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/e806db71
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/e806db71
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/e806db71
Branch: refs/heads/HADOOP-13341
Commit: e806db719053a5b2a7b14f47e6f2962e70008d25
Parents: a445b82
Author: Zhe Zhang <zh...@apache.org>
Authored: Fri Aug 26 16:43:25 2016 -0700
Committer: Zhe Zhang <zh...@apache.org>
Committed: Fri Aug 26 16:43:45 2016 -0700
----------------------------------------------------------------------
.../org/apache/hadoop/hdfs/DFSConfigKeys.java | 9 +
.../apache/hadoop/hdfs/server/mover/Mover.java | 23 +-
.../src/main/resources/hdfs-default.xml | 40 ++++
.../hadoop/hdfs/server/mover/TestMover.java | 212 +++++++++++++++----
4 files changed, 245 insertions(+), 39 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/e806db71/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java
index b4cce4a..2eff3b0 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java
@@ -487,6 +487,15 @@ public class DFSConfigKeys extends CommonConfigurationKeys {
public static final int DFS_MOVER_MOVERTHREADS_DEFAULT = 1000;
public static final String DFS_MOVER_RETRY_MAX_ATTEMPTS_KEY = "dfs.mover.retry.max.attempts";
public static final int DFS_MOVER_RETRY_MAX_ATTEMPTS_DEFAULT = 10;
+ public static final String DFS_MOVER_KEYTAB_ENABLED_KEY =
+ "dfs.mover.keytab.enabled";
+ public static final boolean DFS_MOVER_KEYTAB_ENABLED_DEFAULT = false;
+ public static final String DFS_MOVER_ADDRESS_KEY = "dfs.mover.address";
+ public static final String DFS_MOVER_ADDRESS_DEFAULT= "0.0.0.0:0";
+ public static final String DFS_MOVER_KEYTAB_FILE_KEY =
+ "dfs.mover.keytab.file";
+ public static final String DFS_MOVER_KERBEROS_PRINCIPAL_KEY =
+ "dfs.mover.kerberos.principal";
public static final String DFS_DATANODE_ADDRESS_KEY = "dfs.datanode.address";
public static final int DFS_DATANODE_DEFAULT_PORT = 9866;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/e806db71/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/mover/Mover.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/mover/Mover.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/mover/Mover.java
index cd37b15b..fdb6cfa 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/mover/Mover.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/mover/Mover.java
@@ -47,7 +47,10 @@ import org.apache.hadoop.hdfs.server.protocol.DatanodeStorageReport;
import org.apache.hadoop.hdfs.server.protocol.StorageReport;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.hdfs.protocol.ErasureCodingPolicy;
+import org.apache.hadoop.net.NetUtils;
import org.apache.hadoop.net.NetworkTopology;
+import org.apache.hadoop.security.SecurityUtil;
+import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.util.StringUtils;
import org.apache.hadoop.util.Time;
import org.apache.hadoop.util.Tool;
@@ -57,6 +60,7 @@ import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
+import java.net.InetSocketAddress;
import java.net.URI;
import java.text.DateFormat;
import java.util.*;
@@ -579,6 +583,22 @@ public class Mover {
}
}
+ private static void checkKeytabAndInit(Configuration conf)
+ throws IOException {
+ if (conf.getBoolean(DFSConfigKeys.DFS_MOVER_KEYTAB_ENABLED_KEY,
+ DFSConfigKeys.DFS_MOVER_KEYTAB_ENABLED_DEFAULT)) {
+ LOG.info("Keytab is configured, will login using keytab.");
+ UserGroupInformation.setConfiguration(conf);
+ String addr = conf.get(DFSConfigKeys.DFS_MOVER_ADDRESS_KEY,
+ DFSConfigKeys.DFS_MOVER_ADDRESS_DEFAULT);
+ InetSocketAddress socAddr = NetUtils.createSocketAddr(addr, 0,
+ DFSConfigKeys.DFS_MOVER_ADDRESS_KEY);
+ SecurityUtil.login(conf, DFSConfigKeys.DFS_MOVER_KEYTAB_FILE_KEY,
+ DFSConfigKeys.DFS_MOVER_KERBEROS_PRINCIPAL_KEY,
+ socAddr.getHostName());
+ }
+ }
+
static int run(Map<URI, List<Path>> namenodes, Configuration conf)
throws IOException, InterruptedException {
final long sleeptime =
@@ -588,7 +608,8 @@ public class Mover {
DFSConfigKeys.DFS_NAMENODE_REPLICATION_INTERVAL_DEFAULT) * 1000;
AtomicInteger retryCount = new AtomicInteger(0);
LOG.info("namenodes = " + namenodes);
-
+
+ checkKeytabAndInit(conf);
List<NameNodeConnector> connectors = Collections.emptyList();
try {
connectors = NameNodeConnector.newNameNodeConnectors(namenodes,
http://git-wip-us.apache.org/repos/asf/hadoop/blob/e806db71/hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml b/hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml
index ce880d3..b9a0812 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml
@@ -3712,6 +3712,46 @@
</property>
<property>
+ <name>dfs.mover.keytab.enabled</name>
+ <value>false</value>
+ <description>
+ Set to true to enable login using a keytab for Kerberized Hadoop.
+ </description>
+</property>
+
+<property>
+ <name>dfs.mover.address</name>
+ <value>0.0.0.0:0</value>
+ <description>
+ The hostname used for a keytab based Kerberos login. Keytab based login
+ can be enabled with dfs.mover.keytab.enabled.
+ </description>
+</property>
+
+<property>
+ <name>dfs.mover.keytab.file</name>
+ <value></value>
+ <description>
+ The keytab file used by the Mover to login as its
+ service principal. The principal name is configured with
+ dfs.mover.kerberos.principal. Keytab based login can be
+ enabled with dfs.mover.keytab.enabled.
+ </description>
+</property>
+
+<property>
+ <name>dfs.mover.kerberos.principal</name>
+ <value></value>
+ <description>
+ The Mover principal. This is typically set to
+ mover/_HOST@REALM.TLD. The Mover will substitute _HOST with its
+ own fully qualified hostname at startup. The _HOST placeholder
+ allows using the same configuration setting on different servers.
+ Keytab based login can be enabled with dfs.mover.keytab.enabled.
+ </description>
+</property>
+
+<property>
<name>dfs.namenode.audit.log.async</name>
<value>false</value>
<description>
http://git-wip-us.apache.org/repos/asf/hadoop/blob/e806db71/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/mover/TestMover.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/mover/TestMover.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/mover/TestMover.java
index f382243..b7b750b 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/mover/TestMover.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/mover/TestMover.java
@@ -17,21 +17,47 @@
*/
package org.apache.hadoop.hdfs.server.mover;
+import static org.apache.hadoop.fs.CommonConfigurationKeys.IPC_CLIENT_CONNECT_MAX_RETRIES_ON_SASL_KEY;
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_BLOCK_ACCESS_TOKEN_ENABLE_KEY;
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_CLIENT_HTTPS_KEYSTORE_RESOURCE_KEY;
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DATANODE_HTTPS_ADDRESS_KEY;
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DATANODE_KERBEROS_PRINCIPAL_KEY;
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DATANODE_KEYTAB_FILE_KEY;
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_HTTP_POLICY_KEY;
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_MOVER_ADDRESS_KEY;
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_MOVER_KERBEROS_PRINCIPAL_KEY;
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_MOVER_KEYTAB_ENABLED_KEY;
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_MOVER_KEYTAB_FILE_KEY;
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMENODE_HTTPS_ADDRESS_KEY;
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMENODE_KERBEROS_PRINCIPAL_KEY;
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMENODE_KEYTAB_FILE_KEY;
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_SERVER_HTTPS_KEYSTORE_RESOURCE_KEY;
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_WEB_AUTHENTICATION_KERBEROS_PRINCIPAL_KEY;
+import static org.apache.hadoop.hdfs.client.HdfsClientConfigKeys.DFS_DATA_TRANSFER_PROTECTION_KEY;
+
+import java.io.File;
import java.io.IOException;
import java.net.URI;
-import java.util.*;
+import java.security.PrivilegedExceptionAction;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
import java.util.concurrent.atomic.AtomicInteger;
-import com.google.common.collect.Maps;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.hdfs.DFSConfigKeys;
-import org.apache.hadoop.hdfs.DistributedFileSystem;
import org.apache.hadoop.hdfs.DFSTestUtil;
import org.apache.hadoop.hdfs.DFSUtil;
+import org.apache.hadoop.hdfs.DistributedFileSystem;
import org.apache.hadoop.hdfs.HdfsConfiguration;
import org.apache.hadoop.hdfs.MiniDFSCluster;
import org.apache.hadoop.hdfs.MiniDFSNNTopology;
@@ -47,14 +73,27 @@ import org.apache.hadoop.hdfs.server.balancer.NameNodeConnector;
import org.apache.hadoop.hdfs.server.balancer.TestBalancer;
import org.apache.hadoop.hdfs.server.mover.Mover.MLocation;
import org.apache.hadoop.hdfs.server.namenode.ha.HATestUtil;
+import org.apache.hadoop.http.HttpConfig;
+import org.apache.hadoop.minikdc.MiniKdc;
+import org.apache.hadoop.security.SecurityUtil;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.security.authentication.util.KerberosName;
+import org.apache.hadoop.security.ssl.KeyStoreTestUtil;
import org.apache.hadoop.test.GenericTestUtils;
import org.apache.hadoop.util.ToolRunner;
import org.junit.Assert;
import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
-public class TestMover {
+import com.google.common.base.Supplier;
+import com.google.common.collect.Maps;
- static final int DEFAULT_BLOCK_SIZE = 100;
+public class TestMover {
+ private static final Logger LOG = LoggerFactory.getLogger(TestMover.class);
+ private static final int DEFAULT_BLOCK_SIZE = 100;
+ private File keytabFile;
+ private String principal;
static {
TestBalancer.initTestSetup();
@@ -116,14 +155,11 @@ public class TestMover {
}
}
- @Test
- public void testScheduleBlockWithinSameNode() throws Exception {
- final Configuration conf = new HdfsConfiguration();
- initConf(conf);
+ private void testWithinSameNode(Configuration conf) throws Exception {
final MiniDFSCluster cluster = new MiniDFSCluster.Builder(conf)
.numDataNodes(3)
.storageTypes(
- new StorageType[] { StorageType.DISK, StorageType.ARCHIVE })
+ new StorageType[] {StorageType.DISK, StorageType.ARCHIVE})
.build();
try {
cluster.waitActive();
@@ -133,13 +169,11 @@ public class TestMover {
dfs.mkdirs(dir);
// write to DISK
dfs.setStoragePolicy(dir, "HOT");
- {
- final FSDataOutputStream out = dfs.create(new Path(file));
- out.writeChars("testScheduleWithinSameNode");
- out.close();
- }
+ final FSDataOutputStream out = dfs.create(new Path(file));
+ out.writeChars("testScheduleWithinSameNode");
+ out.close();
- //verify before movement
+ // verify before movement
LocatedBlock lb = dfs.getClient().getLocatedBlocks(file, 0).get(0);
StorageType[] storageTypes = lb.getStorageTypes();
for (StorageType storageType : storageTypes) {
@@ -148,21 +182,49 @@ public class TestMover {
// move to ARCHIVE
dfs.setStoragePolicy(dir, "COLD");
int rc = ToolRunner.run(conf, new Mover.Cli(),
- new String[] { "-p", dir.toString() });
- Assert.assertEquals("Movement to ARCHIVE should be successfull", 0, rc);
+ new String[] {"-p", dir.toString()});
+ Assert.assertEquals("Movement to ARCHIVE should be successful", 0, rc);
- // Wait till namenode notified
- Thread.sleep(3000);
- lb = dfs.getClient().getLocatedBlocks(file, 0).get(0);
- storageTypes = lb.getStorageTypes();
- for (StorageType storageType : storageTypes) {
- Assert.assertTrue(StorageType.ARCHIVE == storageType);
- }
+ // Wait till namenode notified about the block location details
+ waitForLocatedBlockWithArchiveStorageType(dfs, file, 3);
} finally {
cluster.shutdown();
}
}
+ private void waitForLocatedBlockWithArchiveStorageType(
+ final DistributedFileSystem dfs, final String file,
+ int expectedArchiveCount) throws Exception {
+ GenericTestUtils.waitFor(new Supplier<Boolean>() {
+ @Override
+ public Boolean get() {
+ LocatedBlock lb = null;
+ try {
+ lb = dfs.getClient().getLocatedBlocks(file, 0).get(0);
+ } catch (IOException e) {
+ LOG.error("Exception while getting located blocks", e);
+ return false;
+ }
+ int archiveCount = 0;
+ for (StorageType storageType : lb.getStorageTypes()) {
+ if (StorageType.ARCHIVE == storageType) {
+ archiveCount++;
+ }
+ }
+ LOG.info("Archive replica count, expected={} and actual={}",
+ expectedArchiveCount, archiveCount);
+ return expectedArchiveCount == archiveCount;
+ }
+ }, 100, 3000);
+ }
+
+ @Test
+ public void testScheduleBlockWithinSameNode() throws Exception {
+ final Configuration conf = new HdfsConfiguration();
+ initConf(conf);
+ testWithinSameNode(conf);
+ }
+
private void checkMovePaths(List<Path> actual, Path... expected) {
Assert.assertEquals(expected.length, actual.size());
for (Path p : expected) {
@@ -334,19 +396,10 @@ public class TestMover {
dfs.setStoragePolicy(new Path(file), "COLD");
int rc = ToolRunner.run(conf, new Mover.Cli(),
new String[] { "-p", file.toString() });
- Assert.assertEquals("Movement to ARCHIVE should be successfull", 0, rc);
+ Assert.assertEquals("Movement to ARCHIVE should be successful", 0, rc);
- // Wait till namenode notified
- Thread.sleep(3000);
- lb = dfs.getClient().getLocatedBlocks(file, 0).get(0);
- storageTypes = lb.getStorageTypes();
- int archiveCount = 0;
- for (StorageType storageType : storageTypes) {
- if (StorageType.ARCHIVE == storageType) {
- archiveCount++;
- }
- }
- Assert.assertEquals(archiveCount, 2);
+ // Wait till namenode notified about the block location details
+ waitForLocatedBlockWithArchiveStorageType(dfs, file, 2);
} finally {
cluster.shutdown();
}
@@ -514,7 +567,7 @@ public class TestMover {
// run Mover
int rc = ToolRunner.run(conf, new Mover.Cli(),
new String[] { "-p", barDir });
- Assert.assertEquals("Movement to ARCHIVE should be successfull", 0, rc);
+ Assert.assertEquals("Movement to ARCHIVE should be successful", 0, rc);
// verify storage types and locations
locatedBlocks = client.getBlockLocations(fooFile, 0, fileLen);
@@ -562,4 +615,87 @@ public class TestMover {
cluster.shutdown();
}
}
+
+ private void initSecureConf(Configuration conf) throws Exception {
+ String username = "mover";
+ File baseDir = GenericTestUtils.getTestDir(TestMover.class.getSimpleName());
+ FileUtil.fullyDelete(baseDir);
+ Assert.assertTrue(baseDir.mkdirs());
+
+ Properties kdcConf = MiniKdc.createConf();
+ MiniKdc kdc = new MiniKdc(kdcConf, baseDir);
+ kdc.start();
+
+ SecurityUtil.setAuthenticationMethod(
+ UserGroupInformation.AuthenticationMethod.KERBEROS, conf);
+ UserGroupInformation.setConfiguration(conf);
+ KerberosName.resetDefaultRealm();
+ Assert.assertTrue("Expected configuration to enable security",
+ UserGroupInformation.isSecurityEnabled());
+
+ keytabFile = new File(baseDir, username + ".keytab");
+ String keytab = keytabFile.getAbsolutePath();
+ // Windows will not reverse name lookup "127.0.0.1" to "localhost".
+ String krbInstance = Path.WINDOWS ? "127.0.0.1" : "localhost";
+ principal = username + "/" + krbInstance + "@" + kdc.getRealm();
+ String spnegoPrincipal = "HTTP/" + krbInstance + "@" + kdc.getRealm();
+ kdc.createPrincipal(keytabFile, username, username + "/" + krbInstance,
+ "HTTP/" + krbInstance);
+
+ conf.set(DFS_NAMENODE_KERBEROS_PRINCIPAL_KEY, principal);
+ conf.set(DFS_NAMENODE_KEYTAB_FILE_KEY, keytab);
+ conf.set(DFS_DATANODE_KERBEROS_PRINCIPAL_KEY, principal);
+ conf.set(DFS_DATANODE_KEYTAB_FILE_KEY, keytab);
+ conf.set(DFS_WEB_AUTHENTICATION_KERBEROS_PRINCIPAL_KEY, spnegoPrincipal);
+ conf.setBoolean(DFS_BLOCK_ACCESS_TOKEN_ENABLE_KEY, true);
+ conf.set(DFS_DATA_TRANSFER_PROTECTION_KEY, "authentication");
+ conf.set(DFS_HTTP_POLICY_KEY, HttpConfig.Policy.HTTPS_ONLY.name());
+ conf.set(DFS_NAMENODE_HTTPS_ADDRESS_KEY, "localhost:0");
+ conf.set(DFS_DATANODE_HTTPS_ADDRESS_KEY, "localhost:0");
+ conf.setInt(IPC_CLIENT_CONNECT_MAX_RETRIES_ON_SASL_KEY, 10);
+
+ conf.setBoolean(DFS_MOVER_KEYTAB_ENABLED_KEY, true);
+ conf.set(DFS_MOVER_ADDRESS_KEY, "localhost:0");
+ conf.set(DFS_MOVER_KEYTAB_FILE_KEY, keytab);
+ conf.set(DFS_MOVER_KERBEROS_PRINCIPAL_KEY, principal);
+
+ String keystoresDir = baseDir.getAbsolutePath();
+ String sslConfDir = KeyStoreTestUtil.getClasspathDir(TestMover.class);
+ KeyStoreTestUtil.setupSSLConfig(keystoresDir, sslConfDir, conf, false);
+
+ conf.set(DFS_CLIENT_HTTPS_KEYSTORE_RESOURCE_KEY,
+ KeyStoreTestUtil.getClientSSLConfigFileName());
+ conf.set(DFS_SERVER_HTTPS_KEYSTORE_RESOURCE_KEY,
+ KeyStoreTestUtil.getServerSSLConfigFileName());
+ initConf(conf);
+ }
+
+ /**
+ * Test Mover runs fine when logging in with a keytab in kerberized env.
+ * Reusing testWithinSameNode here for basic functionality testing.
+ */
+ @Test(timeout = 300000)
+ public void testMoverWithKeytabs() throws Exception {
+ final Configuration conf = new HdfsConfiguration();
+ try {
+ initSecureConf(conf);
+ final UserGroupInformation ugi = UserGroupInformation
+ .loginUserFromKeytabAndReturnUGI(principal,
+ keytabFile.getAbsolutePath());
+ ugi.doAs(new PrivilegedExceptionAction<Void>() {
+ @Override
+ public Void run() throws Exception {
+ // verify that mover runs Ok.
+ testWithinSameNode(conf);
+ // verify that UGI was logged in using keytab.
+ Assert.assertTrue(UserGroupInformation.isLoginKeytabBased());
+ return null;
+ }
+ });
+ } finally {
+ // Reset UGI so that other tests are not affected.
+ UserGroupInformation.reset();
+ UserGroupInformation.setConfiguration(new Configuration());
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
[6/8] hadoop git commit: YARN-4889. Changes in AMRMClient for
identifying resource-requests explicitly. (Arun Suresh via wangda)
Posted by aw...@apache.org.
YARN-4889. Changes in AMRMClient for identifying resource-requests explicitly. (Arun Suresh via wangda)
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/19c743c1
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/19c743c1
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/19c743c1
Branch: refs/heads/HADOOP-13341
Commit: 19c743c1bbcaf3df8f1d63e557143c960a538c42
Parents: b930dc3
Author: Wangda Tan <wa...@apache.org>
Authored: Fri Aug 26 16:48:00 2016 -0700
Committer: Wangda Tan <wa...@apache.org>
Committed: Fri Aug 26 17:14:12 2016 -0700
----------------------------------------------------------------------
.../yarn/api/records/ResourceRequest.java | 19 +-
.../hadoop/yarn/client/api/AMRMClient.java | 132 +++++++++++-
.../yarn/client/api/async/AMRMClientAsync.java | 18 ++
.../yarn/client/api/impl/AMRMClientImpl.java | 163 ++++++++++-----
.../client/api/impl/RemoteRequestsTable.java | 11 +-
.../yarn/client/api/impl/TestAMRMClient.java | 200 ++++++++++++++++---
.../impl/TestAMRMClientContainerRequest.java | 7 +-
.../api/impl/TestDistributedScheduling.java | 24 ++-
.../yarn/client/api/impl/TestNMClient.java | 6 +-
9 files changed, 468 insertions(+), 112 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/19c743c1/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/ResourceRequest.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/ResourceRequest.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/ResourceRequest.java
index 07f132c..2d6f0f4 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/ResourceRequest.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/ResourceRequest.java
@@ -112,6 +112,10 @@ public abstract class ResourceRequest implements Comparable<ResourceRequest> {
// Compare priority, host and capability
int ret = r1.getPriority().compareTo(r2.getPriority());
if (ret == 0) {
+ ret = Long.compare(
+ r1.getAllocationRequestId(), r2.getAllocationRequestId());
+ }
+ if (ret == 0) {
String h1 = r1.getResourceName();
String h2 = r2.getResourceName();
ret = h1.compareTo(h2);
@@ -381,6 +385,7 @@ public abstract class ResourceRequest implements Comparable<ResourceRequest> {
result = prime * result + ((hostName == null) ? 0 : hostName.hashCode());
result = prime * result + getNumContainers();
result = prime * result + ((priority == null) ? 0 : priority.hashCode());
+ result = prime * result + Long.valueOf(getAllocationRequestId()).hashCode();
return result;
}
@@ -422,6 +427,11 @@ public abstract class ResourceRequest implements Comparable<ResourceRequest> {
.equals(other.getExecutionTypeRequest().getExecutionType())) {
return false;
}
+
+ if (getAllocationRequestId() != other.getAllocationRequestId()) {
+ return false;
+ }
+
if (getNodeLabelExpression() == null) {
if (other.getNodeLabelExpression() != null) {
return false;
@@ -452,7 +462,14 @@ public abstract class ResourceRequest implements Comparable<ResourceRequest> {
int capabilityComparison =
this.getCapability().compareTo(other.getCapability());
if (capabilityComparison == 0) {
- return this.getNumContainers() - other.getNumContainers();
+ int numContainerComparison =
+ this.getNumContainers() - other.getNumContainers();
+ if (numContainerComparison == 0) {
+ return Long.compare(getAllocationRequestId(),
+ other.getAllocationRequestId());
+ } else {
+ return numContainerComparison;
+ }
} else {
return capabilityComparison;
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/19c743c1/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/AMRMClient.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/AMRMClient.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/AMRMClient.java
index 79d587a..2990c05 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/AMRMClient.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/AMRMClient.java
@@ -110,6 +110,7 @@ public abstract class AMRMClient<T extends AMRMClient.ContainerRequest> extends
final List<String> nodes;
final List<String> racks;
final Priority priority;
+ final long allocationRequestId;
final boolean relaxLocality;
final String nodeLabelsExpression;
final ExecutionTypeRequest executionTypeRequest;
@@ -134,6 +135,31 @@ public abstract class AMRMClient<T extends AMRMClient.ContainerRequest> extends
String[] racks, Priority priority) {
this(capability, nodes, racks, priority, true, null);
}
+
+ /**
+ * Instantiates a {@link ContainerRequest} with the given constraints and
+ * locality relaxation enabled.
+ *
+ * @param capability
+ * The {@link Resource} to be requested for each container.
+ * @param nodes
+ * Any hosts to request that the containers are placed on.
+ * @param racks
+ * Any racks to request that the containers are placed on. The
+ * racks corresponding to any hosts requested will be automatically
+ * added to this list.
+ * @param priority
+ * The priority at which to request the containers. Higher
+ * priorities have lower numerical values.
+ * @param allocationRequestId Allocation Request Id
+ */
+ @Public
+ @InterfaceStability.Evolving
+ public ContainerRequest(Resource capability, String[] nodes,
+ String[] racks, Priority priority, long allocationRequestId) {
+ this(capability, nodes, racks, priority, allocationRequestId, true, null,
+ ExecutionTypeRequest.newInstance());
+ }
/**
* Instantiates a {@link ContainerRequest} with the given constraints.
@@ -175,16 +201,79 @@ public abstract class AMRMClient<T extends AMRMClient.ContainerRequest> extends
* @param relaxLocality
* If true, containers for this request may be assigned on hosts
* and racks other than the ones explicitly requested.
+ * @param allocationRequestId Allocation Request Id
+ */
+ @Public
+ @InterfaceStability.Evolving
+ public ContainerRequest(Resource capability, String[] nodes,
+ String[] racks, Priority priority, long allocationRequestId,
+ boolean relaxLocality) {
+ this(capability, nodes, racks, priority, allocationRequestId,
+ relaxLocality, null, ExecutionTypeRequest.newInstance());
+ }
+
+ /**
+ * Instantiates a {@link ContainerRequest} with the given constraints.
+ *
+ * @param capability
+ * The {@link Resource} to be requested for each container.
+ * @param nodes
+ * Any hosts to request that the containers are placed on.
+ * @param racks
+ * Any racks to request that the containers are placed on. The
+ * racks corresponding to any hosts requested will be automatically
+ * added to this list.
+ * @param priority
+ * The priority at which to request the containers. Higher
+ * priorities have lower numerical values.
+ * @param relaxLocality
+ * If true, containers for this request may be assigned on hosts
+ * and racks other than the ones explicitly requested.
* @param nodeLabelsExpression
* Set node labels to allocate resource, now we only support
* asking for only a single node label
*/
public ContainerRequest(Resource capability, String[] nodes, String[] racks,
Priority priority, boolean relaxLocality, String nodeLabelsExpression) {
- this(capability, nodes, racks, priority, relaxLocality,
+ this(capability, nodes, racks, priority, 0, relaxLocality,
nodeLabelsExpression,
ExecutionTypeRequest.newInstance());
}
+
+ /**
+ * Instantiates a {@link ContainerRequest} with the given constraints.
+ *
+ * @param capability
+ * The {@link Resource} to be requested for each container.
+ * @param nodes
+ * Any hosts to request that the containers are placed on.
+ * @param racks
+ * Any racks to request that the containers are placed on. The
+ * racks corresponding to any hosts requested will be automatically
+ * added to this list.
+ * @param priority
+ * The priority at which to request the containers. Higher
+ * priorities have lower numerical values.
+ * @param allocationRequestId
+ * The allocationRequestId of the request. To be used as a tracking
+ * id to match Containers allocated against this request. Will
+ * default to 0 if not specified.
+ * @param relaxLocality
+ * If true, containers for this request may be assigned on hosts
+ * and racks other than the ones explicitly requested.
+ * @param nodeLabelsExpression
+ * Set node labels to allocate resource, now we only support
+ * asking for only a single node label
+ */
+ @Public
+ @InterfaceStability.Evolving
+ public ContainerRequest(Resource capability, String[] nodes, String[] racks,
+ Priority priority, long allocationRequestId, boolean relaxLocality,
+ String nodeLabelsExpression) {
+ this(capability, nodes, racks, priority, allocationRequestId,
+ relaxLocality, nodeLabelsExpression,
+ ExecutionTypeRequest.newInstance());
+ }
/**
* Instantiates a {@link ContainerRequest} with the given constraints.
@@ -200,6 +289,10 @@ public abstract class AMRMClient<T extends AMRMClient.ContainerRequest> extends
* @param priority
* The priority at which to request the containers. Higher
* priorities have lower numerical values.
+ * @param allocationRequestId
+ * The allocationRequestId of the request. To be used as a tracking
+ * id to match Containers allocated against this request. Will
+ * default to 0 if not specified.
* @param relaxLocality
* If true, containers for this request may be assigned on hosts
* and racks other than the ones explicitly requested.
@@ -210,7 +303,8 @@ public abstract class AMRMClient<T extends AMRMClient.ContainerRequest> extends
* Set the execution type of the container request.
*/
public ContainerRequest(Resource capability, String[] nodes, String[] racks,
- Priority priority, boolean relaxLocality, String nodeLabelsExpression,
+ Priority priority, long allocationRequestId, boolean relaxLocality,
+ String nodeLabelsExpression,
ExecutionTypeRequest executionTypeRequest) {
// Validate request
Preconditions.checkArgument(capability != null,
@@ -223,6 +317,7 @@ public abstract class AMRMClient<T extends AMRMClient.ContainerRequest> extends
&& (nodes == null || nodes.length == 0)),
"Can't turn off locality relaxation on a " +
"request with no location constraints");
+ this.allocationRequestId = allocationRequestId;
this.capability = capability;
this.nodes = (nodes != null ? ImmutableList.copyOf(nodes) : null);
this.racks = (racks != null ? ImmutableList.copyOf(racks) : null);
@@ -247,6 +342,10 @@ public abstract class AMRMClient<T extends AMRMClient.ContainerRequest> extends
public Priority getPriority() {
return priority;
}
+
+ public long getAllocationRequestId() {
+ return allocationRequestId;
+ }
public boolean getRelaxLocality() {
return relaxLocality;
@@ -264,6 +363,7 @@ public abstract class AMRMClient<T extends AMRMClient.ContainerRequest> extends
StringBuilder sb = new StringBuilder();
sb.append("Capability[").append(capability).append("]");
sb.append("Priority[").append(priority).append("]");
+ sb.append("AllocationRequestId[").append(allocationRequestId).append("]");
sb.append("ExecutionTypeRequest[").append(executionTypeRequest)
.append("]");
return sb.toString();
@@ -390,6 +490,10 @@ public abstract class AMRMClient<T extends AMRMClient.ContainerRequest> extends
* Each collection in the list contains requests with identical
* <code>Resource</code> size that fit in the given capability. In a
* collection, requests will be returned in the same order as they were added.
+ *
+ * NOTE: This API only matches Container requests that were created by the
+ * client WITHOUT the allocationRequestId being set.
+ *
* @return Collection of request matching the parameters
*/
@InterfaceStability.Evolving
@@ -407,7 +511,11 @@ public abstract class AMRMClient<T extends AMRMClient.ContainerRequest> extends
* Each collection in the list contains requests with identical
* <code>Resource</code> size that fit in the given capability. In a
* collection, requests will be returned in the same order as they were added.
- * specify an <code>ExecutionType</code> .
+ * specify an <code>ExecutionType</code>.
+ *
+ * NOTE: This API only matches Container requests that were created by the
+ * client WITHOUT the allocationRequestId being set.
+ *
* @param priority Priority
* @param resourceName Location
* @param executionType ExecutionType
@@ -421,7 +529,23 @@ public abstract class AMRMClient<T extends AMRMClient.ContainerRequest> extends
throw new UnsupportedOperationException("The sub-class extending" +
" AMRMClient is expected to implement this !!");
}
-
+
+ /**
+ * Get outstanding <code>ContainerRequest</code>s matching the given
+ * allocationRequestId. These ContainerRequests should have been added via
+ * <code>addContainerRequest</code> earlier in the lifecycle. For performance,
+ * the AMRMClient may return its internal collection directly without creating
+ * a copy. Users should not perform mutable operations on the return value.
+ *
+ * NOTE: This API only matches Container requests that were created by the
+ * client WITH the allocationRequestId being set to a non-default value.
+ *
+ * @param allocationRequestId Allocation Request Id
+ * @return Collection of request matching the parameters
+ */
+ @InterfaceStability.Evolving
+ public abstract Collection<T> getMatchingRequests(long allocationRequestId);
+
/**
* Update application's blacklist with addition or removal resources.
*
http://git-wip-us.apache.org/repos/asf/hadoop/blob/19c743c1/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/async/AMRMClientAsync.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/async/AMRMClientAsync.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/async/AMRMClientAsync.java
index d06fadc..10d2a2f 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/async/AMRMClientAsync.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/async/AMRMClientAsync.java
@@ -202,6 +202,10 @@ extends AbstractService {
/**
* Returns all matching ContainerRequests that match the given Priority,
* ResourceName, ExecutionType and Capability.
+ *
+ * NOTE: This matches only requests that were made by the client WITHOUT the
+ * allocationRequestId specified.
+ *
* @param priority Priority.
* @param resourceName Location.
* @param executionType ExecutionType.
@@ -214,6 +218,20 @@ extends AbstractService {
return client.getMatchingRequests(priority, resourceName,
executionType, capability);
}
+
+ /**
+ * Returns all matching ContainerRequests that match the given
+ * AllocationRequestId.
+ *
+ * NOTE: This matches only requests that were made by the client WITH the
+ * allocationRequestId specified.
+ *
+ * @param allocationRequestId AllocationRequestId.
+ * @return All matching ContainerRequests
+ */
+ public Collection<T> getMatchingRequests(long allocationRequestId) {
+ return client.getMatchingRequests(allocationRequestId);
+ }
/**
* Registers this application master with the resource manager. On successful
http://git-wip-us.apache.org/repos/asf/hadoop/blob/19c743c1/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/impl/AMRMClientImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/impl/AMRMClientImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/impl/AMRMClientImpl.java
index 4145944..60834f6 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/impl/AMRMClientImpl.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/impl/AMRMClientImpl.java
@@ -108,10 +108,11 @@ public class AMRMClientImpl<T extends ContainerRequest> extends AMRMClient<T> {
ResourceRequest remoteRequest;
LinkedHashSet<T> containerRequests;
- ResourceRequestInfo(Priority priority, String resourceName,
- Resource capability, boolean relaxLocality) {
+ ResourceRequestInfo(Long allocationRequestId, Priority priority,
+ String resourceName, Resource capability, boolean relaxLocality) {
remoteRequest = ResourceRequest.newInstance(priority, resourceName,
capability, 0);
+ remoteRequest.setAllocationRequestId(allocationRequestId);
remoteRequest.setRelaxLocality(relaxLocality);
containerRequests = new LinkedHashSet<T>();
}
@@ -154,7 +155,8 @@ public class AMRMClientImpl<T extends ContainerRequest> extends AMRMClient<T> {
return (mem0 <= mem1 && cpu0 <= cpu1);
}
- final RemoteRequestsTable remoteRequestsTable = new RemoteRequestsTable<T>();
+ private final Map<Long, RemoteRequestsTable<T>> remoteRequests =
+ new HashMap<>();
protected final Set<ResourceRequest> ask = new TreeSet<ResourceRequest>(
new org.apache.hadoop.yarn.api.records.ResourceRequest.ResourceRequestComparator());
@@ -263,10 +265,12 @@ public class AMRMClientImpl<T extends ContainerRequest> extends AMRMClient<T> {
for(ResourceRequest r : ask) {
// create a copy of ResourceRequest as we might change it while the
// RPC layer is using it to send info across
- askList.add(ResourceRequest.newInstance(r.getPriority(),
+ ResourceRequest rr = ResourceRequest.newInstance(r.getPriority(),
r.getResourceName(), r.getCapability(), r.getNumContainers(),
r.getRelaxLocality(), r.getNodeLabelExpression(),
- r.getExecutionTypeRequest()));
+ r.getExecutionTypeRequest());
+ rr.setAllocationRequestId(r.getAllocationRequestId());
+ askList.add(rr);
}
List<ContainerResourceChangeRequest> increaseList = new ArrayList<>();
List<ContainerResourceChangeRequest> decreaseList = new ArrayList<>();
@@ -318,11 +322,14 @@ public class AMRMClientImpl<T extends ContainerRequest> extends AMRMClient<T> {
synchronized (this) {
release.addAll(this.pendingRelease);
blacklistAdditions.addAll(this.blacklistedNodes);
- @SuppressWarnings("unchecked")
- Iterator<ResourceRequestInfo<T>> reqIter =
- remoteRequestsTable.iterator();
- while (reqIter.hasNext()) {
- addResourceRequestToAsk(reqIter.next().remoteRequest);
+ for (RemoteRequestsTable remoteRequestsTable :
+ remoteRequests.values()) {
+ @SuppressWarnings("unchecked")
+ Iterator<ResourceRequestInfo<T>> reqIter =
+ remoteRequestsTable.iterator();
+ while (reqIter.hasNext()) {
+ addResourceRequestToAsk(reqIter.next().remoteRequest);
+ }
}
change.putAll(this.pendingChange);
}
@@ -498,15 +505,16 @@ public class AMRMClientImpl<T extends ContainerRequest> extends AMRMClient<T> {
// check that specific and non-specific requests cannot be mixed within a
// priority
- checkLocalityRelaxationConflict(req.getPriority(), ANY_LIST,
- req.getRelaxLocality());
+ checkLocalityRelaxationConflict(req.getAllocationRequestId(),
+ req.getPriority(), ANY_LIST, req.getRelaxLocality());
// check that specific rack cannot be mixed with specific node within a
// priority. If node and its rack are both specified then they must be
// in the same request.
// For explicitly requested racks, we set locality relaxation to true
- checkLocalityRelaxationConflict(req.getPriority(), dedupedRacks, true);
- checkLocalityRelaxationConflict(req.getPriority(), inferredRacks,
- req.getRelaxLocality());
+ checkLocalityRelaxationConflict(req.getAllocationRequestId(),
+ req.getPriority(), dedupedRacks, true);
+ checkLocalityRelaxationConflict(req.getAllocationRequestId(),
+ req.getPriority(), inferredRacks, req.getRelaxLocality());
// check if the node label expression specified is valid
checkNodeLabelExpression(req);
@@ -607,6 +615,24 @@ public class AMRMClientImpl<T extends ContainerRequest> extends AMRMClient<T> {
return clusterNodeCount;
}
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public Collection<T> getMatchingRequests(long allocationRequestId) {
+ RemoteRequestsTable remoteRequestsTable = getTable(allocationRequestId);
+ LinkedHashSet<T> list = new LinkedHashSet<>();
+
+ if (remoteRequestsTable != null) {
+ Iterator<ResourceRequestInfo<T>> reqIter =
+ remoteRequestsTable.iterator();
+ while (reqIter.hasNext()) {
+ ResourceRequestInfo<T> resReqInfo = reqIter.next();
+ list.addAll(resReqInfo.containerRequests);
+ }
+ }
+ return list;
+ }
+
@Override
public synchronized List<? extends Collection<T>> getMatchingRequests(
Priority priority,
@@ -617,6 +643,7 @@ public class AMRMClientImpl<T extends ContainerRequest> extends AMRMClient<T> {
}
@Override
+ @SuppressWarnings("unchecked")
public synchronized List<? extends Collection<T>> getMatchingRequests(
Priority priority, String resourceName, ExecutionType executionType,
Resource capability) {
@@ -626,9 +653,9 @@ public class AMRMClientImpl<T extends ContainerRequest> extends AMRMClient<T> {
"The priority at which to request containers should not be null ");
List<LinkedHashSet<T>> list = new LinkedList<LinkedHashSet<T>>();
- @SuppressWarnings("unchecked")
+ RemoteRequestsTable remoteRequestsTable = getTable(0);
List<ResourceRequestInfo<T>> matchingRequests =
- this.remoteRequestsTable.getMatchingRequests(priority, resourceName,
+ remoteRequestsTable.getMatchingRequests(priority, resourceName,
executionType, capability);
// If no exact match. Container may be larger than what was requested.
// get all resources <= capability. map is reverse sorted.
@@ -664,23 +691,26 @@ public class AMRMClientImpl<T extends ContainerRequest> extends AMRMClient<T> {
* ContainerRequests with locality relaxation cannot be made at the same
* priority as ContainerRequests without locality relaxation.
*/
- private void checkLocalityRelaxationConflict(Priority priority,
- Collection<String> locations, boolean relaxLocality) {
+ private void checkLocalityRelaxationConflict(Long allocationReqId,
+ Priority priority, Collection<String> locations, boolean relaxLocality) {
// Locality relaxation will be set to relaxLocality for all implicitly
// requested racks. Make sure that existing rack requests match this.
- @SuppressWarnings("unchecked")
- List<ResourceRequestInfo> allCapabilityMaps =
- remoteRequestsTable.getAllResourceRequestInfos(priority, locations);
- for (ResourceRequestInfo reqs : allCapabilityMaps) {
- ResourceRequest remoteRequest = reqs.remoteRequest;
- boolean existingRelaxLocality = remoteRequest.getRelaxLocality();
- if (relaxLocality != existingRelaxLocality) {
- throw new InvalidContainerRequestException("Cannot submit a "
- + "ContainerRequest asking for location "
- + remoteRequest.getResourceName() + " with locality relaxation "
- + relaxLocality + " when it has already been requested"
- + "with locality relaxation " + existingRelaxLocality);
+ RemoteRequestsTable<T> remoteRequestsTable = getTable(allocationReqId);
+ if (remoteRequestsTable != null) {
+ @SuppressWarnings("unchecked")
+ List<ResourceRequestInfo> allCapabilityMaps =
+ remoteRequestsTable.getAllResourceRequestInfos(priority, locations);
+ for (ResourceRequestInfo reqs : allCapabilityMaps) {
+ ResourceRequest remoteRequest = reqs.remoteRequest;
+ boolean existingRelaxLocality = remoteRequest.getRelaxLocality();
+ if (relaxLocality != existingRelaxLocality) {
+ throw new InvalidContainerRequestException("Cannot submit a "
+ + "ContainerRequest asking for location "
+ + remoteRequest.getResourceName() + " with locality relaxation "
+ + relaxLocality + " when it has already been requested"
+ + "with locality relaxation " + existingRelaxLocality);
+ }
}
}
}
@@ -742,10 +772,17 @@ public class AMRMClientImpl<T extends ContainerRequest> extends AMRMClient<T> {
private void addResourceRequest(Priority priority, String resourceName,
ExecutionTypeRequest execTypeReq, Resource capability, T req,
boolean relaxLocality, String labelExpression) {
+ RemoteRequestsTable<T> remoteRequestsTable =
+ getTable(req.getAllocationRequestId());
+ if (remoteRequestsTable == null) {
+ remoteRequestsTable = new RemoteRequestsTable<T>();
+ putTable(req.getAllocationRequestId(), remoteRequestsTable);
+ }
@SuppressWarnings("unchecked")
ResourceRequestInfo resourceRequestInfo = remoteRequestsTable
- .addResourceRequest(priority, resourceName,
- execTypeReq, capability, req, relaxLocality, labelExpression);
+ .addResourceRequest(req.getAllocationRequestId(), priority,
+ resourceName, execTypeReq, capability, req, relaxLocality,
+ labelExpression);
// Note this down for next interaction with ResourceManager
addResourceRequestToAsk(resourceRequestInfo.remoteRequest);
@@ -761,29 +798,37 @@ public class AMRMClientImpl<T extends ContainerRequest> extends AMRMClient<T> {
private void decResourceRequest(Priority priority, String resourceName,
ExecutionTypeRequest execTypeReq, Resource capability, T req) {
- @SuppressWarnings("unchecked")
- ResourceRequestInfo resourceRequestInfo =
- remoteRequestsTable.decResourceRequest(priority, resourceName,
- execTypeReq, capability, req);
- // send the ResourceRequest to RM even if is 0 because it needs to override
- // a previously sent value. If ResourceRequest was not sent previously then
- // sending 0 aught to be a no-op on RM
- if (resourceRequestInfo != null) {
- addResourceRequestToAsk(resourceRequestInfo.remoteRequest);
-
- // delete entry from map if no longer needed
- if (resourceRequestInfo.remoteRequest.getNumContainers() == 0) {
- this.remoteRequestsTable.remove(priority, resourceName,
- execTypeReq.getExecutionType(), capability);
- }
+ RemoteRequestsTable<T> remoteRequestsTable =
+ getTable(req.getAllocationRequestId());
+ if (remoteRequestsTable != null) {
+ @SuppressWarnings("unchecked")
+ ResourceRequestInfo resourceRequestInfo =
+ remoteRequestsTable.decResourceRequest(priority, resourceName,
+ execTypeReq, capability, req);
+ // send the ResourceRequest to RM even if is 0 because it needs to
+ // override a previously sent value. If ResourceRequest was not sent
+ // previously then sending 0 ought to be a no-op on RM
+ if (resourceRequestInfo != null) {
+ addResourceRequestToAsk(resourceRequestInfo.remoteRequest);
+
+ // delete entry from map if no longer needed
+ if (resourceRequestInfo.remoteRequest.getNumContainers() == 0) {
+ remoteRequestsTable.remove(priority, resourceName,
+ execTypeReq.getExecutionType(), capability);
+ }
- if (LOG.isDebugEnabled()) {
- LOG.debug("AFTER decResourceRequest:" + " applicationId="
- + " priority=" + priority.getPriority()
- + " resourceName=" + resourceName + " numContainers="
- + resourceRequestInfo.remoteRequest.getNumContainers()
- + " #asks=" + ask.size());
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("AFTER decResourceRequest:"
+ + " allocationRequestId=" + req.getAllocationRequestId()
+ + " priority=" + priority.getPriority()
+ + " resourceName=" + resourceName + " numContainers="
+ + resourceRequestInfo.remoteRequest.getNumContainers()
+ + " #asks=" + ask.size());
+ }
}
+ } else {
+ LOG.info("No remoteRequestTable found with allocationRequestId="
+ + req.getAllocationRequestId());
}
}
@@ -829,4 +874,14 @@ public class AMRMClientImpl<T extends ContainerRequest> extends AMRMClient<T> {
currentUGI.addToken(amrmToken);
amrmToken.setService(ClientRMProxy.getAMRMTokenService(getConfig()));
}
+
+ @VisibleForTesting
+ RemoteRequestsTable<T> getTable(long allocationRequestId) {
+ return remoteRequests.get(Long.valueOf(allocationRequestId));
+ }
+
+ RemoteRequestsTable<T> putTable(long allocationRequestId,
+ RemoteRequestsTable<T> table) {
+ return remoteRequests.put(Long.valueOf(allocationRequestId), table);
+ }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/19c743c1/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/impl/RemoteRequestsTable.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/impl/RemoteRequestsTable.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/impl/RemoteRequestsTable.java
index 853a512..110ca79 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/impl/RemoteRequestsTable.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/impl/RemoteRequestsTable.java
@@ -264,15 +264,16 @@ class RemoteRequestsTable<T> implements Iterable<ResourceRequestInfo>{
}
@SuppressWarnings("unchecked")
- ResourceRequestInfo addResourceRequest(Priority priority, String resourceName,
- ExecutionTypeRequest execTypeReq, Resource capability, T req,
- boolean relaxLocality, String labelExpression) {
+ ResourceRequestInfo addResourceRequest(Long allocationRequestId,
+ Priority priority, String resourceName, ExecutionTypeRequest execTypeReq,
+ Resource capability, T req, boolean relaxLocality,
+ String labelExpression) {
ResourceRequestInfo resourceRequestInfo = get(priority, resourceName,
execTypeReq.getExecutionType(), capability);
if (resourceRequestInfo == null) {
resourceRequestInfo =
- new ResourceRequestInfo(priority, resourceName, capability,
- relaxLocality);
+ new ResourceRequestInfo(allocationRequestId, priority, resourceName,
+ capability, relaxLocality);
put(priority, resourceName, execTypeReq.getExecutionType(), capability,
resourceRequestInfo);
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/19c743c1/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestAMRMClient.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestAMRMClient.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestAMRMClient.java
index 1eeeb78..e0ad2c4 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestAMRMClient.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestAMRMClient.java
@@ -33,6 +33,7 @@ import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Set;
@@ -268,6 +269,18 @@ public class TestAMRMClient {
amClient.addContainerRequest(storedContainer5);
amClient.addContainerRequest(storedContainer6);
amClient.addContainerRequest(storedContainer7);
+
+ // Add some CRs with allocReqIds... These will not be returned by
+ // the default getMatchingRequests
+ ContainerRequest storedContainer11 =
+ new ContainerRequest(capability1, nodes, racks, priority, 1);
+ ContainerRequest storedContainer33 =
+ new ContainerRequest(capability3, nodes, racks, priority, 3);
+ ContainerRequest storedContainer43 =
+ new ContainerRequest(capability4, nodes, racks, priority, 3);
+ amClient.addContainerRequest(storedContainer11);
+ amClient.addContainerRequest(storedContainer33);
+ amClient.addContainerRequest(storedContainer43);
// test matching of containers
List<? extends Collection<ContainerRequest>> matches;
@@ -279,6 +292,25 @@ public class TestAMRMClient {
storedRequest = matches.get(0).iterator().next();
assertEquals(storedContainer1, storedRequest);
amClient.removeContainerRequest(storedContainer1);
+
+ // exact match for allocReqId 1
+ Collection<ContainerRequest> reqIdMatches =
+ amClient.getMatchingRequests(1);
+ assertEquals(1, reqIdMatches.size());
+ storedRequest = reqIdMatches.iterator().next();
+ assertEquals(storedContainer11, storedRequest);
+ amClient.removeContainerRequest(storedContainer11);
+
+ // exact match for allocReqId 3
+ reqIdMatches = amClient.getMatchingRequests(3);
+ assertEquals(2, reqIdMatches.size());
+ Iterator<ContainerRequest> iter = reqIdMatches.iterator();
+ storedRequest = iter.next();
+ assertEquals(storedContainer43, storedRequest);
+ amClient.removeContainerRequest(storedContainer43);
+ storedRequest = iter.next();
+ assertEquals(storedContainer33, storedRequest);
+ amClient.removeContainerRequest(storedContainer33);
// exact matching with order maintained
Resource testCapability2 = Resource.newInstance(2000, 1);
@@ -364,26 +396,32 @@ public class TestAMRMClient {
ContainerRequest storedGuarContainer2 =
new ContainerRequest(capability2, nodes, racks, priority);
ContainerRequest storedOpportContainer1 =
- new ContainerRequest(capability1, nodes, racks, priority, true, null,
+ new ContainerRequest(capability1, nodes, racks, priority,
+ 0, true, null,
ExecutionTypeRequest.newInstance(ExecutionType.OPPORTUNISTIC));
ContainerRequest storedOpportContainer2 =
- new ContainerRequest(capability2, nodes, racks, priority, true, null,
+ new ContainerRequest(capability2, nodes, racks, priority,
+ 0, true, null,
ExecutionTypeRequest.newInstance(ExecutionType.OPPORTUNISTIC));
ContainerRequest storedOpportContainer3 =
- new ContainerRequest(capability3, nodes, racks, priority, true, null,
+ new ContainerRequest(capability3, nodes, racks, priority,
+ 0, true, null,
ExecutionTypeRequest.newInstance(ExecutionType.OPPORTUNISTIC));
ContainerRequest storedOpportContainer4 =
- new ContainerRequest(capability4, nodes, racks, priority, true, null,
+ new ContainerRequest(capability4, nodes, racks, priority,
+ 0, true, null,
ExecutionTypeRequest.newInstance(ExecutionType.OPPORTUNISTIC));
ContainerRequest storedOpportContainer5 =
- new ContainerRequest(capability5, nodes, racks, priority, true, null,
+ new ContainerRequest(capability5, nodes, racks, priority,
+ 0, true, null,
ExecutionTypeRequest.newInstance(ExecutionType.OPPORTUNISTIC));
ContainerRequest storedOpportContainer6 =
- new ContainerRequest(capability6, nodes, racks, priority, true, null,
+ new ContainerRequest(capability6, nodes, racks, priority,
+ 0, true, null,
ExecutionTypeRequest.newInstance(ExecutionType.OPPORTUNISTIC));
ContainerRequest storedOpportContainer7 =
new ContainerRequest(capability7, nodes, racks, priority2,
- false, null,
+ 0, false, null,
ExecutionTypeRequest.newInstance(ExecutionType.OPPORTUNISTIC));
amClient.addContainerRequest(storedGuarContainer1);
amClient.addContainerRequest(storedGuarContainer2);
@@ -541,11 +579,13 @@ public class TestAMRMClient {
amClient.addContainerRequest(storedContainer3);
// test addition and storage
- int containersRequestedAny = amClient.remoteRequestsTable.get(priority,
+ RemoteRequestsTable<ContainerRequest> remoteRequestsTable =
+ amClient.getTable(0);
+ int containersRequestedAny = remoteRequestsTable.get(priority,
ResourceRequest.ANY, ExecutionType.GUARANTEED, capability)
.remoteRequest.getNumContainers();
assertEquals(2, containersRequestedAny);
- containersRequestedAny = amClient.remoteRequestsTable.get(priority1,
+ containersRequestedAny = remoteRequestsTable.get(priority1,
ResourceRequest.ANY, ExecutionType.GUARANTEED, capability)
.remoteRequest.getNumContainers();
assertEquals(1, containersRequestedAny);
@@ -584,7 +624,7 @@ public class TestAMRMClient {
amClient.getMatchingRequests(priority1, ResourceRequest.ANY, capability);
assertTrue(matches.isEmpty());
// 0 requests left. everything got cleaned up
- assertTrue(amClient.remoteRequestsTable.isEmpty());
+ assertTrue(amClient.getTable(0).isEmpty());
// go through an exemplary allocation, matching and release cycle
amClient.addContainerRequest(storedContainer1);
@@ -628,7 +668,7 @@ public class TestAMRMClient {
assertEquals(0, amClient.ask.size());
assertEquals(0, allocResponse.getAllocatedContainers().size());
// 0 requests left. everything got cleaned up
- assertTrue(amClient.remoteRequestsTable.isEmpty());
+ assertTrue(remoteRequestsTable.isEmpty());
amClient.unregisterApplicationMaster(FinalApplicationStatus.SUCCEEDED,
null, null);
@@ -780,6 +820,16 @@ public class TestAMRMClient {
@Test (timeout=60000)
public void testAMRMClient() throws YarnException, IOException {
+ initAMRMClientAndTest(false);
+ }
+
+ @Test (timeout=60000)
+ public void testAMRMClientAllocReqId() throws YarnException, IOException {
+ initAMRMClientAndTest(true);
+ }
+
+ private void initAMRMClientAndTest(boolean useAllocReqId)
+ throws YarnException, IOException {
AMRMClient<ContainerRequest> amClient = null;
try {
// start am rm client
@@ -796,7 +846,11 @@ public class TestAMRMClient {
amClient.registerApplicationMaster("Host", 10000, "");
- testAllocation((AMRMClientImpl<ContainerRequest>)amClient);
+ if (useAllocReqId) {
+ testAllocRequestId((AMRMClientImpl<ContainerRequest>)amClient);
+ } else {
+ testAllocation((AMRMClientImpl<ContainerRequest>) amClient);
+ }
amClient.unregisterApplicationMaster(FinalApplicationStatus.SUCCEEDED,
null, null);
@@ -1055,22 +1109,9 @@ public class TestAMRMClient {
new ContainerRequest(capability, nodes, racks, priority));
amClient.removeContainerRequest(
new ContainerRequest(capability, nodes, racks, priority));
-
- int containersRequestedNode = amClient.remoteRequestsTable.get(priority,
- node, ExecutionType.GUARANTEED, capability).remoteRequest
- .getNumContainers();
- int containersRequestedRack = amClient.remoteRequestsTable.get(priority,
- rack, ExecutionType.GUARANTEED, capability).remoteRequest
- .getNumContainers();
- int containersRequestedAny = amClient.remoteRequestsTable.get(priority,
- ResourceRequest.ANY, ExecutionType.GUARANTEED, capability)
- .remoteRequest.getNumContainers();
- assertEquals(2, containersRequestedNode);
- assertEquals(2, containersRequestedRack);
- assertEquals(2, containersRequestedAny);
- assertEquals(3, amClient.ask.size());
- assertEquals(0, amClient.release.size());
+ assertNumContainers(amClient, 0, 2, 2, 2, 3, 0);
+ int containersRequestedAny = 2;
// RM should allocate container within 2 calls to allocate()
int allocatedContainerCount = 0;
@@ -1163,10 +1204,15 @@ public class TestAMRMClient {
// verify that the remove request made in between makeRequest and allocate
// has not been lost
assertEquals(0, snoopRequest.getNumContainers());
-
- iterationsLeft = 3;
+
+ waitForContainerCompletion(3, amClient, releases);
+ }
+
+ private void waitForContainerCompletion(int numIterations,
+ AMRMClientImpl<ContainerRequest> amClient, Set<ContainerId> releases)
+ throws YarnException, IOException {
// do a few iterations to ensure RM is not going send new containers
- while(!releases.isEmpty() || iterationsLeft-- > 0) {
+ while(!releases.isEmpty() || numIterations-- > 0) {
// inform RM of rejection
AllocateResponse allocResponse = amClient.allocate(0.1f);
// RM did not send new containers because AM does not need any
@@ -1181,7 +1227,7 @@ public class TestAMRMClient {
}
}
}
- if(iterationsLeft > 0) {
+ if(numIterations > 0) {
// sleep to make sure NM's heartbeat
sleep(100);
}
@@ -1190,6 +1236,98 @@ public class TestAMRMClient {
assertEquals(0, amClient.release.size());
}
+ private void testAllocRequestId(
+ final AMRMClientImpl<ContainerRequest> amClient) throws YarnException,
+ IOException {
+ // setup container request
+
+ assertEquals(0, amClient.ask.size());
+ assertEquals(0, amClient.release.size());
+
+ amClient.addContainerRequest(
+ new ContainerRequest(capability, nodes, racks, priority));
+ amClient.addContainerRequest(
+ new ContainerRequest(capability, nodes, racks, priority));
+ amClient.addContainerRequest(
+ new ContainerRequest(capability, nodes, racks, priority, 1));
+ amClient.addContainerRequest(
+ new ContainerRequest(capability, nodes, racks, priority, 1));
+ amClient.addContainerRequest(
+ new ContainerRequest(capability, nodes, racks, priority, 2));
+ amClient.addContainerRequest(
+ new ContainerRequest(capability, nodes, racks, priority, 2));
+ amClient.removeContainerRequest(
+ new ContainerRequest(capability, nodes, racks, priority));
+ amClient.removeContainerRequest(
+ new ContainerRequest(capability, nodes, racks, priority, 1));
+ amClient.removeContainerRequest(
+ new ContainerRequest(capability, nodes, racks, priority, 2));
+
+ assertNumContainers(amClient, 0, 1, 1, 1, 9, 0);
+ assertNumContainers(amClient, 1, 1, 1, 1, 9, 0);
+ assertNumContainers(amClient, 2, 1, 1, 1, 9, 0);
+ int containersRequestedAny = 3;
+
+ // RM should allocate container within 2 calls to allocate()
+ List<Container> allocatedContainers = new ArrayList<>();
+ int iterationsLeft = 5;
+ Set<ContainerId> releases = new TreeSet<ContainerId>();
+
+ while (allocatedContainers.size() < containersRequestedAny
+ && iterationsLeft-- > 0) {
+ AllocateResponse allocResponse = amClient.allocate(0.1f);
+ assertEquals(0, amClient.ask.size());
+ assertEquals(0, amClient.release.size());
+
+ allocatedContainers.addAll(allocResponse.getAllocatedContainers());
+ for(Container container : allocResponse.getAllocatedContainers()) {
+ ContainerId rejectContainerId = container.getId();
+ releases.add(rejectContainerId);
+ amClient.releaseAssignedContainer(rejectContainerId);
+ }
+
+ if(allocatedContainers.size() < containersRequestedAny) {
+ // sleep to let NM's heartbeat to RM and trigger allocations
+ sleep(100);
+ }
+ }
+
+ assertEquals(containersRequestedAny, allocatedContainers.size());
+ Set<Long> expAllocIds = new HashSet<>(
+ Arrays.asList(Long.valueOf(0), Long.valueOf(1), Long.valueOf(2)));
+ Set<Long> actAllocIds = new HashSet<>();
+ for (Container ac : allocatedContainers) {
+ actAllocIds.add(Long.valueOf(ac.getAllocationRequestId()));
+ }
+ assertEquals(expAllocIds, actAllocIds);
+ assertEquals(3, amClient.release.size());
+ assertEquals(0, amClient.ask.size());
+
+ waitForContainerCompletion(3, amClient, releases);
+ }
+
+ private void assertNumContainers(AMRMClientImpl<ContainerRequest> amClient,
+ long allocationReqId, int expNode, int expRack, int expAny,
+ int expAsks, int expRelease) {
+ RemoteRequestsTable<ContainerRequest> remoteRequestsTable =
+ amClient.getTable(allocationReqId);
+ int containersRequestedNode = remoteRequestsTable.get(priority,
+ node, ExecutionType.GUARANTEED, capability).remoteRequest
+ .getNumContainers();
+ int containersRequestedRack = remoteRequestsTable.get(priority,
+ rack, ExecutionType.GUARANTEED, capability).remoteRequest
+ .getNumContainers();
+ int containersRequestedAny = remoteRequestsTable.get(priority,
+ ResourceRequest.ANY, ExecutionType.GUARANTEED, capability)
+ .remoteRequest.getNumContainers();
+
+ assertEquals(expNode, containersRequestedNode);
+ assertEquals(expRack, containersRequestedRack);
+ assertEquals(expAny, containersRequestedAny);
+ assertEquals(expAsks, amClient.ask.size());
+ assertEquals(expRelease, amClient.release.size());
+ }
+
class CountDownSupplier implements Supplier<Boolean> {
int counter = 0;
@Override
http://git-wip-us.apache.org/repos/asf/hadoop/blob/19c743c1/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestAMRMClientContainerRequest.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestAMRMClientContainerRequest.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestAMRMClientContainerRequest.java
index 2db33c1..ad18da3 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestAMRMClientContainerRequest.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestAMRMClientContainerRequest.java
@@ -61,7 +61,7 @@ public class TestAMRMClientContainerRequest {
verifyResourceRequest(client, request, ResourceRequest.ANY, true);
ContainerRequest request2 =
new ContainerRequest(capability, new String[] {"host1", "host2"},
- new String[] {"/rack2"}, Priority.newInstance(1), true, null,
+ new String[] {"/rack2"}, Priority.newInstance(1), 0, true, null,
ExecutionTypeRequest.newInstance(
ExecutionType.OPPORTUNISTIC, true));
client.addContainerRequest(request2);
@@ -274,8 +274,9 @@ public class TestAMRMClientContainerRequest {
AMRMClientImpl<ContainerRequest> client, ContainerRequest request,
String location, boolean expectedRelaxLocality,
ExecutionType executionType) {
- ResourceRequest ask = client.remoteRequestsTable.get(request.getPriority(),
- location, executionType, request.getCapability()).remoteRequest;
+ ResourceRequest ask = client.getTable(0)
+ .get(request.getPriority(), location, executionType,
+ request.getCapability()).remoteRequest;
assertEquals(location, ask.getResourceName());
assertEquals(1, ask.getNumContainers());
assertEquals(expectedRelaxLocality, ask.getRelaxLocality());
http://git-wip-us.apache.org/repos/asf/hadoop/blob/19c743c1/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestDistributedScheduling.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestDistributedScheduling.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestDistributedScheduling.java
index 40390d4..4cfc4eb 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestDistributedScheduling.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestDistributedScheduling.java
@@ -366,12 +366,12 @@ public class TestDistributedScheduling extends BaseAMRMProxyE2ETest {
new AMRMClient.ContainerRequest(capability, nodes, racks, priority));
amClient.addContainerRequest(
new AMRMClient.ContainerRequest(capability, null, null, priority2,
- true, null,
+ 0, true, null,
ExecutionTypeRequest.newInstance(
ExecutionType.OPPORTUNISTIC, true)));
amClient.addContainerRequest(
new AMRMClient.ContainerRequest(capability, null, null, priority2,
- true, null,
+ 0, true, null,
ExecutionTypeRequest.newInstance(
ExecutionType.OPPORTUNISTIC, true)));
@@ -381,21 +381,23 @@ public class TestDistributedScheduling extends BaseAMRMProxyE2ETest {
new AMRMClient.ContainerRequest(capability, nodes, racks, priority));
amClient.removeContainerRequest(
new AMRMClient.ContainerRequest(capability, null, null, priority2,
- true, null,
+ 0, true, null,
ExecutionTypeRequest.newInstance(
ExecutionType.OPPORTUNISTIC, true)));
- int containersRequestedNode = amClient.remoteRequestsTable.get(priority,
+ RemoteRequestsTable<ContainerRequest> remoteRequestsTable =
+ amClient.getTable(0);
+ int containersRequestedNode = remoteRequestsTable.get(priority,
node, ExecutionType.GUARANTEED, capability).remoteRequest
.getNumContainers();
- int containersRequestedRack = amClient.remoteRequestsTable.get(priority,
+ int containersRequestedRack = remoteRequestsTable.get(priority,
rack, ExecutionType.GUARANTEED, capability).remoteRequest
.getNumContainers();
- int containersRequestedAny = amClient.remoteRequestsTable.get(priority,
+ int containersRequestedAny = remoteRequestsTable.get(priority,
ResourceRequest.ANY, ExecutionType.GUARANTEED, capability)
.remoteRequest.getNumContainers();
int oppContainersRequestedAny =
- amClient.remoteRequestsTable.get(priority2, ResourceRequest.ANY,
+ remoteRequestsTable.get(priority2, ResourceRequest.ANY,
ExecutionType.OPPORTUNISTIC, capability).remoteRequest
.getNumContainers();
@@ -457,7 +459,7 @@ public class TestDistributedScheduling extends BaseAMRMProxyE2ETest {
new AMRMClient.ContainerRequest(capability, nodes, racks, priority));
amClient.removeContainerRequest(
new AMRMClient.ContainerRequest(capability, nodes, racks, priority2,
- true, null,
+ 0, true, null,
ExecutionTypeRequest.newInstance(
ExecutionType.OPPORTUNISTIC, true)));
assertEquals(4, amClient.ask.size());
@@ -469,7 +471,7 @@ public class TestDistributedScheduling extends BaseAMRMProxyE2ETest {
nodes, racks, priority));
amClient.addContainerRequest(
new AMRMClient.ContainerRequest(capability, nodes, racks, priority2,
- true, null,
+ 0, true, null,
ExecutionTypeRequest.newInstance(
ExecutionType.OPPORTUNISTIC, true)));
@@ -490,7 +492,7 @@ public class TestDistributedScheduling extends BaseAMRMProxyE2ETest {
priority));
amc.removeContainerRequest(
new AMRMClient.ContainerRequest(capability, null, null,
- priority2, true, null,
+ priority2, 0, true, null,
ExecutionTypeRequest.newInstance(
ExecutionType.OPPORTUNISTIC, true)));
throw new Exception();
@@ -571,7 +573,7 @@ public class TestDistributedScheduling extends BaseAMRMProxyE2ETest {
ExecutionTypeRequest execTypeRequest =
ExecutionTypeRequest.newInstance(ExecutionType.OPPORTUNISTIC, true);
ContainerRequest containerRequest = new AMRMClient.ContainerRequest(
- capability, nodes, racks, priority, true, null, execTypeRequest);
+ capability, nodes, racks, priority, 0, true, null, execTypeRequest);
amClient.addContainerRequest(containerRequest);
// Wait until the container is allocated
http://git-wip-us.apache.org/repos/asf/hadoop/blob/19c743c1/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestNMClient.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestNMClient.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestNMClient.java
index 969fb70..3640883 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestNMClient.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestNMClient.java
@@ -252,9 +252,9 @@ public class TestNMClient {
racks, priority));
}
- int containersRequestedAny = rmClient.remoteRequestsTable.get(priority,
- ResourceRequest.ANY, ExecutionType.GUARANTEED, capability)
- .remoteRequest.getNumContainers();
+ int containersRequestedAny = rmClient.getTable(0)
+ .get(priority, ResourceRequest.ANY, ExecutionType.GUARANTEED,
+ capability).remoteRequest.getNumContainers();
// RM should allocate container within 2 calls to allocate()
int allocatedContainerCount = 0;
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
[2/8] hadoop git commit: HDFS-10798. Make the threshold of reporting
FSNamesystem lock contention configurable. Contributed by Erik Krogen.
Posted by aw...@apache.org.
HDFS-10798. Make the threshold of reporting FSNamesystem lock contention configurable. Contributed by Erik Krogen.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/407b519f
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/407b519f
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/407b519f
Branch: refs/heads/HADOOP-13341
Commit: 407b519fb14f79f19ebc4fbdf08204336a7acf77
Parents: 8b7adf4
Author: Zhe Zhang <zh...@apache.org>
Authored: Fri Aug 26 14:19:55 2016 -0700
Committer: Zhe Zhang <zh...@apache.org>
Committed: Fri Aug 26 14:19:55 2016 -0700
----------------------------------------------------------------------
.../java/org/apache/hadoop/hdfs/DFSConfigKeys.java | 5 +++++
.../hadoop/hdfs/server/namenode/FSNamesystem.java | 10 ++++++++--
.../hadoop-hdfs/src/main/resources/hdfs-default.xml | 9 +++++++++
.../hdfs/server/namenode/TestFSNamesystem.java | 15 +++++++++------
4 files changed, 31 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/407b519f/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java
index 3385751..b4cce4a 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java
@@ -407,6 +407,11 @@ public class DFSConfigKeys extends CommonConfigurationKeys {
public static final long
DFS_NAMENODE_MAX_LOCK_HOLD_TO_RELEASE_LEASE_MS_DEFAULT = 25;
+ // Threshold for how long a write lock must be held for the event to be logged
+ public static final String DFS_NAMENODE_WRITE_LOCK_REPORTING_THRESHOLD_MS_KEY =
+ "dfs.namenode.write-lock-reporting-threshold-ms";
+ public static final long DFS_NAMENODE_WRITE_LOCK_REPORTING_THRESHOLD_MS_DEFAULT = 1000L;
+
public static final String DFS_UPGRADE_DOMAIN_FACTOR = "dfs.namenode.upgrade.domain.factor";
public static final int DFS_UPGRADE_DOMAIN_FACTOR_DEFAULT = DFS_REPLICATION_DEFAULT;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/407b519f/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java
index 29f09b9..05fd709 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java
@@ -71,6 +71,8 @@ import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMENODE_MAX_OBJECTS_KEY;
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMENODE_NAME_DIR_KEY;
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMENODE_RESOURCE_CHECK_INTERVAL_DEFAULT;
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMENODE_RESOURCE_CHECK_INTERVAL_KEY;
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMENODE_WRITE_LOCK_REPORTING_THRESHOLD_MS_KEY;
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMENODE_WRITE_LOCK_REPORTING_THRESHOLD_MS_DEFAULT;
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMENODE_RETRY_CACHE_EXPIRYTIME_MILLIS_DEFAULT;
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMENODE_RETRY_CACHE_EXPIRYTIME_MILLIS_KEY;
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMENODE_RETRY_CACHE_HEAP_PERCENT_DEFAULT;
@@ -819,6 +821,10 @@ public class FSNamesystem implements Namesystem, FSNamesystemMBean,
DFS_NAMENODE_MAX_LOCK_HOLD_TO_RELEASE_LEASE_MS_KEY,
DFS_NAMENODE_MAX_LOCK_HOLD_TO_RELEASE_LEASE_MS_DEFAULT);
+ this.writeLockReportingThreshold = conf.getLong(
+ DFS_NAMENODE_WRITE_LOCK_REPORTING_THRESHOLD_MS_KEY,
+ DFS_NAMENODE_WRITE_LOCK_REPORTING_THRESHOLD_MS_DEFAULT);
+
// For testing purposes, allow the DT secret manager to be started regardless
// of whether security is enabled.
alwaysUseDelegationTokensForTests = conf.getBoolean(
@@ -1498,7 +1504,7 @@ public class FSNamesystem implements Namesystem, FSNamesystemMBean,
}
/** Threshold (ms) for long holding write lock report. */
- static final short WRITELOCK_REPORTING_THRESHOLD = 1000;
+ private long writeLockReportingThreshold;
/** Last time stamp for write lock. Keep the longest one for multi-entrance.*/
private long writeLockHeldTimeStamp;
@@ -1532,7 +1538,7 @@ public class FSNamesystem implements Namesystem, FSNamesystemMBean,
this.fsLock.writeLock().unlock();
- if (needReport && writeLockInterval >= WRITELOCK_REPORTING_THRESHOLD) {
+ if (needReport && writeLockInterval >= this.writeLockReportingThreshold) {
LOG.info("FSNamesystem write lock held for " + writeLockInterval +
" ms via\n" + StringUtils.getStackTrace(Thread.currentThread()));
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/407b519f/hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml b/hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml
index e6fde8c..ce880d3 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml
@@ -2615,6 +2615,15 @@
</property>
<property>
+ <name>dfs.namenode.write-lock-reporting-threshold-ms</name>
+ <value>1000</value>
+ <description>When a write lock is held on the namenode for a long time,
+ this will be logged as the lock is released. This sets how long the
+ lock must be held for logging to occur.
+ </description>
+</property>
+
+<property>
<name>dfs.namenode.startup.delay.block.deletion.sec</name>
<value>0</value>
<description>The delay in seconds at which we will pause the blocks deletion
http://git-wip-us.apache.org/repos/asf/hadoop/blob/407b519f/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestFSNamesystem.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestFSNamesystem.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestFSNamesystem.java
index 572b40d..df9001d 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestFSNamesystem.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestFSNamesystem.java
@@ -290,7 +290,10 @@ public class TestFSNamesystem {
*/
@Test(timeout=45000)
public void testFSLockLongHoldingReport() throws Exception {
+ final long writeLockReportingThreshold = 100L;
Configuration conf = new Configuration();
+ conf.setLong(DFSConfigKeys.DFS_NAMENODE_WRITE_LOCK_REPORTING_THRESHOLD_MS_KEY,
+ writeLockReportingThreshold);
FSImage fsImage = Mockito.mock(FSImage.class);
FSEditLog fsEditLog = Mockito.mock(FSEditLog.class);
Mockito.when(fsImage.getEditLog()).thenReturn(fsEditLog);
@@ -301,32 +304,32 @@ public class TestFSNamesystem {
// Don't report if the write lock is held for a short time
fsn.writeLock();
- Thread.sleep(FSNamesystem.WRITELOCK_REPORTING_THRESHOLD / 2);
+ Thread.sleep(writeLockReportingThreshold / 2);
fsn.writeUnlock();
assertFalse(logs.getOutput().contains(GenericTestUtils.getMethodName()));
// Report if the write lock is held for a long time
fsn.writeLock();
- Thread.sleep(FSNamesystem.WRITELOCK_REPORTING_THRESHOLD + 100);
+ Thread.sleep(writeLockReportingThreshold + 10);
logs.clearOutput();
fsn.writeUnlock();
assertTrue(logs.getOutput().contains(GenericTestUtils.getMethodName()));
// Report if the write lock is held (interruptibly) for a long time
fsn.writeLockInterruptibly();
- Thread.sleep(FSNamesystem.WRITELOCK_REPORTING_THRESHOLD + 100);
+ Thread.sleep(writeLockReportingThreshold + 10);
logs.clearOutput();
fsn.writeUnlock();
assertTrue(logs.getOutput().contains(GenericTestUtils.getMethodName()));
// Report if it's held for a long time when re-entering write lock
fsn.writeLock();
- Thread.sleep(FSNamesystem.WRITELOCK_REPORTING_THRESHOLD / 2 + 1);
+ Thread.sleep(writeLockReportingThreshold/ 2 + 1);
fsn.writeLockInterruptibly();
- Thread.sleep(FSNamesystem.WRITELOCK_REPORTING_THRESHOLD / 2 + 1);
+ Thread.sleep(writeLockReportingThreshold / 2 + 1);
fsn.writeLock();
- Thread.sleep(FSNamesystem.WRITELOCK_REPORTING_THRESHOLD / 2);
+ Thread.sleep(writeLockReportingThreshold / 2);
logs.clearOutput();
fsn.writeUnlock();
assertFalse(logs.getOutput().contains(GenericTestUtils.getMethodName()));
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
[7/8] hadoop git commit: HADOOP-13356. Add a function to handle
command_subcommand_OPTS (aw)
Posted by aw...@apache.org.
HADOOP-13356. Add a function to handle command_subcommand_OPTS (aw)
Signed-off-by: Allen Wittenauer <aw...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/defe759b
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/defe759b
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/defe759b
Branch: refs/heads/HADOOP-13341
Commit: defe759bc5a080d164ed218ae5953bcff9380bfa
Parents: 19c743c
Author: Allen Wittenauer <aw...@apache.org>
Authored: Fri Jul 8 09:25:09 2016 -0700
Committer: Allen Wittenauer <aw...@apache.org>
Committed: Fri Aug 26 17:17:29 2016 -0700
----------------------------------------------------------------------
.../src/main/bin/hadoop-functions.sh | 62 ++++++++++++++++
.../test/scripts/hadoop_subcommand_opts.bats | 76 ++++++++++++++++++++
2 files changed, 138 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/defe759b/hadoop-common-project/hadoop-common/src/main/bin/hadoop-functions.sh
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/bin/hadoop-functions.sh b/hadoop-common-project/hadoop-common/src/main/bin/hadoop-functions.sh
index 75554f0..6e58dca 100755
--- a/hadoop-common-project/hadoop-common/src/main/bin/hadoop-functions.sh
+++ b/hadoop-common-project/hadoop-common/src/main/bin/hadoop-functions.sh
@@ -1974,6 +1974,68 @@ function hadoop_verify_user
fi
}
+## @description Add custom (program)_(command)_OPTS to HADOOP_OPTS.
+## @description Also handles the deprecated cases from pre-3.x.
+## @audience public
+## @stability stable
+## @replaceable yes
+## @param program
+## @param subcommand
+## @return will exit on failure conditions
+function hadoop_subcommand_opts
+{
+ declare program=$1
+ declare command=$2
+ declare var
+ declare uvar
+ declare uprogram
+ declare ucommand
+
+ if [[ -z "${program}" || -z "${command}" ]]; then
+ return 1
+ fi
+
+ # bash 4 and up have built-in ways to upper and lower
+ # case the contents of vars. This is faster than
+ # calling tr.
+
+ if [[ -z "${BASH_VERSINFO[0]}" ]] \
+ || [[ "${BASH_VERSINFO[0]}" -lt 4 ]]; then
+ uprogram=$(echo "${program}" | tr '[:lower:]' '[:upper:]')
+ ucommand=$(echo "${command}" | tr '[:lower:]' '[:upper:]')
+ else
+ uprogram=${program^^}
+ ucommand=${command^^}
+ fi
+
+ # HDFS_namenode_OPTS
+ # HADOOP_distcp_OPTS
+ # MAPRED_distcp_OPTS
+ # YARN_sharedcachemanger_OPTS
+ # ...
+ var="${uprogram}_${command}_OPTS"
+
+ # Let's handle all of the deprecation cases early
+ # HADOOP_NAMENODE_OPTS -> HDFS_namenode_OPTS
+ # YARN_RESOURCEMANAGER_OPTS -> YARN_resourcemanager_OPTS
+
+ uvar="${uprogram}_${ucommand}_OPTS"
+ if [[ -n ${!uvar} ]]; then
+ hadoop_deprecate_envvar "${uvar}" "${var}"
+ fi
+
+ uvar="HADOOP_${ucommand}_OPTS"
+ if [[ -n ${!uvar} ]]; then
+ hadoop_deprecate_envvar "${uvar}" "${var}"
+ fi
+
+ if [[ -n ${!var} ]]; then
+ hadoop_debug "Appending ${!var} onto HADOOP_OPTS"
+ HADOOP_OPTS="${HADOOP_OPTS} ${!var}"
+ return 0
+ fi
+}
+
## @description Perform the 'hadoop classpath', etc subcommand with the given
## @description parameters
## @audience private
http://git-wip-us.apache.org/repos/asf/hadoop/blob/defe759b/hadoop-common-project/hadoop-common/src/test/scripts/hadoop_subcommand_opts.bats
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/test/scripts/hadoop_subcommand_opts.bats b/hadoop-common-project/hadoop-common/src/test/scripts/hadoop_subcommand_opts.bats
new file mode 100644
index 0000000..1fbf343
--- /dev/null
+++ b/hadoop-common-project/hadoop-common/src/test/scripts/hadoop_subcommand_opts.bats
@@ -0,0 +1,76 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+load hadoop-functions_test_helper
+
+@test "hadoop_subcommand_opts (missing param)" {
+ HADOOP_OPTS="x"
+ run hadoop_subcommand_opts testvar
+ [ "${status}" = "1" ]
+}
+
+@test "hadoop_subcommand_opts (simple not exist)" {
+ HADOOP_OPTS="x"
+ hadoop_subcommand_opts hadoop subcommand
+ [ "${HADOOP_OPTS}" = "x" ]
+}
+
+@test "hadoop_subcommand_opts (hadoop simple exist)" {
+ HADOOP_OPTS="x"
+ HADOOP_test_OPTS="y"
+ hadoop_subcommand_opts hadoop test
+ echo "${HADOOP_OPTS}"
+ [ "${HADOOP_OPTS}" = "x y" ]
+}
+
+@test "hadoop_subcommand_opts (hadoop complex exist)" {
+ HADOOP_OPTS="x"
+ HADOOP_test_OPTS="y z"
+ hadoop_subcommand_opts hadoop test
+ echo "${HADOOP_OPTS}"
+ [ "${HADOOP_OPTS}" = "x y z" ]
+}
+
+@test "hadoop_subcommand_opts (hdfs simple exist)" {
+ HADOOP_OPTS="x"
+ HDFS_test_OPTS="y"
+ hadoop_subcommand_opts hdfs test
+ echo "${HADOOP_OPTS}"
+ [ "${HADOOP_OPTS}" = "x y" ]
+}
+
+@test "hadoop_subcommand_opts (yarn simple exist)" {
+ HADOOP_OPTS="x"
+ YARN_test_OPTS="y"
+ hadoop_subcommand_opts yarn test
+ echo "${HADOOP_OPTS}"
+ [ "${HADOOP_OPTS}" = "x y" ]
+}
+
+@test "hadoop_subcommand_opts (deprecation case #1)" {
+ HADOOP_OPTS="x"
+ HADOOP_NAMENODE_OPTS="y"
+ hadoop_subcommand_opts hdfs namenode
+ echo "${HADOOP_OPTS}"
+ [ "${HADOOP_OPTS}" = "x y" ]
+}
+
+@test "hadoop_subcommand_opts (deprecation case #2)" {
+ HADOOP_OPTS="x"
+ YARN_RESOURCEMANAGER_OPTS="y"
+ hadoop_subcommand_opts yarn resourcemanager
+ echo "${HADOOP_OPTS}"
+ [ "${HADOOP_OPTS}" = "x y" ]
+}
\ 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
[3/8] hadoop git commit: HDFS-10793. Fix HdfsAuditLogger binary
incompatibility introduced by HDFS-9184. Contributed by Manoj Govindassamy.
Posted by aw...@apache.org.
HDFS-10793. Fix HdfsAuditLogger binary incompatibility introduced by HDFS-9184. Contributed by Manoj Govindassamy.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/a445b82b
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/a445b82b
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/a445b82b
Branch: refs/heads/HADOOP-13341
Commit: a445b82baaa58dcaaa3831e724a9915e9dde57c5
Parents: 407b519
Author: Andrew Wang <wa...@apache.org>
Authored: Fri Aug 26 15:39:18 2016 -0700
Committer: Andrew Wang <wa...@apache.org>
Committed: Fri Aug 26 15:39:18 2016 -0700
----------------------------------------------------------------------
.../hdfs/server/namenode/FSNamesystem.java | 10 +++++++++
.../hdfs/server/namenode/HdfsAuditLogger.java | 23 ++++++++++++++++----
2 files changed, 29 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/a445b82b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java
index 05fd709..52fbaa7 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java
@@ -7010,6 +7010,16 @@ public class FSNamesystem implements Namesystem, FSNamesystemMBean,
logAuditMessage(sb.toString());
}
}
+
+ @Override
+ public void logAuditEvent(boolean succeeded, String userName,
+ InetAddress addr, String cmd, String src, String dst,
+ FileStatus status, UserGroupInformation ugi,
+ DelegationTokenSecretManager dtSecretManager) {
+ this.logAuditEvent(succeeded, userName, addr, cmd, src, dst, status,
+ null /*CallerContext*/, ugi, dtSecretManager);
+ }
+
public void logAuditMessage(String message) {
auditLog.info(message);
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/a445b82b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/HdfsAuditLogger.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/HdfsAuditLogger.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/HdfsAuditLogger.java
index 3e95ce1..894c4df 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/HdfsAuditLogger.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/HdfsAuditLogger.java
@@ -43,9 +43,9 @@ public abstract class HdfsAuditLogger implements AuditLogger {
/**
* Same as
- * {@link #logAuditEvent(boolean, String, InetAddress, String, String, String, FileStatus)}
- * with additional parameters related to logging delegation token tracking
- * IDs.
+ * {@link #logAuditEvent(boolean, String, InetAddress, String, String, String,
+ * FileStatus)} with additional parameters related to logging delegation token
+ * tracking IDs.
*
* @param succeeded Whether authorization succeeded.
* @param userName Name of the user executing the request.
@@ -55,13 +55,28 @@ public abstract class HdfsAuditLogger implements AuditLogger {
* @param dst Path of affected destination file (if any).
* @param stat File information for operations that change the file's metadata
* (permissions, owner, times, etc).
+ * @param callerContext Context information of the caller
* @param ugi UserGroupInformation of the current user, or null if not logging
* token tracking information
* @param dtSecretManager The token secret manager, or null if not logging
* token tracking information
*/
- public abstract void logAuditEvent(boolean succeeded, String userName,
+ public void logAuditEvent(boolean succeeded, String userName,
InetAddress addr, String cmd, String src, String dst,
FileStatus stat, CallerContext callerContext, UserGroupInformation ugi,
+ DelegationTokenSecretManager dtSecretManager) {
+ logAuditEvent(succeeded, userName, addr, cmd, src, dst, stat,
+ ugi, dtSecretManager);
+ }
+
+ /**
+ * Same as
+ * {@link #logAuditEvent(boolean, String, InetAddress, String, String,
+ * String, FileStatus, CallerContext, UserGroupInformation,
+ * DelegationTokenSecretManager)} without {@link CallerContext} information.
+ */
+ public abstract void logAuditEvent(boolean succeeded, String userName,
+ InetAddress addr, String cmd, String src, String dst,
+ FileStatus stat, UserGroupInformation ugi,
DelegationTokenSecretManager dtSecretManager);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org