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 zj...@apache.org on 2015/02/27 17:47:17 UTC
[1/8] hadoop git commit: YARN-3255. RM, NM, JobHistoryServer,
and WebAppProxyServer's main() should support generic options.
Contributed by Konstantin Shvachko.
Repository: hadoop
Updated Branches:
refs/heads/YARN-2928 41a08ad40 -> bf08f7f0e
YARN-3255. RM, NM, JobHistoryServer, and WebAppProxyServer's main() should support generic options. Contributed by Konstantin Shvachko.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/8ca0d957
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/8ca0d957
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/8ca0d957
Branch: refs/heads/YARN-2928
Commit: 8ca0d957c4b1076e801e1cdce5b44aa805de889c
Parents: bfbf076
Author: Konstantin V Shvachko <sh...@apache.org>
Authored: Thu Feb 26 17:12:19 2015 -0800
Committer: Konstantin V Shvachko <sh...@apache.org>
Committed: Thu Feb 26 17:12:19 2015 -0800
----------------------------------------------------------------------
.../java/org/apache/hadoop/mapreduce/v2/hs/JobHistoryServer.java | 2 ++
hadoop-yarn-project/CHANGES.txt | 3 +++
.../org/apache/hadoop/yarn/server/nodemanager/NodeManager.java | 4 +++-
.../hadoop/yarn/server/resourcemanager/ResourceManager.java | 3 +++
.../apache/hadoop/yarn/server/webproxy/WebAppProxyServer.java | 2 ++
5 files changed, 13 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8ca0d957/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/main/java/org/apache/hadoop/mapreduce/v2/hs/JobHistoryServer.java
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/main/java/org/apache/hadoop/mapreduce/v2/hs/JobHistoryServer.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/main/java/org/apache/hadoop/mapreduce/v2/hs/JobHistoryServer.java
index 6d58040..252ac55 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/main/java/org/apache/hadoop/mapreduce/v2/hs/JobHistoryServer.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/main/java/org/apache/hadoop/mapreduce/v2/hs/JobHistoryServer.java
@@ -37,6 +37,7 @@ import org.apache.hadoop.security.SecurityUtil;
import org.apache.hadoop.service.AbstractService;
import org.apache.hadoop.service.CompositeService;
import org.apache.hadoop.util.ExitUtil;
+import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.util.ShutdownHookManager;
import org.apache.hadoop.util.StringUtils;
import org.apache.hadoop.yarn.YarnUncaughtExceptionHandler;
@@ -216,6 +217,7 @@ public class JobHistoryServer extends CompositeService {
new CompositeServiceShutdownHook(jobHistoryServer),
SHUTDOWN_HOOK_PRIORITY);
YarnConfiguration conf = new YarnConfiguration(new JobConf());
+ new GenericOptionsParser(conf, args);
jobHistoryServer.init(conf);
jobHistoryServer.start();
} catch (Throwable t) {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8ca0d957/hadoop-yarn-project/CHANGES.txt
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/CHANGES.txt b/hadoop-yarn-project/CHANGES.txt
index a635592..40f187b 100644
--- a/hadoop-yarn-project/CHANGES.txt
+++ b/hadoop-yarn-project/CHANGES.txt
@@ -330,6 +330,9 @@ Release 2.7.0 - UNRELEASED
YARN-3217. Remove httpclient dependency from hadoop-yarn-server-web-proxy.
(Brahma Reddy Battula via ozawa).
+ YARN-3255. RM, NM, JobHistoryServer, and WebAppProxyServer's main()
+ should support generic options. (shv)
+
OPTIMIZATIONS
YARN-2990. FairScheduler's delay-scheduling always waits for node-local and
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8ca0d957/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/NodeManager.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/NodeManager.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/NodeManager.java
index 7584138..a4be120 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/NodeManager.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/NodeManager.java
@@ -37,6 +37,7 @@ import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem;
import org.apache.hadoop.security.Credentials;
import org.apache.hadoop.security.SecurityUtil;
import org.apache.hadoop.service.CompositeService;
+import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.util.NodeHealthScriptRunner;
import org.apache.hadoop.util.ReflectionUtils;
import org.apache.hadoop.util.ShutdownHookManager;
@@ -523,11 +524,12 @@ public class NodeManager extends CompositeService
return this.context;
}
- public static void main(String[] args) {
+ public static void main(String[] args) throws IOException {
Thread.setDefaultUncaughtExceptionHandler(new YarnUncaughtExceptionHandler());
StringUtils.startupShutdownMessage(NodeManager.class, args, LOG);
NodeManager nodeManager = new NodeManager();
Configuration conf = new YarnConfiguration();
+ new GenericOptionsParser(conf, args);
nodeManager.initAndStartNodeManager(conf, false);
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8ca0d957/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ResourceManager.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/ResourceManager.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ResourceManager.java
index a93372a..8bd8e21 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ResourceManager.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ResourceManager.java
@@ -38,6 +38,7 @@ import org.apache.hadoop.service.AbstractService;
import org.apache.hadoop.service.CompositeService;
import org.apache.hadoop.service.Service;
import org.apache.hadoop.util.ExitUtil;
+import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.util.ReflectionUtils;
import org.apache.hadoop.util.ShutdownHookManager;
import org.apache.hadoop.util.StringUtils;
@@ -1193,6 +1194,8 @@ public class ResourceManager extends CompositeService implements Recoverable {
StringUtils.startupShutdownMessage(ResourceManager.class, argv, LOG);
try {
Configuration conf = new YarnConfiguration();
+ GenericOptionsParser hParser = new GenericOptionsParser(conf, argv);
+ argv = hParser.getRemainingArgs();
// If -format-state-store, then delete RMStateStore; else startup normally
if (argv.length == 1 && argv[0].equals("-format-state-store")) {
deleteRMStateStore(conf);
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8ca0d957/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-web-proxy/src/main/java/org/apache/hadoop/yarn/server/webproxy/WebAppProxyServer.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-web-proxy/src/main/java/org/apache/hadoop/yarn/server/webproxy/WebAppProxyServer.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-web-proxy/src/main/java/org/apache/hadoop/yarn/server/webproxy/WebAppProxyServer.java
index 881e2e7..b1c563c 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-web-proxy/src/main/java/org/apache/hadoop/yarn/server/webproxy/WebAppProxyServer.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-web-proxy/src/main/java/org/apache/hadoop/yarn/server/webproxy/WebAppProxyServer.java
@@ -25,6 +25,7 @@ import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.security.SecurityUtil;
import org.apache.hadoop.service.CompositeService;
import org.apache.hadoop.util.ExitUtil;
+import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.util.ShutdownHookManager;
import org.apache.hadoop.util.StringUtils;
import org.apache.hadoop.yarn.YarnUncaughtExceptionHandler;
@@ -89,6 +90,7 @@ public class WebAppProxyServer extends CompositeService {
StringUtils.startupShutdownMessage(WebAppProxyServer.class, args, LOG);
try {
YarnConfiguration configuration = new YarnConfiguration();
+ new GenericOptionsParser(configuration, args);
WebAppProxyServer proxyServer = startServer(configuration);
proxyServer.proxy.join();
} catch (Throwable t) {
[6/8] hadoop git commit: YARN-2820. Retry in FileSystemRMStateStore
when FS's operations fail due to IOException. Contributed by Zhihai Xu.
Posted by zj...@apache.org.
YARN-2820. Retry in FileSystemRMStateStore when FS's operations fail due to IOException. Contributed by Zhihai Xu.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/01a16219
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/01a16219
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/01a16219
Branch: refs/heads/YARN-2928
Commit: 01a1621930df17a745dd37892996c68fca3447d1
Parents: a979f3b
Author: Tsuyoshi Ozawa <oz...@apache.org>
Authored: Sat Feb 28 00:56:44 2015 +0900
Committer: Tsuyoshi Ozawa <oz...@apache.org>
Committed: Sat Feb 28 00:56:44 2015 +0900
----------------------------------------------------------------------
hadoop-yarn-project/CHANGES.txt | 3 +
.../hadoop/yarn/conf/YarnConfiguration.java | 9 +
.../src/main/resources/yarn-default.xml | 15 +
.../recovery/FileSystemRMStateStore.java | 303 ++++++++++++++-----
.../recovery/TestFSRMStateStore.java | 5 +
5 files changed, 265 insertions(+), 70 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/01a16219/hadoop-yarn-project/CHANGES.txt
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/CHANGES.txt b/hadoop-yarn-project/CHANGES.txt
index 40f187b..38dd9fa 100644
--- a/hadoop-yarn-project/CHANGES.txt
+++ b/hadoop-yarn-project/CHANGES.txt
@@ -333,6 +333,9 @@ Release 2.7.0 - UNRELEASED
YARN-3255. RM, NM, JobHistoryServer, and WebAppProxyServer's main()
should support generic options. (shv)
+ YARN-2820. Retry in FileSystemRMStateStore when FS's operations fail
+ due to IOException. (Zhihai Xu via ozawa)
+
OPTIMIZATIONS
YARN-2990. FairScheduler's delay-scheduling always waits for node-local and
http://git-wip-us.apache.org/repos/asf/hadoop/blob/01a16219/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java
index 05c6cbf..ff06eea 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java
@@ -508,6 +508,15 @@ public class YarnConfiguration extends Configuration {
public static final String DEFAULT_FS_RM_STATE_STORE_RETRY_POLICY_SPEC =
"2000, 500";
+ public static final String FS_RM_STATE_STORE_NUM_RETRIES =
+ RM_PREFIX + "fs.state-store.num-retries";
+ public static final int DEFAULT_FS_RM_STATE_STORE_NUM_RETRIES = 0;
+
+ public static final String FS_RM_STATE_STORE_RETRY_INTERVAL_MS =
+ RM_PREFIX + "fs.state-store.retry-interval-ms";
+ public static final long DEFAULT_FS_RM_STATE_STORE_RETRY_INTERVAL_MS =
+ 1000L;
+
public static final String RM_LEVELDB_STORE_PATH = RM_PREFIX
+ "leveldb-state-store.path";
http://git-wip-us.apache.org/repos/asf/hadoop/blob/01a16219/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/resources/yarn-default.xml
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/resources/yarn-default.xml b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/resources/yarn-default.xml
index a7958a5..df730d5 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/resources/yarn-default.xml
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/resources/yarn-default.xml
@@ -420,6 +420,21 @@
</property>
<property>
+ <description>the number of retries to recover from IOException in
+ FileSystemRMStateStore.
+ </description>
+ <name>yarn.resourcemanager.fs.state-store.num-retries</name>
+ <value>0</value>
+ </property>
+
+ <property>
+ <description>Retry interval in milliseconds in FileSystemRMStateStore.
+ </description>
+ <name>yarn.resourcemanager.fs.state-store.retry-interval-ms</name>
+ <value>1000</value>
+ </property>
+
+ <property>
<description>Local path where the RM state will be stored when using
org.apache.hadoop.yarn.server.resourcemanager.recovery.LeveldbRMStateStore
as the value for yarn.resourcemanager.store.class</description>
http://git-wip-us.apache.org/repos/asf/hadoop/blob/01a16219/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/FileSystemRMStateStore.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/recovery/FileSystemRMStateStore.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/FileSystemRMStateStore.java
index 6e830a0..8147597 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/FileSystemRMStateStore.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/FileSystemRMStateStore.java
@@ -92,6 +92,8 @@ public class FileSystemRMStateStore extends RMStateStore {
Path rmDTSecretManagerRoot;
private Path rmAppRoot;
private Path dtSequenceNumberPath = null;
+ private int fsNumRetries;
+ private long fsRetryInterval;
@VisibleForTesting
Path fsWorkingPath;
@@ -106,6 +108,12 @@ public class FileSystemRMStateStore extends RMStateStore {
rmAppRoot = new Path(rootDirPath, RM_APP_ROOT);
amrmTokenSecretManagerRoot =
new Path(rootDirPath, AMRMTOKEN_SECRET_MANAGER_ROOT);
+ fsNumRetries =
+ conf.getInt(YarnConfiguration.FS_RM_STATE_STORE_NUM_RETRIES,
+ YarnConfiguration.DEFAULT_FS_RM_STATE_STORE_NUM_RETRIES);
+ fsRetryInterval =
+ conf.getLong(YarnConfiguration.FS_RM_STATE_STORE_RETRY_INTERVAL_MS,
+ YarnConfiguration.DEFAULT_FS_RM_STATE_STORE_RETRY_INTERVAL_MS);
}
@Override
@@ -121,14 +129,14 @@ public class FileSystemRMStateStore extends RMStateStore {
conf.set("dfs.client.retry.policy.spec", retryPolicy);
fs = fsWorkingPath.getFileSystem(conf);
- fs.mkdirs(rmDTSecretManagerRoot);
- fs.mkdirs(rmAppRoot);
- fs.mkdirs(amrmTokenSecretManagerRoot);
+ mkdirsWithRetries(rmDTSecretManagerRoot);
+ mkdirsWithRetries(rmAppRoot);
+ mkdirsWithRetries(amrmTokenSecretManagerRoot);
}
@Override
protected synchronized void closeInternal() throws Exception {
- fs.close();
+ closeWithRetries();
}
@Override
@@ -139,9 +147,9 @@ public class FileSystemRMStateStore extends RMStateStore {
@Override
protected synchronized Version loadVersion() throws Exception {
Path versionNodePath = getNodePath(rootDirPath, VERSION_NODE);
- FileStatus status = getFileStatus(versionNodePath);
+ FileStatus status = getFileStatusWithRetries(versionNodePath);
if (status != null) {
- byte[] data = readFile(versionNodePath, status.getLen());
+ byte[] data = readFileWithRetries(versionNodePath, status.getLen());
Version version =
new VersionPBImpl(VersionProto.parseFrom(data));
return version;
@@ -154,10 +162,10 @@ public class FileSystemRMStateStore extends RMStateStore {
Path versionNodePath = getNodePath(rootDirPath, VERSION_NODE);
byte[] data =
((VersionPBImpl) CURRENT_VERSION_INFO).getProto().toByteArray();
- if (fs.exists(versionNodePath)) {
+ if (existsWithRetries(versionNodePath)) {
updateFile(versionNodePath, data);
} else {
- writeFile(versionNodePath, data);
+ writeFileWithRetries(versionNodePath, data);
}
}
@@ -165,10 +173,10 @@ public class FileSystemRMStateStore extends RMStateStore {
public synchronized long getAndIncrementEpoch() throws Exception {
Path epochNodePath = getNodePath(rootDirPath, EPOCH_NODE);
long currentEpoch = 0;
- FileStatus status = getFileStatus(epochNodePath);
+ FileStatus status = getFileStatusWithRetries(epochNodePath);
if (status != null) {
// load current epoch
- byte[] data = readFile(epochNodePath, status.getLen());
+ byte[] data = readFileWithRetries(epochNodePath, status.getLen());
Epoch epoch = new EpochPBImpl(EpochProto.parseFrom(data));
currentEpoch = epoch.getEpoch();
// increment epoch and store it
@@ -179,7 +187,7 @@ public class FileSystemRMStateStore extends RMStateStore {
// initialize epoch file with 1 for the next time.
byte[] storeData = Epoch.newInstance(currentEpoch + 1).getProto()
.toByteArray();
- writeFile(epochNodePath, storeData);
+ writeFileWithRetries(epochNodePath, storeData);
}
return currentEpoch;
}
@@ -201,12 +209,14 @@ public class FileSystemRMStateStore extends RMStateStore {
checkAndResumeUpdateOperation(amrmTokenSecretManagerRoot);
Path amrmTokenSecretManagerStateDataDir =
new Path(amrmTokenSecretManagerRoot, AMRMTOKEN_SECRET_MANAGER_NODE);
- FileStatus status = getFileStatus(amrmTokenSecretManagerStateDataDir);
+ FileStatus status = getFileStatusWithRetries(
+ amrmTokenSecretManagerStateDataDir);
if (status == null) {
return;
}
assert status.isFile();
- byte[] data = readFile(amrmTokenSecretManagerStateDataDir, status.getLen());
+ byte[] data = readFileWithRetries(amrmTokenSecretManagerStateDataDir,
+ status.getLen());
AMRMTokenSecretManagerStatePBImpl stateData =
new AMRMTokenSecretManagerStatePBImpl(
AMRMTokenSecretManagerStateProto.parseFrom(data));
@@ -220,16 +230,18 @@ public class FileSystemRMStateStore extends RMStateStore {
List<ApplicationAttemptStateData> attempts =
new ArrayList<ApplicationAttemptStateData>();
- for (FileStatus appDir : fs.listStatus(rmAppRoot)) {
+ for (FileStatus appDir : listStatusWithRetries(rmAppRoot)) {
checkAndResumeUpdateOperation(appDir.getPath());
- for (FileStatus childNodeStatus : fs.listStatus(appDir.getPath())) {
+ for (FileStatus childNodeStatus :
+ listStatusWithRetries(appDir.getPath())) {
assert childNodeStatus.isFile();
String childNodeName = childNodeStatus.getPath().getName();
- if (checkAndRemovePartialRecord(childNodeStatus.getPath())) {
+ if (checkAndRemovePartialRecordWithRetries(
+ childNodeStatus.getPath())) {
continue;
}
- byte[] childData =
- readFile(childNodeStatus.getPath(), childNodeStatus.getLen());
+ byte[] childData = readFileWithRetries(childNodeStatus.getPath(),
+ childNodeStatus.getLen());
if (childNodeName.startsWith(ApplicationId.appIdStrPrefix)) {
// application
if (LOG.isDebugEnabled()) {
@@ -292,7 +304,7 @@ public class FileSystemRMStateStore extends RMStateStore {
// If it does, the prior updateFile is failed on half way. We need to
// complete replacing the old file first.
FileStatus[] newChildNodes =
- fs.listStatus(path, new PathFilter() {
+ listStatusWithRetries(path, new PathFilter() {
@Override
public boolean accept(Path path) {
return path.getName().endsWith(".new");
@@ -310,12 +322,12 @@ public class FileSystemRMStateStore extends RMStateStore {
}
private void loadRMDTSecretManagerState(RMState rmState) throws Exception {
checkAndResumeUpdateOperation(rmDTSecretManagerRoot);
- FileStatus[] childNodes = fs.listStatus(rmDTSecretManagerRoot);
+ FileStatus[] childNodes = listStatusWithRetries(rmDTSecretManagerRoot);
for(FileStatus childNodeStatus : childNodes) {
assert childNodeStatus.isFile();
String childNodeName = childNodeStatus.getPath().getName();
- if (checkAndRemovePartialRecord(childNodeStatus.getPath())) {
+ if (checkAndRemovePartialRecordWithRetries(childNodeStatus.getPath())) {
continue;
}
if(childNodeName.startsWith(DELEGATION_TOKEN_SEQUENCE_NUMBER_PREFIX)) {
@@ -325,35 +337,36 @@ public class FileSystemRMStateStore extends RMStateStore {
}
Path childNodePath = getNodePath(rmDTSecretManagerRoot, childNodeName);
- byte[] childData = readFile(childNodePath, childNodeStatus.getLen());
+ byte[] childData = readFileWithRetries(childNodePath,
+ childNodeStatus.getLen());
ByteArrayInputStream is = new ByteArrayInputStream(childData);
- DataInputStream fsIn = new DataInputStream(is);
- if(childNodeName.startsWith(DELEGATION_KEY_PREFIX)){
- DelegationKey key = new DelegationKey();
- key.readFields(fsIn);
- rmState.rmSecretManagerState.masterKeyState.add(key);
- if (LOG.isDebugEnabled()) {
- LOG.debug("Loaded delegation key: keyId=" + key.getKeyId()
- + ", expirationDate=" + key.getExpiryDate());
- }
- } else if (childNodeName.startsWith(DELEGATION_TOKEN_PREFIX)) {
- RMDelegationTokenIdentifierData identifierData =
- new RMDelegationTokenIdentifierData();
- identifierData.readFields(fsIn);
- RMDelegationTokenIdentifier identifier =
- identifierData.getTokenIdentifier();
- long renewDate = identifierData.getRenewDate();
-
- rmState.rmSecretManagerState.delegationTokenState.put(identifier,
- renewDate);
- if (LOG.isDebugEnabled()) {
- LOG.debug("Loaded RMDelegationTokenIdentifier: " + identifier
- + " renewDate=" + renewDate);
+ try (DataInputStream fsIn = new DataInputStream(is)) {
+ if (childNodeName.startsWith(DELEGATION_KEY_PREFIX)) {
+ DelegationKey key = new DelegationKey();
+ key.readFields(fsIn);
+ rmState.rmSecretManagerState.masterKeyState.add(key);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Loaded delegation key: keyId=" + key.getKeyId()
+ + ", expirationDate=" + key.getExpiryDate());
+ }
+ } else if (childNodeName.startsWith(DELEGATION_TOKEN_PREFIX)) {
+ RMDelegationTokenIdentifierData identifierData =
+ new RMDelegationTokenIdentifierData();
+ identifierData.readFields(fsIn);
+ RMDelegationTokenIdentifier identifier =
+ identifierData.getTokenIdentifier();
+ long renewDate = identifierData.getRenewDate();
+
+ rmState.rmSecretManagerState.delegationTokenState.put(identifier,
+ renewDate);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Loaded RMDelegationTokenIdentifier: " + identifier
+ + " renewDate=" + renewDate);
+ }
+ } else {
+ LOG.warn("Unknown file for recovering RMDelegationTokenSecretManager");
}
- } else {
- LOG.warn("Unknown file for recovering RMDelegationTokenSecretManager");
}
- fsIn.close();
}
}
@@ -361,7 +374,7 @@ public class FileSystemRMStateStore extends RMStateStore {
public synchronized void storeApplicationStateInternal(ApplicationId appId,
ApplicationStateData appStateDataPB) throws Exception {
Path appDirPath = getAppDir(rmAppRoot, appId);
- fs.mkdirs(appDirPath);
+ mkdirsWithRetries(appDirPath);
Path nodeCreatePath = getNodePath(appDirPath, appId.toString());
LOG.info("Storing info for app: " + appId + " at: " + nodeCreatePath);
@@ -369,7 +382,7 @@ public class FileSystemRMStateStore extends RMStateStore {
try {
// currently throw all exceptions. May need to respond differently for HA
// based on whether we have lost the right to write to FS
- writeFile(nodeCreatePath, appStateData);
+ writeFileWithRetries(nodeCreatePath, appStateData);
} catch (Exception e) {
LOG.info("Error storing info for app: " + appId, e);
throw e;
@@ -408,7 +421,7 @@ public class FileSystemRMStateStore extends RMStateStore {
try {
// currently throw all exceptions. May need to respond differently for HA
// based on whether we have lost the right to write to FS
- writeFile(nodeCreatePath, attemptStateData);
+ writeFileWithRetries(nodeCreatePath, attemptStateData);
} catch (Exception e) {
LOG.info("Error storing info for attempt: " + appAttemptId, e);
throw e;
@@ -444,7 +457,7 @@ public class FileSystemRMStateStore extends RMStateStore {
appState.getApplicationSubmissionContext().getApplicationId();
Path nodeRemovePath = getAppDir(rmAppRoot, appId);
LOG.info("Removing info for app: " + appId + " at: " + nodeRemovePath);
- deleteFile(nodeRemovePath);
+ deleteFileWithRetries(nodeRemovePath);
}
@Override
@@ -460,7 +473,7 @@ public class FileSystemRMStateStore extends RMStateStore {
Path nodeCreatePath = getNodePath(rmDTSecretManagerRoot,
DELEGATION_TOKEN_PREFIX + identifier.getSequenceNumber());
LOG.info("Removing RMDelegationToken_" + identifier.getSequenceNumber());
- deleteFile(nodeCreatePath);
+ deleteFileWithRetries(nodeCreatePath);
}
@Override
@@ -483,7 +496,7 @@ public class FileSystemRMStateStore extends RMStateStore {
updateFile(nodeCreatePath, identifierData.toByteArray());
} else {
LOG.info("Storing RMDelegationToken_" + identifier.getSequenceNumber());
- writeFile(nodeCreatePath, identifierData.toByteArray());
+ writeFileWithRetries(nodeCreatePath, identifierData.toByteArray());
// store sequence number
Path latestSequenceNumberPath = getNodePath(rmDTSecretManagerRoot,
@@ -492,11 +505,12 @@ public class FileSystemRMStateStore extends RMStateStore {
LOG.info("Storing " + DELEGATION_TOKEN_SEQUENCE_NUMBER_PREFIX
+ identifier.getSequenceNumber());
if (dtSequenceNumberPath == null) {
- if (!createFile(latestSequenceNumberPath)) {
+ if (!createFileWithRetries(latestSequenceNumberPath)) {
throw new Exception("Failed to create " + latestSequenceNumberPath);
}
} else {
- if (!renameFile(dtSequenceNumberPath, latestSequenceNumberPath)) {
+ if (!renameFileWithRetries(dtSequenceNumberPath,
+ latestSequenceNumberPath)) {
throw new Exception("Failed to rename " + dtSequenceNumberPath);
}
}
@@ -510,11 +524,11 @@ public class FileSystemRMStateStore extends RMStateStore {
Path nodeCreatePath = getNodePath(rmDTSecretManagerRoot,
DELEGATION_KEY_PREFIX + masterKey.getKeyId());
ByteArrayOutputStream os = new ByteArrayOutputStream();
- DataOutputStream fsOut = new DataOutputStream(os);
- LOG.info("Storing RMDelegationKey_" + masterKey.getKeyId());
- masterKey.write(fsOut);
- writeFile(nodeCreatePath, os.toByteArray());
- fsOut.close();
+ try (DataOutputStream fsOut = new DataOutputStream(os)) {
+ LOG.info("Storing RMDelegationKey_" + masterKey.getKeyId());
+ masterKey.write(fsOut);
+ writeFileWithRetries(nodeCreatePath, os.toByteArray());
+ }
}
@Override
@@ -523,13 +537,13 @@ public class FileSystemRMStateStore extends RMStateStore {
Path nodeCreatePath = getNodePath(rmDTSecretManagerRoot,
DELEGATION_KEY_PREFIX + masterKey.getKeyId());
LOG.info("Removing RMDelegationKey_"+ masterKey.getKeyId());
- deleteFile(nodeCreatePath);
+ deleteFileWithRetries(nodeCreatePath);
}
@Override
- public synchronized void deleteStore() throws IOException {
- if (fs.exists(rootDirPath)) {
- fs.delete(rootDirPath, true);
+ public synchronized void deleteStore() throws Exception {
+ if (existsWithRetries(rootDirPath)) {
+ deleteFileWithRetries(rootDirPath);
}
}
@@ -539,6 +553,146 @@ public class FileSystemRMStateStore extends RMStateStore {
// FileSystem related code
+ private boolean checkAndRemovePartialRecordWithRetries(final Path record)
+ throws Exception {
+ return new FSAction<Boolean>() {
+ @Override
+ public Boolean run() throws Exception {
+ return checkAndRemovePartialRecord(record);
+ }
+ }.runWithRetries();
+ }
+
+ private void mkdirsWithRetries(final Path appDirPath) throws Exception {
+ new FSAction<Void>() {
+ @Override
+ public Void run() throws Exception {
+ fs.mkdirs(appDirPath);
+ return null;
+ }
+ }.runWithRetries();
+ }
+
+ private void writeFileWithRetries(final Path outputPath,final byte[] data)
+ throws Exception {
+ new FSAction<Void>() {
+ @Override
+ public Void run() throws Exception {
+ writeFile(outputPath, data);
+ return null;
+ }
+ }.runWithRetries();
+ }
+
+ private void deleteFileWithRetries(final Path deletePath) throws Exception {
+ new FSAction<Void>() {
+ @Override
+ public Void run() throws Exception {
+ deleteFile(deletePath);
+ return null;
+ }
+ }.runWithRetries();
+ }
+
+ private boolean renameFileWithRetries(final Path src, final Path dst)
+ throws Exception {
+ return new FSAction<Boolean>() {
+ @Override
+ public Boolean run() throws Exception {
+ return renameFile(src, dst);
+ }
+ }.runWithRetries();
+ }
+
+ private boolean createFileWithRetries(final Path newFile) throws Exception {
+ return new FSAction<Boolean>() {
+ @Override
+ public Boolean run() throws Exception {
+ return createFile(newFile);
+ }
+ }.runWithRetries();
+ }
+
+ private FileStatus getFileStatusWithRetries(final Path path)
+ throws Exception {
+ return new FSAction<FileStatus>() {
+ @Override
+ public FileStatus run() throws Exception {
+ return getFileStatus(path);
+ }
+ }.runWithRetries();
+ }
+
+ private boolean existsWithRetries(final Path path) throws Exception {
+ return new FSAction<Boolean>() {
+ @Override
+ public Boolean run() throws Exception {
+ return fs.exists(path);
+ }
+ }.runWithRetries();
+ }
+
+ private byte[] readFileWithRetries(final Path inputPath, final long len)
+ throws Exception {
+ return new FSAction<byte[]>() {
+ @Override
+ public byte[] run() throws Exception {
+ return readFile(inputPath, len);
+ }
+ }.runWithRetries();
+ }
+
+ private FileStatus[] listStatusWithRetries(final Path path)
+ throws Exception {
+ return new FSAction<FileStatus[]>() {
+ @Override
+ public FileStatus[] run() throws Exception {
+ return fs.listStatus(path);
+ }
+ }.runWithRetries();
+ }
+
+ private FileStatus[] listStatusWithRetries(final Path path,
+ final PathFilter filter) throws Exception {
+ return new FSAction<FileStatus[]>() {
+ @Override
+ public FileStatus[] run() throws Exception {
+ return fs.listStatus(path, filter);
+ }
+ }.runWithRetries();
+ }
+
+ private void closeWithRetries() throws Exception {
+ new FSAction<Void>() {
+ @Override
+ public Void run() throws Exception {
+ fs.close();
+ return null;
+ }
+ }.runWithRetries();
+ }
+
+ private abstract class FSAction<T> {
+ abstract T run() throws Exception;
+
+ T runWithRetries() throws Exception {
+ int retry = 0;
+ while (true) {
+ try {
+ return run();
+ } catch (IOException e) {
+ LOG.info("Exception while executing a FS operation.", e);
+ if (++retry > fsNumRetries) {
+ LOG.info("Maxed out FS retries. Giving up!");
+ throw e;
+ }
+ LOG.info("Retrying operation on FS. Retry no. " + retry);
+ Thread.sleep(fsRetryInterval);
+ }
+ }
+ }
+ }
+
private void deleteFile(Path deletePath) throws Exception {
if(!fs.delete(deletePath, true)) {
throw new Exception("Failed to delete " + deletePath);
@@ -595,18 +749,18 @@ public class FileSystemRMStateStore extends RMStateStore {
*/
protected void updateFile(Path outputPath, byte[] data) throws Exception {
Path newPath = new Path(outputPath.getParent(), outputPath.getName() + ".new");
- // use writeFile to make sure .new file is created atomically
- writeFile(newPath, data);
+ // use writeFileWithRetries to make sure .new file is created atomically
+ writeFileWithRetries(newPath, data);
replaceFile(newPath, outputPath);
}
protected void replaceFile(Path srcPath, Path dstPath) throws Exception {
- if (fs.exists(dstPath)) {
- deleteFile(dstPath);
+ if (existsWithRetries(dstPath)) {
+ deleteFileWithRetries(dstPath);
} else {
LOG.info("File doesn't exist. Skip deleting the file " + dstPath);
}
- fs.rename(srcPath, dstPath);
+ renameFileWithRetries(srcPath, dstPath);
}
@Private
@@ -637,8 +791,17 @@ public class FileSystemRMStateStore extends RMStateStore {
if (isUpdate) {
updateFile(nodeCreatePath, stateData);
} else {
- writeFile(nodeCreatePath, stateData);
+ writeFileWithRetries(nodeCreatePath, stateData);
}
}
+ @VisibleForTesting
+ public int getNumRetries() {
+ return fsNumRetries;
+ }
+
+ @VisibleForTesting
+ public long getRetryInterval() {
+ return fsRetryInterval;
+ }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/01a16219/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/TestFSRMStateStore.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/recovery/TestFSRMStateStore.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/TestFSRMStateStore.java
index d0d19e3..675d73c 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/TestFSRMStateStore.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/TestFSRMStateStore.java
@@ -100,7 +100,12 @@ public class TestFSRMStateStore extends RMStateStoreTestBase {
workingDirPathURI.toString());
conf.set(YarnConfiguration.FS_RM_STATE_STORE_RETRY_POLICY_SPEC,
"100,6000");
+ conf.setInt(YarnConfiguration.FS_RM_STATE_STORE_NUM_RETRIES, 5);
+ conf.setLong(YarnConfiguration.FS_RM_STATE_STORE_RETRY_INTERVAL_MS,
+ 900L);
this.store = new TestFileSystemRMStore(conf);
+ Assert.assertEquals(store.getNumRetries(), 5);
+ Assert.assertEquals(store.getRetryInterval(), 900L);
return store;
}
[5/8] hadoop git commit: HDFS-7308. Change the packet chunk size
computation in DFSOutputStream in order to enforce packet size <= 64kB.
Contributed by Takuya Fukudome
Posted by zj...@apache.org.
HDFS-7308. Change the packet chunk size computation in DFSOutputStream in order to enforce packet size <= 64kB. Contributed by Takuya Fukudome
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/a979f3b5
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/a979f3b5
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/a979f3b5
Branch: refs/heads/YARN-2928
Commit: a979f3b58fafebbd6118ec1f861cf3f62c59c9cb
Parents: 4f75b15
Author: Tsz-Wo Nicholas Sze <sz...@hortonworks.com>
Authored: Fri Feb 27 23:45:37 2015 +0800
Committer: Tsz-Wo Nicholas Sze <sz...@hortonworks.com>
Committed: Fri Feb 27 23:45:37 2015 +0800
----------------------------------------------------------------------
hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt | 3 ++
.../org/apache/hadoop/hdfs/DFSOutputStream.java | 3 +-
.../apache/hadoop/hdfs/TestDFSOutputStream.java | 31 ++++++++++++++++++++
3 files changed, 36 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/a979f3b5/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt b/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
index 8556afd..b2422d6 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
+++ b/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
@@ -682,6 +682,9 @@ Release 2.7.0 - UNRELEASED
HDFS-7819. Log WARN message for the blocks which are not in Block ID based
layout (Rakesh R via Colin P. McCabe)
+ HDFS-7308. Change the packet chunk size computation in DFSOutputStream in
+ order to enforce packet size <= 64kB. (Takuya Fukudome via szetszwo)
+
OPTIMIZATIONS
HDFS-7454. Reduce memory footprint for AclEntries in NameNode.
http://git-wip-us.apache.org/repos/asf/hadoop/blob/a979f3b5/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSOutputStream.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSOutputStream.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSOutputStream.java
index 9d7dca9..b3e8c97 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSOutputStream.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSOutputStream.java
@@ -1851,8 +1851,9 @@ public class DFSOutputStream extends FSOutputSummer
}
private void computePacketChunkSize(int psize, int csize) {
+ final int bodySize = psize - PacketHeader.PKT_MAX_HEADER_LEN;
final int chunkSize = csize + getChecksumSize();
- chunksPerPacket = Math.max(psize/chunkSize, 1);
+ chunksPerPacket = Math.max(bodySize/chunkSize, 1);
packetSize = chunkSize*chunksPerPacket;
if (DFSClient.LOG.isDebugEnabled()) {
DFSClient.LOG.debug("computePacketChunkSize: src=" + src +
http://git-wip-us.apache.org/repos/asf/hadoop/blob/a979f3b5/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDFSOutputStream.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDFSOutputStream.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDFSOutputStream.java
index 678a3b8..7269e39 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDFSOutputStream.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDFSOutputStream.java
@@ -18,6 +18,8 @@
package org.apache.hadoop.hdfs;
import java.io.IOException;
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.hadoop.conf.Configuration;
@@ -66,6 +68,35 @@ public class TestDFSOutputStream {
dos.close();
}
+ /**
+ * The computePacketChunkSize() method of DFSOutputStream should set the actual
+ * packet size < 64kB. See HDFS-7308 for details.
+ */
+ @Test
+ public void testComputePacketChunkSize()
+ throws Exception {
+ DistributedFileSystem fs = cluster.getFileSystem();
+ FSDataOutputStream os = fs.create(new Path("/test"));
+ DFSOutputStream dos = (DFSOutputStream) Whitebox.getInternalState(os,
+ "wrappedStream");
+
+ final int packetSize = 64*1024;
+ final int bytesPerChecksum = 512;
+
+ Method method = dos.getClass().getDeclaredMethod("computePacketChunkSize",
+ int.class, int.class);
+ method.setAccessible(true);
+ method.invoke(dos, packetSize, bytesPerChecksum);
+
+ Field field = dos.getClass().getDeclaredField("packetSize");
+ field.setAccessible(true);
+
+ Assert.assertTrue((Integer) field.get(dos) + 33 < packetSize);
+ // If PKT_MAX_HEADER_LEN is 257, actual packet size come to over 64KB
+ // without a fix on HDFS-7308.
+ Assert.assertTrue((Integer) field.get(dos) + 257 < packetSize);
+ }
+
@AfterClass
public static void tearDown() {
cluster.shutdown();
[8/8] hadoop git commit: YARN-3125. Made the distributed shell use
timeline service next gen and add an integration test for it. Contributed by
Junping Du and Li Lu.
Posted by zj...@apache.org.
YARN-3125. Made the distributed shell use timeline service next gen and add an integration test for it. Contributed by Junping Du and Li Lu.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/bf08f7f0
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/bf08f7f0
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/bf08f7f0
Branch: refs/heads/YARN-2928
Commit: bf08f7f0ed4900ce52f98137297dd1a47ba2a536
Parents: 667f69c
Author: Zhijie Shen <zj...@apache.org>
Authored: Fri Feb 27 08:46:42 2015 -0800
Committer: Zhijie Shen <zj...@apache.org>
Committed: Fri Feb 27 08:46:42 2015 -0800
----------------------------------------------------------------------
hadoop-yarn-project/CHANGES.txt | 3 +
.../distributedshell/ApplicationMaster.java | 163 +++++++++++++++++--
.../applications/distributedshell/Client.java | 20 ++-
.../distributedshell/TestDistributedShell.java | 79 +++++++--
.../aggregator/BaseAggregatorService.java | 6 +
.../aggregator/PerNodeAggregatorServer.java | 4 +-
6 files changed, 250 insertions(+), 25 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/bf08f7f0/hadoop-yarn-project/CHANGES.txt
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/CHANGES.txt b/hadoop-yarn-project/CHANGES.txt
index c659126..1f4a68d 100644
--- a/hadoop-yarn-project/CHANGES.txt
+++ b/hadoop-yarn-project/CHANGES.txt
@@ -20,6 +20,9 @@ Branch YARN-2928: Timeline Server Next Generation: Phase 1
YARN-3087. Made the REST server of per-node aggregator work alone in NM
daemon. (Li Lu via zjshen)
+ YARN-3125. Made the distributed shell use timeline service next gen and
+ add an integration test for it. (Junping Du and Li Lu via zjshen)
+
IMPROVEMENTS
OPTIMIZATIONS
http://git-wip-us.apache.org/repos/asf/hadoop/blob/bf08f7f0/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/src/main/java/org/apache/hadoop/yarn/applications/distributedshell/ApplicationMaster.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/src/main/java/org/apache/hadoop/yarn/applications/distributedshell/ApplicationMaster.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/src/main/java/org/apache/hadoop/yarn/applications/distributedshell/ApplicationMaster.java
index a9a7091..db49166 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/src/main/java/org/apache/hadoop/yarn/applications/distributedshell/ApplicationMaster.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/src/main/java/org/apache/hadoop/yarn/applications/distributedshell/ApplicationMaster.java
@@ -207,6 +207,8 @@ public class ApplicationMaster {
private int appMasterRpcPort = -1;
// Tracking url to which app master publishes info for clients to monitor
private String appMasterTrackingUrl = "";
+
+ private boolean newTimelineService = false;
// App Master configuration
// No. of containers to run shell command on
@@ -360,7 +362,8 @@ public class ApplicationMaster {
"No. of containers on which the shell command needs to be executed");
opts.addOption("priority", true, "Application Priority. Default 0");
opts.addOption("debug", false, "Dump out debug information");
-
+ opts.addOption("timeline_service_version", true,
+ "Version for timeline service");
opts.addOption("help", false, "Print usage");
CommandLine cliParser = new GnuParser().parse(opts, args);
@@ -499,13 +502,30 @@ public class ApplicationMaster {
if (conf.getBoolean(YarnConfiguration.TIMELINE_SERVICE_ENABLED,
YarnConfiguration.DEFAULT_TIMELINE_SERVICE_ENABLED)) {
+ if (cliParser.hasOption("timeline_service_version")) {
+ String timelineServiceVersion =
+ cliParser.getOptionValue("timeline_service_version", "v1");
+ if (timelineServiceVersion.trim().equalsIgnoreCase("v1")) {
+ newTimelineService = false;
+ } else if (timelineServiceVersion.trim().equalsIgnoreCase("v2")) {
+ newTimelineService = true;
+ } else {
+ throw new IllegalArgumentException(
+ "timeline_service_version is not set properly, should be 'v1' or 'v2'");
+ }
+ }
// Creating the Timeline Client
- timelineClient = TimelineClient.createTimelineClient();
+ timelineClient = TimelineClient.createTimelineClient(
+ appAttemptID.getApplicationId());
timelineClient.init(conf);
timelineClient.start();
} else {
timelineClient = null;
LOG.warn("Timeline service is not enabled");
+ if (cliParser.hasOption("timeline_service_version")) {
+ throw new IllegalArgumentException(
+ "Timeline service is not enabled");
+ }
}
return true;
@@ -556,8 +576,14 @@ public class ApplicationMaster {
appSubmitterUgi.addCredentials(credentials);
if(timelineClient != null) {
- publishApplicationAttemptEvent(timelineClient, appAttemptID.toString(),
- DSEvent.DS_APP_ATTEMPT_START, domainId, appSubmitterUgi);
+ if (newTimelineService) {
+ publishApplicationAttemptEventOnNewTimelineService(timelineClient,
+ appAttemptID.toString(), DSEvent.DS_APP_ATTEMPT_START, domainId,
+ appSubmitterUgi);
+ } else {
+ publishApplicationAttemptEvent(timelineClient, appAttemptID.toString(),
+ DSEvent.DS_APP_ATTEMPT_START, domainId, appSubmitterUgi);
+ }
}
AMRMClientAsync.CallbackHandler allocListener = new RMCallbackHandler();
@@ -626,8 +652,14 @@ public class ApplicationMaster {
numRequestedContainers.set(numTotalContainers);
if(timelineClient != null) {
- publishApplicationAttemptEvent(timelineClient, appAttemptID.toString(),
- DSEvent.DS_APP_ATTEMPT_END, domainId, appSubmitterUgi);
+ if (newTimelineService) {
+ publishApplicationAttemptEventOnNewTimelineService(timelineClient,
+ appAttemptID.toString(), DSEvent.DS_APP_ATTEMPT_START, domainId,
+ appSubmitterUgi);
+ } else {
+ publishApplicationAttemptEvent(timelineClient, appAttemptID.toString(),
+ DSEvent.DS_APP_ATTEMPT_START, domainId, appSubmitterUgi);
+ }
}
}
@@ -740,8 +772,13 @@ public class ApplicationMaster {
+ containerStatus.getContainerId());
}
if(timelineClient != null) {
- publishContainerEndEvent(
- timelineClient, containerStatus, domainId, appSubmitterUgi);
+ if (newTimelineService) {
+ publishContainerEndEventOnNewTimelineService(
+ timelineClient, containerStatus, domainId, appSubmitterUgi);
+ } else {
+ publishContainerEndEvent(
+ timelineClient, containerStatus, domainId, appSubmitterUgi);
+ }
}
}
@@ -858,9 +895,15 @@ public class ApplicationMaster {
applicationMaster.nmClientAsync.getContainerStatusAsync(containerId, container.getNodeId());
}
if(applicationMaster.timelineClient != null) {
- ApplicationMaster.publishContainerStartEvent(
- applicationMaster.timelineClient, container,
- applicationMaster.domainId, applicationMaster.appSubmitterUgi);
+ if (applicationMaster.newTimelineService) {
+ ApplicationMaster.publishContainerStartEventOnNewTimelineService(
+ applicationMaster.timelineClient, container,
+ applicationMaster.domainId, applicationMaster.appSubmitterUgi);
+ } else {
+ ApplicationMaster.publishContainerStartEvent(
+ applicationMaster.timelineClient, container,
+ applicationMaster.domainId, applicationMaster.appSubmitterUgi);
+ }
}
}
@@ -1147,4 +1190,102 @@ public class ApplicationMaster {
e instanceof UndeclaredThrowableException ? e.getCause() : e);
}
}
+
+ private static void publishContainerStartEventOnNewTimelineService(
+ final TimelineClient timelineClient, Container container, String domainId,
+ UserGroupInformation ugi) {
+ final org.apache.hadoop.yarn.api.records.timelineservice.TimelineEntity entity =
+ new org.apache.hadoop.yarn.api.records.timelineservice.TimelineEntity();
+ entity.setId(container.getId().toString());
+ entity.setType(DSEntity.DS_CONTAINER.toString());
+ //entity.setDomainId(domainId);
+ entity.addInfo("user", ugi.getShortUserName());
+
+ org.apache.hadoop.yarn.api.records.timelineservice.TimelineEvent event =
+ new org.apache.hadoop.yarn.api.records.timelineservice.TimelineEvent();
+ event.setTimestamp(System.currentTimeMillis());
+ event.setId(DSEvent.DS_CONTAINER_START.toString());
+ event.addInfo("Node", container.getNodeId().toString());
+ event.addInfo("Resources", container.getResource().toString());
+ entity.addEvent(event);
+
+ try {
+ ugi.doAs(new PrivilegedExceptionAction<Object>() {
+ @Override
+ public TimelinePutResponse run() throws Exception {
+ timelineClient.putEntities(entity);
+ return null;
+ }
+ });
+ } catch (Exception e) {
+ LOG.error("Container start event could not be published for "
+ + container.getId().toString(),
+ e instanceof UndeclaredThrowableException ? e.getCause() : e);
+ }
+ }
+
+ private static void publishContainerEndEventOnNewTimelineService(
+ final TimelineClient timelineClient, ContainerStatus container,
+ String domainId, UserGroupInformation ugi) {
+ final org.apache.hadoop.yarn.api.records.timelineservice.TimelineEntity entity =
+ new org.apache.hadoop.yarn.api.records.timelineservice.TimelineEntity();
+ entity.setId(container.getContainerId().toString());
+ entity.setType(DSEntity.DS_CONTAINER.toString());
+ //entity.setDomainId(domainId);
+ entity.addInfo("user", ugi.getShortUserName());
+ org.apache.hadoop.yarn.api.records.timelineservice.TimelineEvent event =
+ new org.apache.hadoop.yarn.api.records.timelineservice.TimelineEvent();
+ event.setTimestamp(System.currentTimeMillis());
+ event.setId(DSEvent.DS_CONTAINER_END.toString());
+ event.addInfo("State", container.getState().name());
+ event.addInfo("Exit Status", container.getExitStatus());
+ entity.addEvent(event);
+
+ try {
+ ugi.doAs(new PrivilegedExceptionAction<Object>() {
+ @Override
+ public TimelinePutResponse run() throws Exception {
+ timelineClient.putEntities(entity);
+ return null;
+ }
+ });
+ } catch (Exception e) {
+ LOG.error("Container end event could not be published for "
+ + container.getContainerId().toString(),
+ e instanceof UndeclaredThrowableException ? e.getCause() : e);
+ }
+ }
+
+ private static void publishApplicationAttemptEventOnNewTimelineService(
+ final TimelineClient timelineClient, String appAttemptId,
+ DSEvent appEvent, String domainId, UserGroupInformation ugi) {
+ final org.apache.hadoop.yarn.api.records.timelineservice.TimelineEntity entity =
+ new org.apache.hadoop.yarn.api.records.timelineservice.TimelineEntity();
+ entity.setId(appAttemptId);
+ entity.setType(DSEntity.DS_APP_ATTEMPT.toString());
+ //entity.setDomainId(domainId);
+ entity.addInfo("user", ugi.getShortUserName());
+ org.apache.hadoop.yarn.api.records.timelineservice.TimelineEvent event =
+ new org.apache.hadoop.yarn.api.records.timelineservice.TimelineEvent();
+ event.setId(appEvent.toString());
+ event.setTimestamp(System.currentTimeMillis());
+ entity.addEvent(event);
+
+ try {
+ ugi.doAs(new PrivilegedExceptionAction<Object>() {
+ @Override
+ public TimelinePutResponse run() throws Exception {
+ timelineClient.putEntities(entity);
+ return null;
+ }
+ });
+ } catch (Exception e) {
+ LOG.error("App Attempt "
+ + (appEvent.equals(DSEvent.DS_APP_ATTEMPT_START) ? "start" : "end")
+ + " event could not be published for "
+ + appAttemptId.toString(),
+ e instanceof UndeclaredThrowableException ? e.getCause() : e);
+ }
+ }
+
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/bf08f7f0/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/src/main/java/org/apache/hadoop/yarn/applications/distributedshell/Client.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/src/main/java/org/apache/hadoop/yarn/applications/distributedshell/Client.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/src/main/java/org/apache/hadoop/yarn/applications/distributedshell/Client.java
index 0e9a4e4..02a627e 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/src/main/java/org/apache/hadoop/yarn/applications/distributedshell/Client.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/src/main/java/org/apache/hadoop/yarn/applications/distributedshell/Client.java
@@ -185,6 +185,8 @@ public class Client {
// Command line options
private Options opts;
+
+ private String timelineServiceVersion;
private static final String shellCommandPath = "shellCommands";
private static final String shellArgsPath = "shellArgs";
@@ -260,6 +262,7 @@ public class Client {
opts.addOption("container_vcores", true, "Amount of virtual cores to be requested to run the shell command");
opts.addOption("num_containers", true, "No. of containers on which the shell command needs to be executed");
opts.addOption("log_properties", true, "log4j.properties file");
+ opts.addOption("timeline_service_version", true, "Version for timeline service");
opts.addOption("keep_containers_across_application_attempts", false,
"Flag to indicate whether to keep containers across application attempts." +
" If the flag is true, running containers will not be killed when" +
@@ -354,6 +357,16 @@ public class Client {
throw new IllegalArgumentException("Invalid virtual cores specified for application master, exiting."
+ " Specified virtual cores=" + amVCores);
}
+
+ if (cliParser.hasOption("timeline_service_version")) {
+ timelineServiceVersion =
+ cliParser.getOptionValue("timeline_service_version", "v1");
+ if (! (timelineServiceVersion.trim().equalsIgnoreCase("v1") ||
+ timelineServiceVersion.trim().equalsIgnoreCase("v2"))) {
+ throw new IllegalArgumentException(
+ "timeline_service_version is not set properly, should be 'v1' or 'v2'");
+ }
+ }
if (!cliParser.hasOption("jar")) {
throw new IllegalArgumentException("No jar file specified for application master");
@@ -633,11 +646,14 @@ public class Client {
for (Map.Entry<String, String> entry : shellEnv.entrySet()) {
vargs.add("--shell_env " + entry.getKey() + "=" + entry.getValue());
- }
+ }
if (debugFlag) {
vargs.add("--debug");
}
+ if (timelineServiceVersion != null) {
+ vargs.add("--timeline_service_version " + timelineServiceVersion);
+ }
vargs.add("1>" + ApplicationConstants.LOG_DIR_EXPANSION_VAR + "/AppMaster.stdout");
vargs.add("2>" + ApplicationConstants.LOG_DIR_EXPANSION_VAR + "/AppMaster.stderr");
@@ -647,7 +663,7 @@ public class Client {
command.append(str).append(" ");
}
- LOG.info("Completed setting up app master command " + command.toString());
+ LOG.info("Completed setting up app master command " + command.toString());
List<String> commands = new ArrayList<String>();
commands.add(command.toString());
http://git-wip-us.apache.org/repos/asf/hadoop/blob/bf08f7f0/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/src/test/java/org/apache/hadoop/yarn/applications/distributedshell/TestDistributedShell.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/src/test/java/org/apache/hadoop/yarn/applications/distributedshell/TestDistributedShell.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/src/test/java/org/apache/hadoop/yarn/applications/distributedshell/TestDistributedShell.java
index 46b5850..71466cb 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/src/test/java/org/apache/hadoop/yarn/applications/distributedshell/TestDistributedShell.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/src/test/java/org/apache/hadoop/yarn/applications/distributedshell/TestDistributedShell.java
@@ -49,10 +49,13 @@ import org.apache.hadoop.yarn.client.api.YarnClient;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.server.MiniYARNCluster;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacityScheduler;
+import org.apache.hadoop.yarn.server.timelineservice.aggregator.PerNodeAggregatorServer;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
+import org.junit.Rule;
import org.junit.Test;
+import org.junit.rules.TestName;
public class TestDistributedShell {
@@ -62,10 +65,14 @@ public class TestDistributedShell {
protected MiniYARNCluster yarnCluster = null;
protected YarnConfiguration conf = null;
private static final int NUM_NMS = 1;
+ private static final String TIMELINE_AUX_SERVICE_NAME = "timeline_aggregator";
protected final static String APPMASTER_JAR =
JarFinder.getJar(ApplicationMaster.class);
+ @Rule
+ public TestName currTestName= new TestName();
+
@Before
public void setup() throws Exception {
setupInternal(NUM_NMS);
@@ -74,18 +81,30 @@ public class TestDistributedShell {
protected void setupInternal(int numNodeManager) throws Exception {
LOG.info("Starting up YARN cluster");
-
+
conf = new YarnConfiguration();
conf.setInt(YarnConfiguration.RM_SCHEDULER_MINIMUM_ALLOCATION_MB, 128);
conf.set("yarn.log.dir", "target");
conf.setBoolean(YarnConfiguration.TIMELINE_SERVICE_ENABLED, true);
+ // mark if we need to launch the v1 timeline server
+ boolean enableATSV1 = false;
+ if (!currTestName.getMethodName().toLowerCase().contains("v2")) {
+ // disable aux-service based timeline aggregators
+ conf.set(YarnConfiguration.NM_AUX_SERVICES, "");
+ enableATSV1 = true;
+ } else {
+ // enable aux-service based timeline aggregators
+ conf.set(YarnConfiguration.NM_AUX_SERVICES, TIMELINE_AUX_SERVICE_NAME);
+ conf.set(YarnConfiguration.NM_AUX_SERVICES + "." + TIMELINE_AUX_SERVICE_NAME
+ + ".class", PerNodeAggregatorServer.class.getName());
+ }
conf.set(YarnConfiguration.RM_SCHEDULER, CapacityScheduler.class.getName());
conf.setBoolean(YarnConfiguration.NODE_LABELS_ENABLED, true);
-
+
if (yarnCluster == null) {
yarnCluster =
new MiniYARNCluster(TestDistributedShell.class.getSimpleName(), 1,
- numNodeManager, 1, 1, true);
+ numNodeManager, 1, 1, enableATSV1);
yarnCluster.init(conf);
yarnCluster.start();
@@ -140,15 +159,21 @@ public class TestDistributedShell {
@Test(timeout=90000)
public void testDSShellWithDomain() throws Exception {
- testDSShell(true);
+ testDSShell(true, "v1");
}
@Test(timeout=90000)
public void testDSShellWithoutDomain() throws Exception {
- testDSShell(false);
+ testDSShell(false, "v1");
}
- public void testDSShell(boolean haveDomain) throws Exception {
+ @Test(timeout=90000)
+ public void testDSShellWithoutDomainV2() throws Exception {
+ testDSShell(false, "v2");
+ }
+
+ public void testDSShell(boolean haveDomain, String timelineVersion)
+ throws Exception {
String[] args = {
"--jar",
APPMASTER_JAR,
@@ -175,9 +200,17 @@ public class TestDistributedShell {
"writer_user writer_group",
"--create"
};
- List<String> argsList = new ArrayList<String>(Arrays.asList(args));
- argsList.addAll(Arrays.asList(domainArgs));
- args = argsList.toArray(new String[argsList.size()]);
+ args = mergeArgs(args, domainArgs);
+ }
+ boolean isTestingTimelineV2 = false;
+ if (timelineVersion.equalsIgnoreCase("v2")) {
+ String[] timelineArgs = {
+ "--timeline_service_version",
+ "v2"
+ };
+ isTestingTimelineV2 = true;
+ args = mergeArgs(args, timelineArgs);
+ LOG.info("Setup: Using timeline v2!");
}
LOG.info("Initializing DS Client");
@@ -231,7 +264,15 @@ public class TestDistributedShell {
LOG.info("Client run completed. Result=" + result);
Assert.assertTrue(result.get());
- TimelineDomain domain = null;
+ if (!isTestingTimelineV2) {
+ checkTimelineV1(haveDomain);
+ } else {
+ checkTimelineV2(haveDomain);
+ }
+ }
+
+ private void checkTimelineV1(boolean haveDomain) throws Exception {
+ TimelineDomain domain = null;
if (haveDomain) {
domain = yarnCluster.getApplicationHistoryServer()
.getTimelineStore().getDomain("TEST_DOMAIN");
@@ -275,6 +316,24 @@ public class TestDistributedShell {
}
}
+ private void checkTimelineV2(boolean haveDomain) {
+ // TODO check timeline V2 here after we have a storage layer
+ }
+
+ /**
+ * Utility function to merge two String arrays to form a new String array for
+ * our argumemts.
+ *
+ * @param args
+ * @param newArgs
+ * @return a String array consists of {args, newArgs}
+ */
+ private String[] mergeArgs(String[] args, String[] newArgs) {
+ List<String> argsList = new ArrayList<String>(Arrays.asList(args));
+ argsList.addAll(Arrays.asList(newArgs));
+ return argsList.toArray(new String[argsList.size()]);
+ }
+
/*
* NetUtils.getHostname() returns a string in the form "hostname/ip".
* Sometimes the hostname we get is the FQDN and sometimes the short name. In
http://git-wip-us.apache.org/repos/asf/hadoop/blob/bf08f7f0/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/aggregator/BaseAggregatorService.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/aggregator/BaseAggregatorService.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/aggregator/BaseAggregatorService.java
index 46e5574..e362139 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/aggregator/BaseAggregatorService.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/aggregator/BaseAggregatorService.java
@@ -72,6 +72,12 @@ public class BaseAggregatorService extends CompositeService {
*/
public void postEntities(TimelineEntities entities,
UserGroupInformation callerUgi) {
+ // Add this output temporarily for our prototype
+ // TODO remove this after we have an actual implementation
+ LOG.info("SUCCESS - TIMELINE V2 PROTOTYPE");
+ LOG.info("postEntities(entities=" + entities + ", callerUgi=" +
+ callerUgi + ")");
+
// TODO implement
if (LOG.isDebugEnabled()) {
LOG.debug("postEntities(entities=" + entities + ", callerUgi=" +
http://git-wip-us.apache.org/repos/asf/hadoop/blob/bf08f7f0/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/aggregator/PerNodeAggregatorServer.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/aggregator/PerNodeAggregatorServer.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/aggregator/PerNodeAggregatorServer.java
index 55c6271..deb21c7 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/aggregator/PerNodeAggregatorServer.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/aggregator/PerNodeAggregatorServer.java
@@ -118,8 +118,8 @@ public class PerNodeAggregatorServer extends AuxiliaryService {
.setConf(conf)
.addEndpoint(URI.create("http://" + bindAddress));
timelineRestServer = builder.build();
- // TODO: replace this by an authentification filter in future.
- HashMap<String, String> options = new HashMap<String, String>();
+ // TODO: replace this by an authentication filter in future.
+ HashMap<String, String> options = new HashMap<>();
String username = conf.get(HADOOP_HTTP_STATIC_USER,
DEFAULT_HADOOP_HTTP_STATIC_USER);
options.put(HADOOP_HTTP_STATIC_USER, username);
[4/8] hadoop git commit: HDFS-6753. Initialize checkDisk when
DirectoryScanner not able to get files list for scanning (Contributed by
J.Andreina)
Posted by zj...@apache.org.
HDFS-6753. Initialize checkDisk when DirectoryScanner not able to get files list for scanning (Contributed by J.Andreina)
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/4f75b156
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/4f75b156
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/4f75b156
Branch: refs/heads/YARN-2928
Commit: 4f75b15628a76881efc39054612dc128e23d27be
Parents: 2954e65
Author: Vinayakumar B <vi...@apache.org>
Authored: Fri Feb 27 16:36:28 2015 +0530
Committer: Vinayakumar B <vi...@apache.org>
Committed: Fri Feb 27 16:36:28 2015 +0530
----------------------------------------------------------------------
hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt | 3 +++
.../apache/hadoop/hdfs/server/datanode/DataNode.java | 2 +-
.../hadoop/hdfs/server/datanode/DirectoryScanner.java | 12 +++++++++---
.../hdfs/server/datanode/TestDirectoryScanner.java | 9 ++++++---
4 files changed, 19 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/4f75b156/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt b/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
index ba553dc..8556afd 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
+++ b/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
@@ -1040,6 +1040,9 @@ Release 2.7.0 - UNRELEASED
HDFS-7774. Unresolved symbols error while compiling HDFS on Windows 7/32 bit.
(Kiran Kumar M R via cnauroth)
+ HDFS-6753. Initialize checkDisk when DirectoryScanner not able to get
+ files list for scanning (J.Andreina via vinayakumarb)
+
BREAKDOWN OF HDFS-7584 SUBTASKS AND RELATED JIRAS
HDFS-7720. Quota by Storage Type API, tools and ClientNameNode
http://git-wip-us.apache.org/repos/asf/hadoop/blob/4f75b156/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java
index f233e02..92ddb7b 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java
@@ -815,7 +815,7 @@ public class DataNode extends ReconfigurableBase
reason = "verifcation is not supported by SimulatedFSDataset";
}
if (reason == null) {
- directoryScanner = new DirectoryScanner(data, conf);
+ directoryScanner = new DirectoryScanner(this, data, conf);
directoryScanner.start();
} else {
LOG.info("Periodic Directory Tree Verification scan is disabled because " +
http://git-wip-us.apache.org/repos/asf/hadoop/blob/4f75b156/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DirectoryScanner.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DirectoryScanner.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DirectoryScanner.java
index 09c2914..c7ee21e 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DirectoryScanner.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DirectoryScanner.java
@@ -63,6 +63,7 @@ public class DirectoryScanner implements Runnable {
private final long scanPeriodMsecs;
private volatile boolean shouldRun = false;
private boolean retainDiffs = false;
+ private final DataNode datanode;
final ScanInfoPerBlockPool diffs = new ScanInfoPerBlockPool();
final Map<String, Stats> stats = new HashMap<String, Stats>();
@@ -308,7 +309,8 @@ public class DirectoryScanner implements Runnable {
}
}
- DirectoryScanner(FsDatasetSpi<?> dataset, Configuration conf) {
+ DirectoryScanner(DataNode datanode, FsDatasetSpi<?> dataset, Configuration conf) {
+ this.datanode = datanode;
this.dataset = dataset;
int interval = conf.getInt(DFSConfigKeys.DFS_DATANODE_DIRECTORYSCAN_INTERVAL_KEY,
DFSConfigKeys.DFS_DATANODE_DIRECTORYSCAN_INTERVAL_DEFAULT);
@@ -547,7 +549,7 @@ public class DirectoryScanner implements Runnable {
for (int i = 0; i < volumes.size(); i++) {
if (isValid(dataset, volumes.get(i))) {
ReportCompiler reportCompiler =
- new ReportCompiler(volumes.get(i));
+ new ReportCompiler(datanode,volumes.get(i));
Future<ScanInfoPerBlockPool> result =
reportCompileThreadPool.submit(reportCompiler);
compilersInProgress.put(i, result);
@@ -585,8 +587,10 @@ public class DirectoryScanner implements Runnable {
private static class ReportCompiler
implements Callable<ScanInfoPerBlockPool> {
private final FsVolumeSpi volume;
+ private final DataNode datanode;
- public ReportCompiler(FsVolumeSpi volume) {
+ public ReportCompiler(DataNode datanode, FsVolumeSpi volume) {
+ this.datanode = datanode;
this.volume = volume;
}
@@ -611,6 +615,8 @@ public class DirectoryScanner implements Runnable {
files = FileUtil.listFiles(dir);
} catch (IOException ioe) {
LOG.warn("Exception occured while compiling report: ", ioe);
+ // Initiate a check on disk failure.
+ datanode.checkDiskErrorAsync();
// Ignore this directory and proceed.
return report;
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/4f75b156/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDirectoryScanner.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDirectoryScanner.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDirectoryScanner.java
index 681768e..0e0e959 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDirectoryScanner.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDirectoryScanner.java
@@ -299,10 +299,11 @@ public class TestDirectoryScanner {
.build();
try {
cluster.waitActive();
+ DataNode dataNode = cluster.getDataNodes().get(0);
bpid = cluster.getNamesystem().getBlockPoolId();
fds = DataNodeTestUtils.getFSDataset(cluster.getDataNodes().get(0));
client = cluster.getFileSystem().getClient();
- scanner = new DirectoryScanner(fds, CONF);
+ scanner = new DirectoryScanner(dataNode, fds, CONF);
scanner.setRetainDiffs(true);
FsDatasetTestUtil.stopLazyWriter(cluster.getDataNodes().get(0));
@@ -340,9 +341,10 @@ public class TestDirectoryScanner {
try {
cluster.waitActive();
bpid = cluster.getNamesystem().getBlockPoolId();
+ DataNode dataNode = cluster.getDataNodes().get(0);
fds = DataNodeTestUtils.getFSDataset(cluster.getDataNodes().get(0));
client = cluster.getFileSystem().getClient();
- scanner = new DirectoryScanner(fds, CONF);
+ scanner = new DirectoryScanner(dataNode, fds, CONF);
scanner.setRetainDiffs(true);
FsDatasetTestUtil.stopLazyWriter(cluster.getDataNodes().get(0));
@@ -389,7 +391,8 @@ public class TestDirectoryScanner {
client = cluster.getFileSystem().getClient();
CONF.setInt(DFSConfigKeys.DFS_DATANODE_DIRECTORYSCAN_THREADS_KEY,
parallelism);
- scanner = new DirectoryScanner(fds, CONF);
+ DataNode dataNode = cluster.getDataNodes().get(0);
+ scanner = new DirectoryScanner(dataNode, fds, CONF);
scanner.setRetainDiffs(true);
// Add files with 100 blocks
[2/8] hadoop git commit: HADOOP-11569. Provide Merge API for MapFile
to merge multiple similar MapFiles to one MapFile. Contributed by Vinayakumar
B.
Posted by zj...@apache.org.
HADOOP-11569. Provide Merge API for MapFile to merge multiple similar MapFiles to one MapFile. Contributed by Vinayakumar B.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/48c7ee75
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/48c7ee75
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/48c7ee75
Branch: refs/heads/YARN-2928
Commit: 48c7ee7553af94a57952bca03b49c04b9bbfab45
Parents: 8ca0d95
Author: Tsuyoshi Ozawa <oz...@apache.org>
Authored: Fri Feb 27 17:46:07 2015 +0900
Committer: Tsuyoshi Ozawa <oz...@apache.org>
Committed: Fri Feb 27 17:46:07 2015 +0900
----------------------------------------------------------------------
hadoop-common-project/hadoop-common/CHANGES.txt | 3 +
.../main/java/org/apache/hadoop/io/MapFile.java | 143 +++++++++++++++++++
.../java/org/apache/hadoop/io/TestMapFile.java | 56 ++++++++
3 files changed, 202 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/48c7ee75/hadoop-common-project/hadoop-common/CHANGES.txt
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/CHANGES.txt b/hadoop-common-project/hadoop-common/CHANGES.txt
index 1d9a6d4..6d4da77 100644
--- a/hadoop-common-project/hadoop-common/CHANGES.txt
+++ b/hadoop-common-project/hadoop-common/CHANGES.txt
@@ -445,6 +445,9 @@ Release 2.7.0 - UNRELEASED
HADOOP-11510. Expose truncate API via FileContext. (yliu)
+ HADOOP-11569. Provide Merge API for MapFile to merge multiple similar MapFiles
+ to one MapFile. (Vinayakumar B via ozawa)
+
IMPROVEMENTS
HADOOP-11483. HardLink.java should use the jdk7 createLink method (aajisaka)
http://git-wip-us.apache.org/repos/asf/hadoop/blob/48c7ee75/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/MapFile.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/MapFile.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/MapFile.java
index 84c9dcc..ee76458 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/MapFile.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/MapFile.java
@@ -25,6 +25,7 @@ import java.util.Arrays;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.HadoopIllegalArgumentException;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.conf.Configuration;
@@ -824,6 +825,148 @@ public class MapFile {
return cnt;
}
+ /**
+ * Class to merge multiple MapFiles of same Key and Value types to one MapFile
+ */
+ public static class Merger {
+ private Configuration conf;
+ private WritableComparator comparator = null;
+ private Reader[] inReaders;
+ private Writer outWriter;
+ private Class<Writable> valueClass = null;
+ private Class<WritableComparable> keyClass = null;
+
+ public Merger(Configuration conf) throws IOException {
+ this.conf = conf;
+ }
+
+ /**
+ * Merge multiple MapFiles to one Mapfile
+ *
+ * @param inMapFiles
+ * @param outMapFile
+ * @throws IOException
+ */
+ public void merge(Path[] inMapFiles, boolean deleteInputs,
+ Path outMapFile) throws IOException {
+ try {
+ open(inMapFiles, outMapFile);
+ mergePass();
+ } finally {
+ close();
+ }
+ if (deleteInputs) {
+ for (int i = 0; i < inMapFiles.length; i++) {
+ Path path = inMapFiles[i];
+ delete(path.getFileSystem(conf), path.toString());
+ }
+ }
+ }
+
+ /*
+ * Open all input files for reading and verify the key and value types. And
+ * open Output file for writing
+ */
+ @SuppressWarnings("unchecked")
+ private void open(Path[] inMapFiles, Path outMapFile) throws IOException {
+ inReaders = new Reader[inMapFiles.length];
+ for (int i = 0; i < inMapFiles.length; i++) {
+ Reader reader = new Reader(inMapFiles[i], conf);
+ if (keyClass == null || valueClass == null) {
+ keyClass = (Class<WritableComparable>) reader.getKeyClass();
+ valueClass = (Class<Writable>) reader.getValueClass();
+ } else if (keyClass != reader.getKeyClass()
+ || valueClass != reader.getValueClass()) {
+ throw new HadoopIllegalArgumentException(
+ "Input files cannot be merged as they"
+ + " have different Key and Value classes");
+ }
+ inReaders[i] = reader;
+ }
+
+ if (comparator == null) {
+ Class<? extends WritableComparable> cls;
+ cls = keyClass.asSubclass(WritableComparable.class);
+ this.comparator = WritableComparator.get(cls, conf);
+ } else if (comparator.getKeyClass() != keyClass) {
+ throw new HadoopIllegalArgumentException(
+ "Input files cannot be merged as they"
+ + " have different Key class compared to"
+ + " specified comparator");
+ }
+
+ outWriter = new MapFile.Writer(conf, outMapFile,
+ MapFile.Writer.keyClass(keyClass),
+ MapFile.Writer.valueClass(valueClass));
+ }
+
+ /**
+ * Merge all input files to output map file.<br>
+ * 1. Read first key/value from all input files to keys/values array. <br>
+ * 2. Select the least key and corresponding value. <br>
+ * 3. Write the selected key and value to output file. <br>
+ * 4. Replace the already written key/value in keys/values arrays with the
+ * next key/value from the selected input <br>
+ * 5. Repeat step 2-4 till all keys are read. <br>
+ */
+ private void mergePass() throws IOException {
+ // re-usable array
+ WritableComparable[] keys = new WritableComparable[inReaders.length];
+ Writable[] values = new Writable[inReaders.length];
+ // Read first key/value from all inputs
+ for (int i = 0; i < inReaders.length; i++) {
+ keys[i] = ReflectionUtils.newInstance(keyClass, null);
+ values[i] = ReflectionUtils.newInstance(valueClass, null);
+ if (!inReaders[i].next(keys[i], values[i])) {
+ // Handle empty files
+ keys[i] = null;
+ values[i] = null;
+ }
+ }
+
+ do {
+ int currentEntry = -1;
+ WritableComparable currentKey = null;
+ Writable currentValue = null;
+ for (int i = 0; i < keys.length; i++) {
+ if (keys[i] == null) {
+ // Skip Readers reached EOF
+ continue;
+ }
+ if (currentKey == null || comparator.compare(currentKey, keys[i]) > 0) {
+ currentEntry = i;
+ currentKey = keys[i];
+ currentValue = values[i];
+ }
+ }
+ if (currentKey == null) {
+ // Merge Complete
+ break;
+ }
+ // Write the selected key/value to merge stream
+ outWriter.append(currentKey, currentValue);
+ // Replace the already written key/value in keys/values arrays with the
+ // next key/value from the selected input
+ if (!inReaders[currentEntry].next(keys[currentEntry],
+ values[currentEntry])) {
+ // EOF for this file
+ keys[currentEntry] = null;
+ values[currentEntry] = null;
+ }
+ } while (true);
+ }
+
+ private void close() throws IOException {
+ for (int i = 0; i < inReaders.length; i++) {
+ IOUtils.closeStream(inReaders[i]);
+ inReaders[i] = null;
+ }
+ if (outWriter != null) {
+ outWriter.close();
+ outWriter = null;
+ }
+ }
+ }
public static void main(String[] args) throws Exception {
String usage = "Usage: MapFile inFile outFile";
http://git-wip-us.apache.org/repos/asf/hadoop/blob/48c7ee75/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/io/TestMapFile.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/io/TestMapFile.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/io/TestMapFile.java
index ced74fb..3f14de0 100644
--- a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/io/TestMapFile.java
+++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/io/TestMapFile.java
@@ -21,6 +21,10 @@ import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
@@ -730,4 +734,56 @@ public class TestMapFile {
reader.close();
}
}
+
+ @Test
+ public void testMerge() throws Exception {
+ final String TEST_METHOD_KEY = "testMerge.mapfile";
+ int SIZE = 10;
+ int ITERATIONS = 5;
+ Path[] in = new Path[5];
+ List<Integer> expected = new ArrayList<Integer>();
+ for (int j = 0; j < 5; j++) {
+ try (MapFile.Writer writer = createWriter(TEST_METHOD_KEY + "." + j,
+ IntWritable.class, Text.class)) {
+ in[j] = new Path(TEST_DIR, TEST_METHOD_KEY + "." + j);
+ for (int i = 0; i < SIZE; i++) {
+ expected.add(i + j);
+ writer.append(new IntWritable(i + j), new Text("Value:" + (i + j)));
+ }
+ }
+ }
+ // Sort expected values
+ Collections.sort(expected);
+ // Merge all 5 files
+ MapFile.Merger merger = new MapFile.Merger(conf);
+ merger.merge(in, true, new Path(TEST_DIR, TEST_METHOD_KEY));
+
+ try (MapFile.Reader reader = createReader(TEST_METHOD_KEY,
+ IntWritable.class)) {
+ int start = 0;
+ // test iteration
+ Text startValue = new Text("Value:" + start);
+ int i = 0;
+ while (i++ < ITERATIONS) {
+ Iterator<Integer> expectedIterator = expected.iterator();
+ IntWritable key = new IntWritable(start);
+ Text value = startValue;
+ IntWritable prev = new IntWritable(start);
+ while (reader.next(key, value)) {
+ assertTrue("Next key should be always equal or more",
+ prev.get() <= key.get());
+ assertEquals(expectedIterator.next().intValue(), key.get());
+ prev.set(key.get());
+ }
+ reader.reset();
+ }
+ }
+
+ // inputs should be deleted
+ for (int j = 0; j < in.length; j++) {
+ Path path = in[j];
+ assertFalse("inputs should be deleted",
+ path.getFileSystem(conf).exists(path));
+ }
+ }
}
[3/8] hadoop git commit: Revert "HDFS-7769. TestHDFSCLI should not
create files in hdfs project root dir."
Posted by zj...@apache.org.
Revert "HDFS-7769. TestHDFSCLI should not create files in hdfs project root dir."
This reverts commit 7c6b6547eeed110e1a842e503bfd33afe04fa814.
Conflicts:
hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/2954e654
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/2954e654
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/2954e654
Branch: refs/heads/YARN-2928
Commit: 2954e654677bd1807d22fae7becc4464d9eff00b
Parents: 48c7ee7
Author: Tsz-Wo Nicholas Sze <sz...@hortonworks.com>
Authored: Fri Feb 27 18:25:32 2015 +0800
Committer: Tsz-Wo Nicholas Sze <sz...@hortonworks.com>
Committed: Fri Feb 27 18:25:32 2015 +0800
----------------------------------------------------------------------
hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt | 3 ---
.../hadoop-hdfs/src/test/resources/testHDFSConf.xml | 4 ++--
2 files changed, 2 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/2954e654/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt b/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
index ae83898..ba553dc 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
+++ b/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
@@ -975,9 +975,6 @@ Release 2.7.0 - UNRELEASED
HDFS-7714. Simultaneous restart of HA NameNodes and DataNode can cause
DataNode to register successfully with only one NameNode.(vinayakumarb)
- HDFS-7769. TestHDFSCLI should not create files in hdfs project root dir.
- (szetszwo)
-
HDFS-7753. Fix Multithreaded correctness Warnings in BackupImage.
(Rakesh R and shv)
http://git-wip-us.apache.org/repos/asf/hadoop/blob/2954e654/hadoop-hdfs-project/hadoop-hdfs/src/test/resources/testHDFSConf.xml
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/resources/testHDFSConf.xml b/hadoop-hdfs-project/hadoop-hdfs/src/test/resources/testHDFSConf.xml
index 2d3de1f..e59b05a 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/resources/testHDFSConf.xml
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/resources/testHDFSConf.xml
@@ -16483,8 +16483,8 @@
<command>-fs NAMENODE -mkdir -p /user/USERNAME/dir1</command>
<command>-fs NAMENODE -copyFromLocal CLITEST_DATA/data15bytes /user/USERNAME/dir1</command>
<command>-fs NAMENODE -copyFromLocal CLITEST_DATA/data30bytes /user/USERNAME/dir1</command>
- <command>-fs NAMENODE -getmerge /user/USERNAME/dir1 CLITEST_DATA/file</command>
- <command>-cat CLITEST_DATA/file</command>
+ <command>-fs NAMENODE -getmerge /user/USERNAME/dir1 data</command>
+ <command>-cat data</command>
</test-commands>
<cleanup-commands>
<command>-fs NAMENODE -rm -r /user/USERNAME</command>
[7/8] hadoop git commit: Merge remote-tracking branch 'apache/trunk'
into YARN-2928
Posted by zj...@apache.org.
Merge remote-tracking branch 'apache/trunk' into YARN-2928
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/667f69c3
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/667f69c3
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/667f69c3
Branch: refs/heads/YARN-2928
Commit: 667f69c3627f12b45ee398e70a9055b5e31a8a86
Parents: 41a08ad 01a1621
Author: Zhijie Shen <zj...@apache.org>
Authored: Fri Feb 27 08:42:37 2015 -0800
Committer: Zhijie Shen <zj...@apache.org>
Committed: Fri Feb 27 08:42:37 2015 -0800
----------------------------------------------------------------------
hadoop-common-project/hadoop-common/CHANGES.txt | 3 +
.../main/java/org/apache/hadoop/io/MapFile.java | 143 +++++++++
.../java/org/apache/hadoop/io/TestMapFile.java | 56 ++++
hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt | 9 +-
.../org/apache/hadoop/hdfs/DFSOutputStream.java | 3 +-
.../hadoop/hdfs/server/datanode/DataNode.java | 2 +-
.../hdfs/server/datanode/DirectoryScanner.java | 12 +-
.../apache/hadoop/hdfs/TestDFSOutputStream.java | 31 ++
.../server/datanode/TestDirectoryScanner.java | 9 +-
.../src/test/resources/testHDFSConf.xml | 4 +-
.../mapreduce/v2/hs/JobHistoryServer.java | 2 +
hadoop-yarn-project/CHANGES.txt | 6 +
.../hadoop/yarn/conf/YarnConfiguration.java | 9 +
.../src/main/resources/yarn-default.xml | 15 +
.../yarn/server/nodemanager/NodeManager.java | 4 +-
.../server/resourcemanager/ResourceManager.java | 3 +
.../recovery/FileSystemRMStateStore.java | 303 ++++++++++++++-----
.../recovery/TestFSRMStateStore.java | 5 +
.../yarn/server/webproxy/WebAppProxyServer.java | 2 +
19 files changed, 537 insertions(+), 84 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/667f69c3/hadoop-yarn-project/CHANGES.txt
----------------------------------------------------------------------