Skip to content

TEZ-4617 prevent resource leaks for Closeables #402

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
dk2k wants to merge 1 commit into
base: master
Choose a base branch
from
+219 −212
Open

TEZ-4617 prevent resource leaks for Closeables #402

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
43 changes: 22 additions & 21 deletions tez-dag/src/main/java/org/apache/tez/dag/app/web/AMWebController.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -893,28 +893,29 @@ public static class StaticAMView extends View {
@Override
public void render() {
response().setContentType(MimeType.HTML);
PrintWriter pw = writer();
pw.write("<html>");
pw.write("<head>");
pw.write("<meta charset=\"utf-8\">");
pw.write("<title>Redirecting to Tez UI</title>");
pw.write("</head>");
pw.write("<body>");
if (historyUrl == null || historyUrl.isEmpty()) {
pw.write("<h1>Tez UI Url is not defined.</h1>" +
"<p>To enable tracking url pointing to Tez UI, set the config <b>" +
TezConfiguration.TEZ_HISTORY_URL_BASE + "</b> in the tez-site.xml.</p>");
} else {
pw.write("<h1>Redirecting to Tez UI</h1>. <p>If you are not redirected shortly, click " +
"<a href='" + historyUrl + "'><b>here</b></a></p>"
);
pw.write("<script type='text/javascript'>setTimeout(function() { " +
"window.location.replace('" + historyUrl + "');" +
"}, 0); </script>");
try(PrintWriter pw = writer()) {
pw.write("<html>");
pw.write("<head>");
pw.write("<meta charset=\"utf-8\">");
pw.write("<title>Redirecting to Tez UI</title>");
pw.write("</head>");
pw.write("<body>");
if (historyUrl == null || historyUrl.isEmpty()) {
pw.write("<h1>Tez UI Url is not defined.</h1>" +
"<p>To enable tracking url pointing to Tez UI, set the config <b>" +
TezConfiguration.TEZ_HISTORY_URL_BASE + "</b> in the tez-site.xml.</p>");
} else {
pw.write("<h1>Redirecting to Tez UI</h1>. <p>If you are not redirected shortly, click " +
"<a href='" + historyUrl + "'><b>here</b></a></p>"
);
pw.write("<script type='text/javascript'>setTimeout(function() { " +
"window.location.replace('" + historyUrl + "');" +
"}, 0); </script>");
}
pw.write("</body>");
pw.write("</html>");
pw.flush();
}
pw.write("</body>");
pw.write("</html>");
pw.flush();
}
}

Expand Down
11 changes: 6 additions & 5 deletions tez-tests/src/main/java/org/apache/tez/mapreduce/examples/BroadcastAndOneToOneExample.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -130,12 +130,13 @@ private DAG createDAG(FileSystem fs, TezConfiguration tezConf,

int numBroadcastTasks = 2;
int numOneToOneTasks = 3;
int numNMs;
if (doLocalityCheck) {
YarnClient yarnClient = YarnClient.createYarnClient();
yarnClient.init(tezConf);
yarnClient.start();
int numNMs = yarnClient.getNodeReports(NodeState.RUNNING).size();
yarnClient.stop();
try (YarnClient yarnClient = YarnClient.createYarnClient()) {
yarnClient.init(tezConf);
yarnClient.start();
numNMs = yarnClient.getNodeReports(NodeState.RUNNING).size();
}
// create enough 1-1 tasks to run in parallel
numOneToOneTasks = numNMs - numBroadcastTasks - 1;// 1 AM
if (numOneToOneTasks < 1) {
Expand Down
163 changes: 83 additions & 80 deletions tez-tests/src/main/java/org/apache/tez/mapreduce/examples/Join.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -82,92 +82,95 @@ static int printUsage() {
@SuppressWarnings("deprecation")
public int run(String[] args) throws Exception {
Configuration conf = getConf();
JobClient client = new JobClient(conf);
ClusterStatus cluster = client.getClusterStatus();
int num_reduces = (int) (cluster.getMaxReduceTasks() * 0.9);
String join_reduces = conf.get(REDUCES_PER_HOST);
if (join_reduces != null) {
num_reduces = cluster.getTaskTrackers() *
Integer.parseInt(join_reduces);
}
Job job = new Job(conf);
job.setJobName("join");
job.setJarByClass(Sort.class);

job.setMapperClass(Mapper.class);
job.setReducerClass(Reducer.class);

Class<? extends InputFormat> inputFormatClass =
SequenceFileInputFormat.class;
Class<? extends OutputFormat> outputFormatClass =
SequenceFileOutputFormat.class;
Class<? extends WritableComparable> outputKeyClass = BytesWritable.class;
Class<? extends Writable> outputValueClass = TupleWritable.class;
String op = "inner";
List<String> otherArgs = new ArrayList<String>();
for(int i=0; i < args.length; ++i) {
try {
if ("-r".equals(args[i])) {
num_reduces = Integer.parseInt(args[++i]);
} else if ("-inFormat".equals(args[i])) {
inputFormatClass =
Class.forName(args[++i]).asSubclass(InputFormat.class);
} else if ("-outFormat".equals(args[i])) {
outputFormatClass =
Class.forName(args[++i]).asSubclass(OutputFormat.class);
} else if ("-outKey".equals(args[i])) {
outputKeyClass =
Class.forName(args[++i]).asSubclass(WritableComparable.class);
} else if ("-outValue".equals(args[i])) {
outputValueClass =
Class.forName(args[++i]).asSubclass(Writable.class);
} else if ("-joinOp".equals(args[i])) {
op = args[++i];
} else {
otherArgs.add(args[i]);
try (JobClient client = new JobClient(conf)) {

ClusterStatus cluster = client.getClusterStatus();
int num_reduces = (int) (cluster.getMaxReduceTasks() * 0.9);
String join_reduces = conf.get(REDUCES_PER_HOST);
if (join_reduces != null) {
num_reduces = cluster.getTaskTrackers() *
Integer.parseInt(join_reduces);
}

Job job = new Job(conf);
job.setJobName("join");
job.setJarByClass(Sort.class);

job.setMapperClass(Mapper.class);
job.setReducerClass(Reducer.class);

Class<? extends InputFormat> inputFormatClass =
SequenceFileInputFormat.class;
Class<? extends OutputFormat> outputFormatClass =
SequenceFileOutputFormat.class;
Class<? extends WritableComparable> outputKeyClass = BytesWritable.class;
Class<? extends Writable> outputValueClass = TupleWritable.class;
String op = "inner";
List<String> otherArgs = new ArrayList<String>();
for (int i = 0; i < args.length; ++i) {
try {
if ("-r".equals(args[i])) {
num_reduces = Integer.parseInt(args[++i]);
} else if ("-inFormat".equals(args[i])) {
inputFormatClass =
Class.forName(args[++i]).asSubclass(InputFormat.class);
} else if ("-outFormat".equals(args[i])) {
outputFormatClass =
Class.forName(args[++i]).asSubclass(OutputFormat.class);
} else if ("-outKey".equals(args[i])) {
outputKeyClass =
Class.forName(args[++i]).asSubclass(WritableComparable.class);
} else if ("-outValue".equals(args[i])) {
outputValueClass =
Class.forName(args[++i]).asSubclass(Writable.class);
} else if ("-joinOp".equals(args[i])) {
op = args[++i];
} else {
otherArgs.add(args[i]);
}
} catch (NumberFormatException except) {
System.out.println("ERROR: Integer expected instead of " + args[i]);
return printUsage();
} catch (ArrayIndexOutOfBoundsException except) {
System.out.println("ERROR: Required parameter missing from " +
args[i - 1]);
return printUsage(); // exits
}
} catch (NumberFormatException except) {
System.out.println("ERROR: Integer expected instead of " + args[i]);
return printUsage();
} catch (ArrayIndexOutOfBoundsException except) {
System.out.println("ERROR: Required parameter missing from " +
args[i-1]);
return printUsage(); // exits
}
}

// Set user-supplied (possibly default) job configs
job.setNumReduceTasks(num_reduces);
// Set user-supplied (possibly default) job configs
job.setNumReduceTasks(num_reduces);

if (otherArgs.size() < 2) {
System.out.println("ERROR: Wrong number of parameters: ");
return printUsage();
}
if (otherArgs.size() < 2) {
System.out.println("ERROR: Wrong number of parameters: ");
return printUsage();
}

FileOutputFormat.setOutputPath(job,
new Path(otherArgs.remove(otherArgs.size() - 1)));
List<Path> plist = new ArrayList<Path>(otherArgs.size());
for (String s : otherArgs) {
plist.add(new Path(s));
}
FileOutputFormat.setOutputPath(job,
new Path(otherArgs.remove(otherArgs.size() - 1)));
List<Path> plist = new ArrayList<Path>(otherArgs.size());
for (String s : otherArgs) {
plist.add(new Path(s));
}

job.setInputFormatClass(CompositeInputFormat.class);
job.getConfiguration().set(CompositeInputFormat.JOIN_EXPR,
CompositeInputFormat.compose(op, inputFormatClass,
plist.toArray(new Path[0])));
job.setOutputFormatClass(outputFormatClass);

job.setOutputKeyClass(outputKeyClass);
job.setOutputValueClass(outputValueClass);

Date startTime = new Date();
System.out.println("Job started: " + startTime);
int ret = job.waitForCompletion(true) ? 0 : 1 ;
Date end_time = new Date();
System.out.println("Job ended: " + end_time);
System.out.println("The job took " +
(end_time.getTime() - startTime.getTime()) /1000 + " seconds.");
return ret;
job.setInputFormatClass(CompositeInputFormat.class);
job.getConfiguration().set(CompositeInputFormat.JOIN_EXPR,
CompositeInputFormat.compose(op, inputFormatClass,
plist.toArray(new Path[0])));
job.setOutputFormatClass(outputFormatClass);

job.setOutputKeyClass(outputKeyClass);
job.setOutputValueClass(outputValueClass);

Date startTime = new Date();
System.out.println("Job started: " + startTime);
int ret = job.waitForCompletion(true) ? 0 : 1;
Date end_time = new Date();
System.out.println("Job ended: " + end_time);
System.out.println("The job took " +
(end_time.getTime() - startTime.getTime()) / 1000 + " seconds.");
return ret;
}
}

public static void main(String[] args) throws Exception {
Expand Down
125 changes: 63 additions & 62 deletions tez-tests/src/main/java/org/apache/tez/mapreduce/examples/RandomTextWriter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -180,71 +180,72 @@ public int run(String[] args) throws Exception {
}

Configuration conf = getConf();
JobClient client = new JobClient(conf);
ClusterStatus cluster = client.getClusterStatus();
int numMapsPerHost = conf.getInt(MAPS_PER_HOST, 10);
long numBytesToWritePerMap = conf.getLong(BYTES_PER_MAP,
1*1024*1024*1024);
if (numBytesToWritePerMap == 0) {
System.err.println("Cannot have " + BYTES_PER_MAP +" set to 0");
return -2;
}
long totalBytesToWrite = conf.getLong(TOTAL_BYTES,
numMapsPerHost*numBytesToWritePerMap*cluster.getTaskTrackers());
int numMaps = (int) (totalBytesToWrite / numBytesToWritePerMap);
if (numMaps == 0 && totalBytesToWrite > 0) {
numMaps = 1;
conf.setLong(BYTES_PER_MAP, totalBytesToWrite);
}
conf.setInt(MRJobConfig.NUM_MAPS, numMaps);

Job job = new Job(conf);

job.setJarByClass(RandomTextWriter.class);
job.setJobName("random-text-writer");

job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);

job.setInputFormatClass(RandomWriter.RandomInputFormat.class);
job.setMapperClass(RandomTextMapper.class);

Class<? extends OutputFormat> outputFormatClass =
SequenceFileOutputFormat.class;
List<String> otherArgs = new ArrayList<String>();
for(int i=0; i < args.length; ++i) {
try {
if ("-outFormat".equals(args[i])) {
outputFormatClass =
Class.forName(args[++i]).asSubclass(OutputFormat.class);
} else {
otherArgs.add(args[i]);
try (JobClient client = new JobClient(conf)) {
ClusterStatus cluster = client.getClusterStatus();
int numMapsPerHost = conf.getInt(MAPS_PER_HOST, 10);
long numBytesToWritePerMap = conf.getLong(BYTES_PER_MAP,
1 * 1024 * 1024 * 1024);
if (numBytesToWritePerMap == 0) {
System.err.println("Cannot have " + BYTES_PER_MAP + " set to 0");
return -2;
}
long totalBytesToWrite = conf.getLong(TOTAL_BYTES,
numMapsPerHost * numBytesToWritePerMap * cluster.getTaskTrackers());
int numMaps = (int) (totalBytesToWrite / numBytesToWritePerMap);
if (numMaps == 0 && totalBytesToWrite > 0) {
numMaps = 1;
conf.setLong(BYTES_PER_MAP, totalBytesToWrite);
}
conf.setInt(MRJobConfig.NUM_MAPS, numMaps);

Job job = new Job(conf);

job.setJarByClass(RandomTextWriter.class);
job.setJobName("random-text-writer");

job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);

job.setInputFormatClass(RandomWriter.RandomInputFormat.class);
job.setMapperClass(RandomTextMapper.class);

Class<? extends OutputFormat> outputFormatClass =
SequenceFileOutputFormat.class;
List<String> otherArgs = new ArrayList<String>();
for (int i = 0; i < args.length; ++i) {
try {
if ("-outFormat".equals(args[i])) {
outputFormatClass =
Class.forName(args[++i]).asSubclass(OutputFormat.class);
} else {
otherArgs.add(args[i]);
}
} catch (ArrayIndexOutOfBoundsException except) {
System.out.println("ERROR: Required parameter missing from " +
args[i - 1]);
return printUsage(); // exits
}
} catch (ArrayIndexOutOfBoundsException except) {
System.out.println("ERROR: Required parameter missing from " +
args[i-1]);
return printUsage(); // exits
}
}

job.setOutputFormatClass(outputFormatClass);
FileOutputFormat.setOutputPath(job, new Path(otherArgs.get(0)));

System.out.println("Running " + numMaps + " maps.");

// reducer NONE
job.setNumReduceTasks(0);

Date startTime = new Date();
System.out.println("Job started: " + startTime);
int ret = job.waitForCompletion(true) ? 0 : 1;
Date endTime = new Date();
System.out.println("Job ended: " + endTime);
System.out.println("The job took " +
(endTime.getTime() - startTime.getTime()) /1000 +
" seconds.");

return ret;
job.setOutputFormatClass(outputFormatClass);
FileOutputFormat.setOutputPath(job, new Path(otherArgs.get(0)));

System.out.println("Running " + numMaps + " maps.");

// reducer NONE
job.setNumReduceTasks(0);

Date startTime = new Date();
System.out.println("Job started: " + startTime);
int ret = job.waitForCompletion(true) ? 0 : 1;
Date endTime = new Date();
System.out.println("Job ended: " + endTime);
System.out.println("The job took " +
(endTime.getTime() - startTime.getTime()) / 1000 +
" seconds.");

return ret;
}
}

public static void main(String[] args) throws Exception {
Expand Down
Loading