Apache
/
hadoop
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

YARN-2122. In AllocationFileLoaderService, the reloadThread should be created in init() and started in start(). (Robert Kanter via kasha) 

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/trunk@1601045 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Karthik Kambatla 2014-06-07 00:49:20 +00:00
parent 9528ed939d
commit 969bef3c77
3 changed files with 52 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
hadoop-yarn-project/CHANGES.txt
View File

@ -139,6 +139,9 @@ Release 2.5.0 - UNRELEASED
YARN-1977. Add tests on getApplicationRequest with filtering start time range. (junping_du)
YARN-2122. In AllocationFileLoaderService, the reloadThread should be created
in init() and started in start(). (Robert Kanter via kasha)
OPTIMIZATIONS
BUG FIXES
@ -165,6 +168,7 @@ Release 2.5.0 - UNRELEASED
YARN-2011. Fix typo and warning in TestLeafQueue (Chen He via junping_du)
YARN-2042. String shouldn't be compared using == in
QueuePlacementRule#NestedUserQueue#getQueueForApp (Chen He via Sandy Ryza)

5
hadoop-yarn-project/hadoop-yarn/dev-support/findbugs-exclude.xml
View File

@ -142,6 +142,11 @@
<Class name="org.apache.hadoop.yarn.server.nodemanager.containermanager.logaggregation.LogAggregationService" />
<Bug pattern="IS2_INCONSISTENT_SYNC" />
</Match>
<Match>
<Class name="org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.AllocationFileLoaderService" />
<Field name="allocFile" />
<Bug pattern="IS2_INCONSISTENT_SYNC" />
</Match>
<!-- Inconsistent sync warning - minimumAllocation is only initialized once and never changed -->
<Match>
<Class name="org.apache.hadoop.yarn.server.resourcemanager.scheduler.AbstractYarnScheduler" />

82
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/AllocationFileLoaderService.java
View File

@ -98,55 +98,59 @@ public class AllocationFileLoaderService extends AbstractService {
}
@Override
public void init(Configuration conf) {
public void serviceInit(Configuration conf) throws Exception {
this.allocFile = getAllocationFile(conf);
super.init(conf);
}
@Override
public void start() {
if (allocFile == null) {
return;
}
reloadThread = new Thread() {
public void run() {
while (running) {
long time = clock.getTime();
long lastModified = allocFile.lastModified();
if (lastModified > lastSuccessfulReload &&
time > lastModified + ALLOC_RELOAD_WAIT_MS) {
try {
reloadAllocations();
} catch (Exception ex) {
if (allocFile != null) {
reloadThread = new Thread() {
@Override
public void run() {
while (running) {
long time = clock.getTime();
long lastModified = allocFile.lastModified();
if (lastModified > lastSuccessfulReload &&
time > lastModified + ALLOC_RELOAD_WAIT_MS) {
try {
reloadAllocations();
} catch (Exception ex) {
if (!lastReloadAttemptFailed) {
LOG.error("Failed to reload fair scheduler config file - " +
"will use existing allocations.", ex);
}
lastReloadAttemptFailed = true;
}
} else if (lastModified == 0l) {
if (!lastReloadAttemptFailed) {
LOG.error("Failed to reload fair scheduler config file - " +
"will use existing allocations.", ex);
LOG.warn("Failed to reload fair scheduler config file because" +
" last modified returned 0. File exists: "
+ allocFile.exists());
}
lastReloadAttemptFailed = true;
}
} else if (lastModified == 0l) {
if (!lastReloadAttemptFailed) {
LOG.warn("Failed to reload fair scheduler config file because" +
" last modified returned 0. File exists: " + allocFile.exists());
try {
Thread.sleep(reloadIntervalMs);
} catch (InterruptedException ex) {
LOG.info(
"Interrupted while waiting to reload alloc configuration");
}
lastReloadAttemptFailed = true;
}
try {
Thread.sleep(reloadIntervalMs);
} catch (InterruptedException ex) {
LOG.info("Interrupted while waiting to reload alloc configuration");
}
}
}
};
reloadThread.setName("AllocationFileReloader");
reloadThread.setDaemon(true);
reloadThread.start();
super.start();
};
reloadThread.setName("AllocationFileReloader");
reloadThread.setDaemon(true);
}
super.serviceInit(conf);
}
@Override
public void stop() {
public void serviceStart() throws Exception {
if (reloadThread != null) {
reloadThread.start();
}
super.serviceStart();
}
@Override
public void serviceStop() throws Exception {
running = false;
if (reloadThread != null) {
reloadThread.interrupt();
@ -156,7 +160,7 @@ public class AllocationFileLoaderService extends AbstractService {
LOG.warn("reloadThread fails to join.");
}
}
super.stop();
super.serviceStop();
}
/**