shameersss1 commented on code in PR #263:
URL: https://github.com/apache/tez/pull/263#discussion_r1104318922
##########
tez-mapreduce/src/main/java/org/apache/hadoop/mapred/split/TezGroupedSplitsInputFormat.java:
##########
@@ -183,23 +247,45 @@ protected boolean initNextRecordReader() throws
IOException {
// if all chunks have been processed, nothing more to do.
if (idx == groupedSplit.wrappedSplits.size()) {
+ if (initReaderExecService != null) {
+ LOG.info("Shutting down the init record reader threadpool");
+ initReaderExecService.shutdownNow();
+ }
return false;
}
if (LOG.isDebugEnabled()) {
- LOG.debug("Init record reader for index " + idx + " of " +
+ LOG.debug("Init record reader for index " + idx + " of " +
groupedSplit.wrappedSplits.size());
}
// get a record reader for the idx-th chunk
try {
- curReader = wrappedInputFormat.getRecordReader(
- groupedSplit.wrappedSplits.get(idx), job, reporter);
+ // get the cur reader directly when async split opening is disabled
+ if (initReaderExecService == null) {
+ curReader =
wrappedInputFormat.getRecordReader(groupedSplit.wrappedSplits.get(idx), job,
reporter);
+ } else {
+ preInitReaders();
+ curReader = initedReaders.take().get();
+ }
} catch (Exception e) {
- throw new RuntimeException (e);
+ failureOccurred.set(true);
+ if (e instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ }
+ if (initedReaders != null) {
+ cancelsFutures();
+ }
+ throw new RuntimeException(e);
}
idx++;
- return true;
+ return curReader != null;
+ }
+
+ private void cancelsFutures() {
Review Comment:
Ack!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
