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

YARN-3882. AggregatedLogFormat should close aclScanner and ownerScanner 

after create them. Contributed by zhihai xu
This commit is contained in:
Xuan 2015-07-04 21:51:58 -07:00
parent 2eae130ab9
commit 688617d6d7
2 changed files with 49 additions and 37 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -589,6 +589,9 @@ Release 2.8.0 - UNRELEASED
YARN-3875. FSSchedulerNode#reserveResource() doesn't print Application Id YARN-3875. FSSchedulerNode#reserveResource() doesn't print Application Id
properly in log. (Bibin A Chundatt via devaraj) properly in log. (Bibin A Chundatt via devaraj)
YARN-3882. AggregatedLogFormat should close aclScanner and ownerScanner
after create them. (zhihai xu via xgong)
Release 2.7.2 - UNRELEASED Release 2.7.2 - UNRELEASED
INCOMPATIBLE CHANGES INCOMPATIBLE CHANGES

83
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/logaggregation/AggregatedLogFormat.java
View File

@ -489,18 +489,23 @@ public LogReader(Configuration conf, Path remoteAppLogFile)
* @throws IOException * @throws IOException
*/ */
public String getApplicationOwner() throws IOException { public String getApplicationOwner() throws IOException {
TFile.Reader.Scanner ownerScanner = reader.createScanner(); TFile.Reader.Scanner ownerScanner = null;
LogKey key = new LogKey(); try {
while (!ownerScanner.atEnd()) { ownerScanner = reader.createScanner();
TFile.Reader.Scanner.Entry entry = ownerScanner.entry(); LogKey key = new LogKey();
key.readFields(entry.getKeyStream()); while (!ownerScanner.atEnd()) {
if (key.toString().equals(APPLICATION_OWNER_KEY.toString())) { TFile.Reader.Scanner.Entry entry = ownerScanner.entry();
DataInputStream valueStream = entry.getValueStream(); key.readFields(entry.getKeyStream());
return valueStream.readUTF(); if (key.toString().equals(APPLICATION_OWNER_KEY.toString())) {
DataInputStream valueStream = entry.getValueStream();
return valueStream.readUTF();
}
ownerScanner.advance();
} }
ownerScanner.advance(); return null;
} finally {
IOUtils.cleanup(LOG, ownerScanner);
} }
return null;
} }
/** /**
@ -513,38 +518,42 @@ public String getApplicationOwner() throws IOException {
public Map<ApplicationAccessType, String> getApplicationAcls() public Map<ApplicationAccessType, String> getApplicationAcls()
throws IOException { throws IOException {
// TODO Seek directly to the key once a comparator is specified. // TODO Seek directly to the key once a comparator is specified.
TFile.Reader.Scanner aclScanner = reader.createScanner(); TFile.Reader.Scanner aclScanner = null;
LogKey key = new LogKey(); try {
Map<ApplicationAccessType, String> acls = aclScanner = reader.createScanner();
new HashMap<ApplicationAccessType, String>(); LogKey key = new LogKey();
while (!aclScanner.atEnd()) { Map<ApplicationAccessType, String> acls =
TFile.Reader.Scanner.Entry entry = aclScanner.entry(); new HashMap<ApplicationAccessType, String>();
key.readFields(entry.getKeyStream()); while (!aclScanner.atEnd()) {
if (key.toString().equals(APPLICATION_ACL_KEY.toString())) { TFile.Reader.Scanner.Entry entry = aclScanner.entry();
DataInputStream valueStream = entry.getValueStream(); key.readFields(entry.getKeyStream());
while (true) { if (key.toString().equals(APPLICATION_ACL_KEY.toString())) {
String appAccessOp = null; DataInputStream valueStream = entry.getValueStream();
String aclString = null; while (true) {
try { String appAccessOp = null;
appAccessOp = valueStream.readUTF(); String aclString = null;
} catch (EOFException e) { try {
// Valid end of stream. appAccessOp = valueStream.readUTF();
break; } catch (EOFException e) {
// Valid end of stream.
break;
}
try {
aclString = valueStream.readUTF();
} catch (EOFException e) {
throw new YarnRuntimeException("Error reading ACLs", e);
}
acls.put(ApplicationAccessType.valueOf(appAccessOp), aclString);
} }
try {
aclString = valueStream.readUTF();
} catch (EOFException e) {
throw new YarnRuntimeException("Error reading ACLs", e);
}
acls.put(ApplicationAccessType.valueOf(appAccessOp), aclString);
} }
aclScanner.advance();
} }
aclScanner.advance(); return acls;
} finally {
IOUtils.cleanup(LOG, aclScanner);
} }
return acls;
} }
/** /**
* Read the next key and return the value-stream. * Read the next key and return the value-stream.
* *