You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tez.apache.org by ss...@apache.org on 2014/10/15 20:38:03 UTC
git commit: TEZ-1637. Improved shuffle error handling across NM
restarts. Contributed by Rajesh Balamohan.
Repository: tez
Updated Branches:
refs/heads/master c04c7b5e5 -> 625362182
TEZ-1637. Improved shuffle error handling across NM restarts.
Contributed by Rajesh Balamohan.
Project: http://git-wip-us.apache.org/repos/asf/tez/repo
Commit: http://git-wip-us.apache.org/repos/asf/tez/commit/62536218
Tree: http://git-wip-us.apache.org/repos/asf/tez/tree/62536218
Diff: http://git-wip-us.apache.org/repos/asf/tez/diff/62536218
Branch: refs/heads/master
Commit: 62536218271317d8b8e57c7ce172d247a95db3ba
Parents: c04c7b5
Author: Siddharth Seth <ss...@apache.org>
Authored: Wed Oct 15 11:37:19 2014 -0700
Committer: Siddharth Seth <ss...@apache.org>
Committed: Wed Oct 15 11:37:19 2014 -0700
----------------------------------------------------------------------
CHANGES.txt | 1 +
.../library/common/shuffle/impl/Fetcher.java | 163 +++++++++++++------
.../exceptions/FetcherReadTimeoutException.java | 49 ++++++
.../runtime/library/shuffle/common/Fetcher.java | 97 +++++++++--
.../library/shuffle/common/HttpConnection.java | 4 +-
.../common/shuffle/impl/TestFetcher.java | 145 ++++++++++++++++-
6 files changed, 391 insertions(+), 68 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tez/blob/62536218/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 9fd3b74..1b5c063 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -26,6 +26,7 @@ ALL CHANGES:
TEZ-1649. ShuffleVertexManager auto reduce parallelism can cause jobs to hang indefinitely.
TEZ-1566. Reduce log verbosity.
TEZ-1083. Enable IFile RLE for DefaultSorter.
+ TEZ-1637. Improved shuffle error handling across NM restarts.
Release 0.5.1: 2014-10-02
http://git-wip-us.apache.org/repos/asf/tez/blob/62536218/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/Fetcher.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/Fetcher.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/Fetcher.java
index 83ded06..ea1bac3 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/Fetcher.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/Fetcher.java
@@ -20,10 +20,12 @@ package org.apache.tez.runtime.library.common.shuffle.impl;
import java.io.DataInputStream;
import java.io.IOException;
import java.net.HttpURLConnection;
+import java.net.SocketTimeoutException;
import java.net.URL;
import java.util.Arrays;
import java.util.Iterator;
import java.util.LinkedHashSet;
+import java.util.LinkedList;
import java.util.List;
import java.util.Set;
@@ -41,12 +43,12 @@ import org.apache.hadoop.yarn.api.ApplicationConstants;
import org.apache.tez.common.TezRuntimeFrameworkConfigs;
import org.apache.tez.common.counters.TezCounter;
import org.apache.tez.runtime.api.InputContext;
-import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
import org.apache.tez.runtime.library.common.Constants;
import org.apache.tez.runtime.library.common.InputAttemptIdentifier;
import org.apache.tez.runtime.library.common.shuffle.impl.MapOutput.Type;
import org.apache.tez.runtime.library.common.sort.impl.TezIndexRecord;
import org.apache.tez.runtime.library.common.sort.impl.TezSpillRecord;
+import org.apache.tez.runtime.library.exceptions.FetcherReadTimeoutException;
import org.apache.tez.runtime.library.shuffle.common.HttpConnection;
import org.apache.tez.runtime.library.shuffle.common.HttpConnection.HttpConnectionParams;
import org.apache.tez.runtime.library.shuffle.common.ShuffleUtils;
@@ -82,7 +84,8 @@ class Fetcher extends Thread {
private final CompressionCodec codec;
private final SecretKey jobTokenSecret;
- private volatile boolean stopped = false;
+ @VisibleForTesting
+ volatile boolean stopped = false;
private final boolean ifileReadAhead;
private final int ifileReadAheadLength;
@@ -95,6 +98,9 @@ class Fetcher extends Thread {
HttpConnectionParams httpConnectionParams;
final static String localhostName = NetUtils.getHostname();
+
+ // Initiative value is 0, which means it hasn't retried yet.
+ private long retryStartTime = 0;
public Fetcher(HttpConnectionParams httpConnectionParams,
ShuffleScheduler scheduler, MergeManager merger,
@@ -216,6 +222,8 @@ class Fetcher extends Thread {
*/
@VisibleForTesting
protected void copyFromHost(MapHost host) throws IOException {
+ // reset retryStartTime for a new host
+ retryStartTime = 0;
// Get completed maps on 'host'
List<InputAttemptIdentifier> srcAttempts = scheduler.getMapsForHost(host);
currentPartition = host.getPartitionId();
@@ -235,55 +243,17 @@ class Fetcher extends Thread {
remaining = new LinkedHashSet<InputAttemptIdentifier>(srcAttempts);
// Construct the url and connect
- boolean connectSucceeded = false;
-
- try {
- URL url = ShuffleUtils.constructInputURL(host.getBaseUrl(), srcAttempts,
- httpConnectionParams.getKeepAlive());
- httpConnection = new HttpConnection(url, httpConnectionParams,
- logIdentifier, jobTokenSecret);
- connectSucceeded = httpConnection.connect();
-
+ if (!setupConnection(host, srcAttempts)) {
if (stopped) {
- LOG.info("Detected fetcher has been shutdown after connection establishment. Returning");
cleanupCurrentConnection(true);
- putBackRemainingMapOutputs(host);
- return;
- }
- input = httpConnection.getInputStream();
- httpConnection.validate();
- } catch (IOException ie) {
- if (stopped) {
- LOG.info("Not reporting fetch failure, since an Exception was caught after shutdown");
- cleanupCurrentConnection(true);
- putBackRemainingMapOutputs(host);
- return;
- }
- ioErrs.increment(1);
- if (!connectSucceeded) {
- LOG.warn("Failed to connect to " + host + " with " + remaining.size() + " inputs", ie);
- connectionErrs.increment(1);
- } else {
- LOG.warn("Failed to verify reply after connecting to " + host + " with " + remaining.size()
- + " inputs pending", ie);
}
-
- // At this point, either the connection failed, or the initial header verification failed.
- // The error does not relate to any specific Input. Report all of them as failed.
- // This ends up indirectly penalizing the host (multiple failures reported on the single host)
- for(InputAttemptIdentifier left: remaining) {
- // Need to be handling temporary glitches ..
- // Report read error to the AM to trigger source failure heuristics
- scheduler.copyFailed(left, host, connectSucceeded, !connectSucceeded);
- }
-
// Add back all remaining maps - which at this point is ALL MAPS the
// Fetcher was started with. The Scheduler takes care of retries,
// reporting too many failures etc.
putBackRemainingMapOutputs(host);
return;
}
-
+
try {
// Loop through available map-outputs and fetch them
// On any error, faildTasks is not null and we exit
@@ -291,10 +261,24 @@ class Fetcher extends Thread {
// yet_to_be_fetched list and marking the failed tasks.
InputAttemptIdentifier[] failedTasks = null;
while (!remaining.isEmpty() && failedTasks == null) {
- // fail immediately after first failure because we dont know how much to
+ // fail immediately after first failure because we dont know how much to
// skip for this error in the input stream. So we cannot move on to the
// remaining outputs. YARN-1773. Will get to them in the next retry.
- failedTasks = copyMapOutput(host, input);
+ try {
+ failedTasks = copyMapOutput(host, input);
+ } catch (FetcherReadTimeoutException e) {
+ // Setup connection again if disconnected
+ cleanupCurrentConnection(true);
+
+ // Connect with retry
+ if (!setupConnection(host, new LinkedList<InputAttemptIdentifier>(remaining))) {
+ if (stopped) {
+ cleanupCurrentConnection(true);
+ }
+ failedTasks = new InputAttemptIdentifier[] {getNextRemainingAttempt()};
+ break;
+ }
+ }
}
if(failedTasks != null && failedTasks.length > 0) {
@@ -317,6 +301,50 @@ class Fetcher extends Thread {
}
@VisibleForTesting
+ boolean setupConnection(MapHost host, List<InputAttemptIdentifier> attempts)
+ throws IOException {
+ boolean connectSucceeded = false;
+ try {
+ URL url = ShuffleUtils.constructInputURL(host.getBaseUrl(), attempts,
+ httpConnectionParams.getKeepAlive());
+ httpConnection = new HttpConnection(url, httpConnectionParams,
+ logIdentifier, jobTokenSecret);
+ connectSucceeded = httpConnection.connect();
+
+ if (stopped) {
+ LOG.info("Detected fetcher has been shutdown after connection establishment. Returning");
+ return false;
+ }
+ input = httpConnection.getInputStream();
+ httpConnection.validate();
+ return true;
+ } catch (IOException ie) {
+ if (stopped) {
+ LOG.info("Not reporting fetch failure, since an Exception was caught after shutdown");
+ return false;
+ }
+ ioErrs.increment(1);
+ if (!connectSucceeded) {
+ LOG.warn("Failed to connect to " + host + " with " + remaining.size() + " inputs", ie);
+ connectionErrs.increment(1);
+ } else {
+ LOG.warn("Failed to verify reply after connecting to " + host + " with " + remaining.size()
+ + " inputs pending", ie);
+ }
+
+ // At this point, either the connection failed, or the initial header verification failed.
+ // The error does not relate to any specific Input. Report all of them as failed.
+ // This ends up indirectly penalizing the host (multiple failures reported on the single host)
+ for(InputAttemptIdentifier left: remaining) {
+ // Need to be handling temporary glitches ..
+ // Report read error to the AM to trigger source failure heuristics
+ scheduler.copyFailed(left, host, connectSucceeded, !connectSucceeded);
+ }
+ return false;
+ }
+ }
+
+ @VisibleForTesting
protected void putBackRemainingMapOutputs(MapHost host) {
// Cycle through remaining MapOutputs
boolean isFirst = true;
@@ -336,8 +364,8 @@ class Fetcher extends Thread {
private static InputAttemptIdentifier[] EMPTY_ATTEMPT_ID_ARRAY = new InputAttemptIdentifier[0];
- private InputAttemptIdentifier[] copyMapOutput(MapHost host,
- DataInputStream input) {
+ protected InputAttemptIdentifier[] copyMapOutput(MapHost host,
+ DataInputStream input) throws FetcherReadTimeoutException {
MapOutput mapOutput = null;
InputAttemptIdentifier srcAttemptId = null;
long decompressedLength = -1;
@@ -417,9 +445,12 @@ class Fetcher extends Thread {
throw new IOException("Unknown mapOutput type while fetching shuffle data:" +
mapOutput.getType());
}
-
+
// Inform the shuffle scheduler
long endTime = System.currentTimeMillis();
+ // Reset retryStartTime as map task make progress if retried before.
+ retryStartTime = 0;
+
scheduler.copySucceeded(srcAttemptId, host, compressedLength, decompressedLength,
endTime - startTime, mapOutput);
// Note successful shuffle
@@ -437,6 +468,13 @@ class Fetcher extends Thread {
// Don't need to put back - since that's handled by the invoker
return EMPTY_ATTEMPT_ID_ARRAY;
}
+ if (shouldRetry(host, ioe)) {
+ //release mem/file handles
+ if (mapOutput != null) {
+ mapOutput.abort();
+ }
+ throw new FetcherReadTimeoutException(ioe);
+ }
ioErrs.increment(1);
if (srcAttemptId == null || mapOutput == null) {
LOG.info("fetcher#" + id + " failed to read map header" +
@@ -459,6 +497,37 @@ class Fetcher extends Thread {
}
}
+
+ /**
+ * Check connection needs to be re-established.
+ *
+ * @param host
+ * @param ioe
+ * @return true to indicate connection retry. false otherwise.
+ * @throws IOException
+ */
+ private boolean shouldRetry(MapHost host, IOException ioe) {
+ if (!(ioe instanceof SocketTimeoutException)) {
+ return false;
+ }
+ // First time to retry.
+ long currentTime = System.currentTimeMillis();
+ if (retryStartTime == 0) {
+ retryStartTime = currentTime;
+ }
+
+ if (currentTime - retryStartTime < httpConnectionParams.getReadTimeout()) {
+ LOG.warn("Shuffle output from " + host.getHostIdentifier() +
+ " failed, retry it.");
+ //retry connecting to the host
+ return true;
+ } else {
+ // timeout, prepare to be failed.
+ LOG.warn("Timeout for copying MapOutput with retry on host " + host
+ + "after " + httpConnectionParams.getReadTimeout() + "milliseconds.");
+ return false;
+ }
+ }
/**
* Do some basic verification on the input received -- Being defensive
http://git-wip-us.apache.org/repos/asf/tez/blob/62536218/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/exceptions/FetcherReadTimeoutException.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/exceptions/FetcherReadTimeoutException.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/exceptions/FetcherReadTimeoutException.java
new file mode 100644
index 0000000..32b3187
--- /dev/null
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/exceptions/FetcherReadTimeoutException.java
@@ -0,0 +1,49 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tez.runtime.library.exceptions;
+
+
+import org.apache.hadoop.classification.InterfaceAudience.Public;
+import org.apache.hadoop.classification.InterfaceStability.Evolving;
+import org.apache.tez.dag.api.TezException;
+
+@Public
+@Evolving
+/**
+ * Exception invoked when socket read timeout happens in fetcher
+ */
+public class FetcherReadTimeoutException extends TezException {
+
+ public FetcherReadTimeoutException() {
+ super("Socket timeout while reading");
+ }
+
+ public FetcherReadTimeoutException(String message) {
+ super(message);
+ }
+
+ public FetcherReadTimeoutException(Throwable cause) {
+ super("Socket timeout while reading", cause);
+ }
+
+ public FetcherReadTimeoutException(String message, Throwable cause) {
+ super(message, cause);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/tez/blob/62536218/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/shuffle/common/Fetcher.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/shuffle/common/Fetcher.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/shuffle/common/Fetcher.java
index e25124b..6a3b6da 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/shuffle/common/Fetcher.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/shuffle/common/Fetcher.java
@@ -21,17 +21,17 @@ package org.apache.tez.runtime.library.shuffle.common;
import java.io.DataInputStream;
import java.io.File;
import java.io.IOException;
-import java.io.InputStream;
import java.io.RandomAccessFile;
+import java.net.SocketTimeoutException;
import java.net.URL;
import java.nio.channels.FileChannel;
import java.nio.channels.FileLock;
import java.nio.channels.OverlappingFileLockException;
import java.util.Arrays;
import java.util.HashMap;
-import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashSet;
+import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
@@ -42,13 +42,9 @@ import javax.crypto.SecretKey;
import com.google.common.annotations.VisibleForTesting;
-import org.apache.commons.io.IOUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FSDataOutputStream;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.LocalDirAllocator;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RawLocalFileSystem;
@@ -56,7 +52,6 @@ import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.util.DiskChecker.DiskErrorException;
import org.apache.hadoop.yarn.api.ApplicationConstants;
import org.apache.hadoop.yarn.api.records.ApplicationId;
-import org.apache.tez.common.TezRuntimeFrameworkConfigs;
import org.apache.tez.dag.api.TezUncheckedException;
import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
import org.apache.tez.runtime.library.common.Constants;
@@ -64,11 +59,11 @@ import org.apache.tez.runtime.library.common.InputAttemptIdentifier;
import org.apache.tez.runtime.library.common.shuffle.impl.ShuffleHeader;
import org.apache.tez.runtime.library.common.sort.impl.TezIndexRecord;
import org.apache.tez.runtime.library.common.sort.impl.TezSpillRecord;
+import org.apache.tez.runtime.library.exceptions.FetcherReadTimeoutException;
import org.apache.tez.runtime.library.shuffle.common.FetchedInput.Type;
import org.apache.tez.runtime.library.shuffle.common.HttpConnection.HttpConnectionParams;
import com.google.common.base.Preconditions;
-import com.google.common.collect.Iterables;
/**
* Responsible for fetching inputs served by the ShuffleHandler for a single
@@ -122,6 +117,9 @@ public class Fetcher implements Callable<FetchResult> {
private final Path lockPath;
private final RawLocalFileSystem localFs;
+ // Initiative value is 0, which means it hasn't retried yet.
+ private long retryStartTime = 0;
+
private final boolean isDebugEnabled = LOG.isDebugEnabled();
private Fetcher(FetcherCallback fetcherCallback, HttpConnectionParams params,
@@ -392,12 +390,11 @@ public class Fetcher implements Callable<FetchResult> {
return doHttpFetch(null);
}
- @VisibleForTesting
- protected HostFetchResult doHttpFetch(CachingCallBack callback) {
+ private HostFetchResult setupConnection(List<InputAttemptIdentifier> attempts) {
try {
StringBuilder baseURI = ShuffleUtils.constructBaseURIForShuffleHandler(host,
port, partition, appId.toString(), httpConnectionParams.isSSLShuffleEnabled());
- this.url = ShuffleUtils.constructInputURL(baseURI.toString(), srcAttempts,
+ this.url = ShuffleUtils.constructInputURL(baseURI.toString(), attempts,
httpConnectionParams.getKeepAlive());
httpConnection = new HttpConnection(url, httpConnectionParams, logIdentifier, shuffleSecret);
@@ -433,14 +430,23 @@ public class Fetcher implements Callable<FetchResult> {
if (isShutDown.get()) {
LOG.info("Not reporting fetch failure, since an Exception was caught after shutdown");
} else {
- InputAttemptIdentifier firstAttempt = srcAttempts.get(0);
+ InputAttemptIdentifier firstAttempt = attempts.get(0);
LOG.warn("Fetch Failure from host while connecting: " + host + ", attempt: " + firstAttempt
+ " Informing ShuffleManager: ", e);
return new HostFetchResult(new FetchResult(host, port, partition, remaining),
new InputAttemptIdentifier[] { firstAttempt }, false);
}
}
+ return null;
+ }
+ @VisibleForTesting
+ protected HostFetchResult doHttpFetch(CachingCallBack callback) {
+
+ HostFetchResult connectionsWithRetryResult = setupConnection(srcAttempts);
+ if (connectionsWithRetryResult != null) {
+ return connectionsWithRetryResult;
+ }
// By this point, the connection is setup and the response has been
// validated.
@@ -461,7 +467,19 @@ public class Fetcher implements Callable<FetchResult> {
// yet_to_be_fetched list and marking the failed tasks.
InputAttemptIdentifier[] failedInputs = null;
while (!remaining.isEmpty() && failedInputs == null) {
- failedInputs = fetchInputs(input, callback);
+ try {
+ failedInputs = fetchInputs(input, callback);
+ } catch (FetcherReadTimeoutException e) {
+ //clean up connection
+ shutdownInternal(true);
+
+ // Connect again.
+ connectionsWithRetryResult = setupConnection(
+ new LinkedList<InputAttemptIdentifier>(remaining));
+ if (connectionsWithRetryResult != null) {
+ break;
+ }
+ }
}
return new HostFetchResult(new FetchResult(host, port, partition, remaining), failedInputs,
@@ -579,13 +597,17 @@ public class Fetcher implements Callable<FetchResult> {
}
private void shutdownInternal() {
+ shutdownInternal(false);
+ }
+
+ private void shutdownInternal(boolean disconnect) {
// Synchronizing on isShutDown to ensure we don't run into a parallel close
// Can't synchronize on the main class itself since that would cause the
// shutdown request to block
synchronized (isShutDown) {
try {
if (httpConnection != null) {
- httpConnection.cleanup(false);
+ httpConnection.cleanup(disconnect);
}
} catch (IOException e) {
LOG.info("Exception while shutting down fetcher on " + logIdentifier + " : "
@@ -597,7 +619,8 @@ public class Fetcher implements Callable<FetchResult> {
}
}
- private InputAttemptIdentifier[] fetchInputs(DataInputStream input, CachingCallBack callback) {
+ private InputAttemptIdentifier[] fetchInputs(DataInputStream input,
+ CachingCallBack callback) throws FetcherReadTimeoutException {
FetchedInput fetchedInput = null;
InputAttemptIdentifier srcAttemptId = null;
long decompressedLength = -1;
@@ -688,6 +711,8 @@ public class Fetcher implements Callable<FetchResult> {
// Inform the shuffle scheduler
long endTime = System.currentTimeMillis();
+ // Reset retryStartTime as map task make progress if retried before.
+ retryStartTime = 0;
fetcherCallback.fetchSucceeded(host, srcAttemptId, fetchedInput,
compressedLength, decompressedLength, (endTime - startTime));
@@ -697,6 +722,17 @@ public class Fetcher implements Callable<FetchResult> {
// metrics.successFetch();
return null;
} catch (IOException ioe) {
+ if (shouldRetry(srcAttemptId, ioe)) {
+ //release mem/file handles
+ if (fetchedInput != null) {
+ try {
+ fetchedInput.abort();
+ } catch (IOException e) {
+ LOG.info("Failure to cleanup fetchedInput: " + fetchedInput);
+ }
+ }
+ throw new FetcherReadTimeoutException(ioe);
+ }
// ZZZ Add some shutdown code here
// ZZZ Make sure any assigned memory inputs are aborted
// ioErrs.increment(1);
@@ -725,6 +761,37 @@ public class Fetcher implements Callable<FetchResult> {
}
/**
+ * Check connection needs to be re-established.
+ *
+ * @param srcAttemptId
+ * @param ioe
+ * @return true to indicate connection retry. false otherwise.
+ * @throws IOException
+ */
+ private boolean shouldRetry(InputAttemptIdentifier srcAttemptId, IOException ioe) {
+ if (!(ioe instanceof SocketTimeoutException)) {
+ return false;
+ }
+ // First time to retry.
+ long currentTime = System.currentTimeMillis();
+ if (retryStartTime == 0) {
+ retryStartTime = currentTime;
+ }
+
+ if (currentTime - retryStartTime < httpConnectionParams.getReadTimeout()) {
+ LOG.warn("Shuffle output from " + srcAttemptId +
+ " failed, retry it.");
+ //retry connecting to the host
+ return true;
+ } else {
+ // timeout, prepare to be failed.
+ LOG.warn("Timeout for copying MapOutput with retry on host " + host
+ + "after " + httpConnectionParams.getReadTimeout() + "milliseconds.");
+ return false;
+ }
+ }
+
+ /**
* Do some basic verification on the input received -- Being defensive
*
* @param compressedLength
http://git-wip-us.apache.org/repos/asf/tez/blob/62536218/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/shuffle/common/HttpConnection.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/shuffle/common/HttpConnection.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/shuffle/common/HttpConnection.java
index 3a2c34b..c4e8da4 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/shuffle/common/HttpConnection.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/shuffle/common/HttpConnection.java
@@ -30,6 +30,7 @@ import java.util.concurrent.TimeUnit;
import javax.crypto.SecretKey;
import javax.net.ssl.HttpsURLConnection;
+import com.google.common.annotations.VisibleForTesting;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
@@ -57,7 +58,8 @@ public class HttpConnection {
//Shared by many threads
private static SSLFactory sslFactory;
- private HttpURLConnection connection;
+ @VisibleForTesting
+ protected HttpURLConnection connection;
private DataInputStream input;
private boolean connectionSucceeed;
http://git-wip-us.apache.org/repos/asf/tez/blob/62536218/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/impl/TestFetcher.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/impl/TestFetcher.java b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/impl/TestFetcher.java
index 522e152..e97ff91 100644
--- a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/impl/TestFetcher.java
+++ b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/impl/TestFetcher.java
@@ -19,6 +19,9 @@
package org.apache.tez.runtime.library.common.shuffle.impl;
import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyBoolean;
+import static org.mockito.Matchers.anyInt;
+import static org.mockito.Matchers.anyList;
import static org.mockito.Matchers.anyLong;
import static org.mockito.Matchers.anyString;
import static org.mockito.Matchers.eq;
@@ -31,28 +34,44 @@ import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
+import java.io.ByteArrayInputStream;
+import java.io.DataInputStream;
import java.io.IOException;
+import java.net.HttpURLConnection;
+import java.net.URL;
import java.util.Arrays;
import java.util.List;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.tez.common.counters.TezCounters;
import org.apache.tez.dag.api.TezConfiguration;
import org.apache.tez.runtime.api.InputContext;
+import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
import org.apache.tez.runtime.library.common.InputAttemptIdentifier;
+import org.apache.tez.runtime.library.common.security.SecureShuffleUtils;
import org.apache.tez.runtime.library.common.sort.impl.TezIndexRecord;
+import org.apache.tez.runtime.library.exceptions.FetcherReadTimeoutException;
+import org.apache.tez.runtime.library.shuffle.common.HttpConnection;
+import org.apache.tez.runtime.library.shuffle.common.ShuffleUtils;
import org.junit.Assert;
import org.junit.Test;
import org.mockito.ArgumentCaptor;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
+import javax.crypto.SecretKey;
+
public class TestFetcher {
public static final String SHUFFLE_INPUT_FILE_PREFIX = "shuffle_input_file_";
public static final String HOST = "localhost";
public static final int PORT = 0;
+ static final Log LOG = LogFactory.getLog(TestFetcher.class);
+
@Test(timeout = 5000)
public void testSetupLocalDiskFetch() throws Exception {
Configuration conf = new TezConfiguration();
@@ -69,7 +88,7 @@ public class TestFetcher {
Fetcher spyFetcher = spy(fetcher);
MapHost host = new MapHost(1, HOST + ":" + PORT,
- "http://" + HOST + ":" +PORT + "/mapOutput?job=job_123&&reduce=1&map=");
+ "http://" + HOST + ":" + PORT + "/mapOutput?job=job_123&&reduce=1&map=");
List<InputAttemptIdentifier> srcAttempts = Arrays.asList(
new InputAttemptIdentifier(0, 1, InputAttemptIdentifier.PATH_PREFIX + "pathComponent_0"),
new InputAttemptIdentifier(1, 2, InputAttemptIdentifier.PATH_PREFIX + "pathComponent_1"),
@@ -79,7 +98,7 @@ public class TestFetcher {
);
final int FIRST_FAILED_ATTEMPT_IDX = 2;
final int SECOND_FAILED_ATTEMPT_IDX = 4;
- final int[] sucessfulAttemptsIndexes = {0, 1, 3};
+ final int[] sucessfulAttemptsIndexes = { 0, 1, 3 };
doReturn(srcAttempts).when(scheduler).getMapsForHost(host);
@@ -144,10 +163,10 @@ public class TestFetcher {
}
private void verifyCopySucceeded(ShuffleScheduler scheduler, MapHost host,
- List<InputAttemptIdentifier> srcAttempts, long p) throws
+ List<InputAttemptIdentifier> srcAttempts, long p) throws
IOException {
// need to verify filename, offsets, sizes wherever they are used.
- InputAttemptIdentifier srcAttemptToMatch = srcAttempts.get((int)p);
+ InputAttemptIdentifier srcAttemptToMatch = srcAttempts.get((int) p);
String filenameToMatch = SHUFFLE_INPUT_FILE_PREFIX + srcAttemptToMatch.getPathComponent();
ArgumentCaptor<MapOutput> captureMapOutput = ArgumentCaptor.forClass(MapOutput.class);
verify(scheduler).copySucceeded(eq(srcAttemptToMatch), eq(host), eq(p * 100),
@@ -156,6 +175,122 @@ public class TestFetcher {
// cannot use the equals of MapOutput as it compares id which is private. so doing it manually
MapOutput m = captureMapOutput.getAllValues().get(0);
Assert.assertTrue(m.getType().equals(MapOutput.Type.DISK_DIRECT) &&
- m.getAttemptIdentifier().equals(srcAttemptToMatch));
+ m.getAttemptIdentifier().equals(srcAttemptToMatch));
+ }
+
+ static class FakeHttpConnection extends HttpConnection {
+
+ public FakeHttpConnection(URL url,
+ HttpConnectionParams connParams, String logIdentifier, SecretKey jobTokenSecret)
+ throws IOException {
+ super(url, connParams, logIdentifier, jobTokenSecret);
+ this.connection = mock(HttpURLConnection.class);
+ when(connection.getResponseCode()).thenReturn(200);
+ when(connection.getHeaderField(ShuffleHeader.HTTP_HEADER_NAME))
+ .thenReturn(ShuffleHeader.DEFAULT_HTTP_HEADER_NAME);
+ when(connection.getHeaderField(ShuffleHeader.HTTP_HEADER_VERSION))
+ .thenReturn(ShuffleHeader.DEFAULT_HTTP_HEADER_VERSION);
+ when(connection.getHeaderField(SecureShuffleUtils.HTTP_HEADER_REPLY_URL_HASH)).thenReturn("");
+ }
+
+ public DataInputStream getInputStream() throws IOException {
+ byte[] b = new byte[1024];
+ ByteArrayInputStream bin = new ByteArrayInputStream(b);
+ return new DataInputStream(bin);
+ }
+
+ public void validate() throws IOException {
+ //empty
+ }
+
+ public void cleanup(boolean disconnect) throws IOException {
+ LOG.info("HttpConnection cleanup called with disconnect=" + disconnect);
+ //ignore
+ }
}
+
+ @Test
+ public void testWithRetry() throws Exception {
+ Configuration conf = new TezConfiguration();
+ conf.setInt(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_READ_TIMEOUT, 3000);
+ conf.setInt(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_CONNECT_TIMEOUT, 3000);
+ ShuffleScheduler scheduler = mock(ShuffleScheduler.class);
+ MergeManager merger = mock(MergeManager.class);
+
+ ShuffleClientMetrics metrics = mock(ShuffleClientMetrics.class);
+ Shuffle shuffle = mock(Shuffle.class);
+ InputContext inputContext = mock(InputContext.class);
+ when(inputContext.getCounters()).thenReturn(new TezCounters());
+ when(inputContext.getSourceVertexName()).thenReturn("");
+
+ HttpConnection.HttpConnectionParams httpConnectionParams =
+ ShuffleUtils.constructHttpShuffleConnectionParams(conf);
+ Fetcher mockFetcher =
+ new Fetcher(httpConnectionParams, scheduler, merger, metrics, shuffle, null,
+ false, 0, null, inputContext, conf, false);
+ final Fetcher fetcher = spy(mockFetcher);
+
+ final MapHost host = new MapHost(1, HOST + ":" + PORT,
+ "http://" + HOST + ":" + PORT + "/mapOutput?job=job_123&&reduce=1&map=");
+ final List<InputAttemptIdentifier> srcAttempts = Arrays.asList(
+ new InputAttemptIdentifier(0, 1, InputAttemptIdentifier.PATH_PREFIX + "pathComponent_0"),
+ new InputAttemptIdentifier(1, 2, InputAttemptIdentifier.PATH_PREFIX + "pathComponent_1"),
+ new InputAttemptIdentifier(3, 4, InputAttemptIdentifier.PATH_PREFIX + "pathComponent_3")
+ );
+ doReturn(srcAttempts).when(scheduler).getMapsForHost(host);
+ doReturn(true).when(fetcher).setupConnection(host, srcAttempts);
+
+ URL url = ShuffleUtils.constructInputURL(host.getBaseUrl(), srcAttempts, false);
+ fetcher.httpConnection = new FakeHttpConnection(url, null, "", null);
+
+ doAnswer(new Answer<MapOutput>() {
+ @Override
+ public MapOutput answer(InvocationOnMock invocation) throws Throwable {
+ Object[] args = invocation.getArguments();
+ MapOutput mapOutput = mock(MapOutput.class);
+ doReturn(MapOutput.Type.MEMORY).when(mapOutput).getType();
+ doReturn(args[0]).when(mapOutput).getAttemptIdentifier();
+ return mapOutput;
+ }
+ }).when(merger).reserve(any(InputAttemptIdentifier.class), anyInt(), anyInt(), anyInt());
+
+ //Create read timeout when reading data
+ doAnswer(new Answer<Void>() {
+ @Override public Void answer(InvocationOnMock invocation) throws Throwable {
+ // Emulate host down for 4 seconds.
+ Thread.sleep(4000);
+ doReturn(false).when(fetcher).setupConnection(host, srcAttempts);
+
+ // Throw IOException when fetcher tries to connect again to the same node
+ throw new FetcherReadTimeoutException("creating fetcher socket read timeout exception");
+ }
+ }).when(fetcher).copyMapOutput(any(MapHost.class), any(DataInputStream.class));
+
+ try {
+ fetcher.copyFromHost(host);
+ } catch(IOException e) {
+ //ignore
+ }
+ //setup connection should be called twice (1 for connect and another for retry)
+ verify(fetcher, times(2)).setupConnection(any(MapHost.class), anyList());
+ //since copyMapOutput consistently fails, it should call copyFailed once
+ verify(scheduler, times(1)).copyFailed(any(InputAttemptIdentifier.class), any(MapHost.class),
+ anyBoolean(), anyBoolean());
+
+ verify(fetcher, times(1)).putBackRemainingMapOutputs(any(MapHost.class));
+ verify(scheduler, times(3)).putBackKnownMapOutput(any(MapHost.class),
+ any(InputAttemptIdentifier.class));
+
+
+ //Verify by stopping the fetcher abruptly
+ try {
+ fetcher.stopped = false; // flag to indicate fetcher stopped
+ fetcher.copyFromHost(host);
+ verify(fetcher, times(2)).putBackRemainingMapOutputs(any(MapHost.class));
+ } catch(IOException e) {
+ //ignore
+ }
+
+ }
+
}