Repository: tez Updated Branches: refs/heads/master 7be325eab -> 9dabf9476
http://git-wip-us.apache.org/repos/asf/tez/blob/9dabf947/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/FetcherOrderedGrouped.java ---------------------------------------------------------------------- diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/FetcherOrderedGrouped.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/FetcherOrderedGrouped.java index 0248f13..a4d38ce 100644 --- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/FetcherOrderedGrouped.java +++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/FetcherOrderedGrouped.java @@ -27,6 +27,8 @@ import java.util.LinkedList; import java.util.List; import java.util.concurrent.atomic.AtomicInteger; +import org.apache.tez.http.BaseHttpConnection; +import org.apache.tez.http.HttpConnectionParams; import org.apache.tez.common.CallableWithNdc; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -43,8 +45,6 @@ import org.apache.tez.runtime.library.common.shuffle.orderedgrouped.MapOutput.Ty 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.common.shuffle.HttpConnection; -import org.apache.tez.runtime.library.common.shuffle.HttpConnection.HttpConnectionParams; import org.apache.tez.runtime.library.common.shuffle.ShuffleUtils; import com.google.common.annotations.VisibleForTesting; @@ -86,11 +86,13 @@ class FetcherOrderedGrouped extends CallableWithNdc<Void> { private final boolean ifileReadAhead; private final int ifileReadAheadLength; - private LinkedList<InputAttemptIdentifier> remaining; + @VisibleForTesting + LinkedList<InputAttemptIdentifier> remaining; volatile DataInputStream input; - volatile HttpConnection httpConnection; + volatile BaseHttpConnection httpConnection; + private final boolean asyncHttp; // Initiative value is 0, which means it hasn't retried yet. @@ -114,7 +116,8 @@ class FetcherOrderedGrouped extends CallableWithNdc<Void> { TezCounter badIdErrsCounter, TezCounter wrongMapErrsCounter, TezCounter connectionErrsCounter, - TezCounter wrongReduceErrsCounter) { + TezCounter wrongReduceErrsCounter, + boolean asyncHttp) { this.scheduler = scheduler; this.allocator = allocator; this.metrics = metrics; @@ -134,6 +137,7 @@ class FetcherOrderedGrouped extends CallableWithNdc<Void> { this.ifileReadAhead = ifileReadAhead; this.ifileReadAheadLength = ifileReadAheadLength; this.httpConnectionParams = httpConnectionParams; + this.asyncHttp = asyncHttp; if (codec != null) { this.codec = codec; } else { @@ -311,8 +315,8 @@ class FetcherOrderedGrouped extends CallableWithNdc<Void> { boolean connectSucceeded = false; try { URL url = ShuffleUtils.constructInputURL(host.getBaseUrl(), attempts, - httpConnectionParams.getKeepAlive()); - httpConnection = new HttpConnection(url, httpConnectionParams, + httpConnectionParams.isKeepAlive()); + httpConnection = ShuffleUtils.getHttpConnection(asyncHttp, url, httpConnectionParams, logIdentifier, jobTokenSecretManager); connectSucceeded = httpConnection.connect(); @@ -323,7 +327,10 @@ class FetcherOrderedGrouped extends CallableWithNdc<Void> { input = httpConnection.getInputStream(); httpConnection.validate(); return true; - } catch (IOException ie) { + } catch (IOException | InterruptedException ie) { + if (ie instanceof InterruptedException) { + Thread.currentThread().interrupt(); //reset status + } if (stopped) { LOG.info("Not reporting fetch failure, since an Exception was caught after shutdown"); return false; http://git-wip-us.apache.org/repos/asf/tez/blob/9dabf947/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/ShuffleScheduler.java ---------------------------------------------------------------------- diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/ShuffleScheduler.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/ShuffleScheduler.java index 85c3a30..75dca64 100644 --- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/ShuffleScheduler.java +++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/ShuffleScheduler.java @@ -54,10 +54,10 @@ import com.google.common.util.concurrent.MoreExecutors; import com.google.common.util.concurrent.ThreadFactoryBuilder; import org.apache.hadoop.io.compress.CompressionCodec; import org.apache.hadoop.security.UserGroupInformation; +import org.apache.tez.http.HttpConnectionParams; import org.apache.tez.common.CallableWithNdc; import org.apache.tez.common.security.JobTokenSecretManager; import org.apache.tez.dag.api.TezConstants; -import org.apache.tez.runtime.library.common.shuffle.HttpConnection; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.apache.hadoop.conf.Configuration; @@ -144,7 +144,7 @@ class ShuffleScheduler { private final ListeningExecutorService fetcherExecutor; - private final HttpConnection.HttpConnectionParams httpConnectionParams; + private final HttpConnectionParams httpConnectionParams; private final FetchedInputAllocatorOrderedGrouped allocator; private final ShuffleClientMetrics shuffleMetrics; private final Shuffle shuffle; @@ -157,6 +157,7 @@ class ShuffleScheduler { private final boolean localDiskFetchEnabled; private final String localHostname; private final int shufflePort; + private final boolean asyncHttp; private final TezCounter ioErrsCounter; private final TezCounter wrongLengthErrsCounter; @@ -245,8 +246,8 @@ class ShuffleScheduler { this.startTime = startTime; this.lastProgressTime = startTime; - this.httpConnectionParams = - ShuffleUtils.constructHttpShuffleConnectionParams(conf); + this.asyncHttp = conf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_USE_ASYNC_HTTP, false); + this.httpConnectionParams = ShuffleUtils.getHttpConnectionParams(conf); this.shuffleMetrics = new ShuffleClientMetrics(inputContext.getDAGName(), inputContext.getTaskVertexName(), inputContext.getTaskIndex(), this.conf, UserGroupInformation.getCurrentUser().getShortUserName()); @@ -1016,7 +1017,7 @@ class ShuffleScheduler { shuffleMetrics, shuffle, jobTokenSecretManager, ifileReadAhead, ifileReadAheadLength, codec, conf, localDiskFetchEnabled, localHostname, shufflePort, srcNameTrimmed, mapHost, ioErrsCounter, wrongLengthErrsCounter, badIdErrsCounter, wrongMapErrsCounter, - connectionErrsCounter, wrongReduceErrsCounter); + connectionErrsCounter, wrongReduceErrsCounter, asyncHttp); } private class FetchFutureCallback implements FutureCallback<Void> { http://git-wip-us.apache.org/repos/asf/tez/blob/9dabf947/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/input/OrderedGroupedKVInput.java ---------------------------------------------------------------------- diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/input/OrderedGroupedKVInput.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/input/OrderedGroupedKVInput.java index 12a5955..7399359 100644 --- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/input/OrderedGroupedKVInput.java +++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/input/OrderedGroupedKVInput.java @@ -330,6 +330,7 @@ public class OrderedGroupedKVInput extends AbstractLogicalInput { confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_FACTOR); confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_COMBINE_MIN_SPILLS); confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_COMBINER_CLASS); + confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_USE_ASYNC_HTTP); confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_PARALLEL_COPIES); confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_FETCH_FAILURES_LIMIT); confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_FETCH_MAX_TASK_OUTPUT_AT_ONCE); http://git-wip-us.apache.org/repos/asf/tez/blob/9dabf947/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/input/UnorderedKVInput.java ---------------------------------------------------------------------- diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/input/UnorderedKVInput.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/input/UnorderedKVInput.java index 7fc9317..1016263 100644 --- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/input/UnorderedKVInput.java +++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/input/UnorderedKVInput.java @@ -242,6 +242,7 @@ public class UnorderedKVInput extends AbstractLogicalInput { confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_BYTES); confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_IO_FILE_BUFFER_SIZE); confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_FACTOR); + confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_USE_ASYNC_HTTP); confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_PARALLEL_COPIES); confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_FETCH_FAILURES_LIMIT); confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_FETCH_MAX_TASK_OUTPUT_AT_ONCE); http://git-wip-us.apache.org/repos/asf/tez/blob/9dabf947/tez-runtime-library/src/test/java/org/apache/tez/http/TestHttpConnection.java ---------------------------------------------------------------------- diff --git a/tez-runtime-library/src/test/java/org/apache/tez/http/TestHttpConnection.java b/tez-runtime-library/src/test/java/org/apache/tez/http/TestHttpConnection.java new file mode 100644 index 0000000..8f0a7ad --- /dev/null +++ b/tez-runtime-library/src/test/java/org/apache/tez/http/TestHttpConnection.java @@ -0,0 +1,202 @@ +/** + * 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 + * <p/> + * http://www.apache.org/licenses/LICENSE-2.0 + * <p/> + * 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.http; + +import com.google.common.base.Throwables; +import org.apache.tez.http.async.netty.AsyncHttpConnection; +import org.apache.tez.common.security.JobTokenSecretManager; +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.Test; +import org.mockito.invocation.InvocationOnMock; +import org.mockito.stubbing.Answer; + +import java.io.IOException; +import java.net.URISyntaxException; +import java.net.URL; +import java.nio.channels.ClosedByInterruptException; +import java.util.concurrent.Callable; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.ThreadFactory; + +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.when; + +public class TestHttpConnection { + + private static int connTimeout = 5000; + private static int readTimeout = 5000; + + private static final String NOT_HOSTED_URL = "http://10.255.255.255:10221";; + + private static ExecutorService executorService; + private static URL url; + private static JobTokenSecretManager tokenSecretManager; + + private Thread currentThread; + + @BeforeClass + public static void setup() throws IOException, URISyntaxException { + executorService = Executors.newFixedThreadPool(1, + new ThreadFactory() { + public Thread newThread(Runnable r) { + Thread t = Executors.defaultThreadFactory().newThread(r); + t.setDaemon(true); + return t; + } + }); + url = new URL(NOT_HOSTED_URL); + tokenSecretManager = mock(JobTokenSecretManager.class); + } + + @AfterClass + public static void cleanup() throws Exception { + executorService.shutdownNow(); + } + + public void baseTest(Callable<Void> worker, CountDownLatch latch, String message) throws + InterruptedException { + long startTime = System.currentTimeMillis(); + try { + Future future = executorService.submit(worker); + future.get(); + } catch (ExecutionException e) { + assertTrue(e.getCause().getCause() instanceof IOException); + assertTrue(e.getMessage(), e.getMessage().contains(message)); + long elapsedTime = System.currentTimeMillis() - startTime; + assertTrue("elapasedTime=" + elapsedTime + " should be greater than " + connTimeout, + elapsedTime > connTimeout); + } + assertTrue(latch.getCount() == 0); + } + + @Test(timeout = 20000) + public void testConnectionTimeout() throws IOException, InterruptedException { + HttpConnectionParams params = getConnectionParams(); + + //For http + CountDownLatch latch = new CountDownLatch(1); + HttpConnection httpConn = getHttpConnection(params); + baseTest(new Worker(latch, httpConn, false), latch, "Failed to connect"); + + //For async http + latch = new CountDownLatch(1); + AsyncHttpConnection asyncHttpConn = getAsyncHttpConnection(params); + baseTest(new Worker(latch, asyncHttpConn, false), latch, "connection timed out"); + } + + @Test(timeout = 20000) + @SuppressWarnings("unchecked") + //Should be interruptible + public void testAsyncHttpConnectionInterrupt() + throws IOException, InterruptedException, ExecutionException { + CountDownLatch latch = new CountDownLatch(1); + HttpConnectionParams params = getConnectionParams(); + AsyncHttpConnection asyncHttpConn = getAsyncHttpConnection(params); + Future future = executorService.submit(new Worker(latch, asyncHttpConn, true)); + + while(currentThread == null) { + synchronized (this) { + wait(100); + } + } + + assertTrue("currentThread is still null", currentThread != null); + + //Try interrupting the thread (exception verification happens in the worker itself) + currentThread.interrupt(); + + future.get(); + assertTrue(latch.getCount() == 0); + } + + HttpConnectionParams getConnectionParams() { + HttpConnectionParams params = mock(HttpConnectionParams.class); + when(params.getBufferSize()).thenReturn(8192); + when(params.getKeepAliveMaxConnections()).thenReturn(1); + when(params.getConnectionTimeout()).thenReturn(connTimeout); + when(params.getReadTimeout()).thenReturn(readTimeout); + return params; + } + + HttpConnection getHttpConnection(HttpConnectionParams params) throws IOException { + HttpConnection realConn = new HttpConnection(url, params, "log", tokenSecretManager); + HttpConnection connection = spy(realConn); + + doAnswer(new Answer() { + public Void answer(InvocationOnMock invocation) { + return null; + } + }).when(connection).computeEncHash(); + return connection; + } + + AsyncHttpConnection getAsyncHttpConnection(HttpConnectionParams params) throws IOException { + AsyncHttpConnection realConn = new AsyncHttpConnection(url, params, "log", tokenSecretManager); + AsyncHttpConnection connection = spy(realConn); + + doAnswer(new Answer() { + public Void answer(InvocationOnMock invocation) { + return null; + } + }).when(connection).computeEncHash(); + return connection; + } + + class Worker implements Callable<Void> { + private CountDownLatch latch; + private BaseHttpConnection connection; + private boolean expectingInterrupt; + + public Worker(CountDownLatch latch, BaseHttpConnection connection, boolean expectingInterrupt) { + this.latch = latch; + this.connection = connection; + this.expectingInterrupt = expectingInterrupt; + } + + @Override + public Void call() throws Exception { + try { + currentThread = Thread.currentThread(); + connection.connect(); + fail(); + } catch(Throwable t) { + if (expectingInterrupt) { + //ClosedByInterruptException normally; InterruptedException if + // TezBodyDeferringAsyncHandler quits otherwise + assertTrue((t instanceof InterruptedException) || (t instanceof ClosedByInterruptException)); + } + } finally { + latch.countDown(); + if (connection != null) { + connection.cleanup(true); + } + } + return null; + } + } +} http://git-wip-us.apache.org/repos/asf/tez/blob/9dabf947/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/TestFetcher.java ---------------------------------------------------------------------- diff --git a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/TestFetcher.java b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/TestFetcher.java index 4ef187d..34c2ca7 100644 --- a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/TestFetcher.java +++ b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/TestFetcher.java @@ -70,7 +70,8 @@ public class TestFetcher { final boolean DISABLE_LOCAL_FETCH = false; Fetcher.FetcherBuilder builder = new Fetcher.FetcherBuilder(fetcherCallback, null, null, - ApplicationId.newInstance(0, 1), null, "fetcherTest", conf, ENABLE_LOCAL_FETCH, HOST, PORT); + ApplicationId.newInstance(0, 1), null, "fetcherTest", conf, ENABLE_LOCAL_FETCH, HOST, + PORT, false); builder.assignWork(HOST, PORT, 0, Arrays.asList(srcAttempts)); Fetcher fetcher = spy(builder.build()); @@ -88,7 +89,7 @@ public class TestFetcher { // when enabled and hostname does not match use http fetch. builder = new Fetcher.FetcherBuilder(fetcherCallback, null, null, ApplicationId.newInstance(0, 1), null, "fetcherTest", conf, ENABLE_LOCAL_FETCH, HOST, - PORT); + PORT, false); builder.assignWork(HOST + "_OTHER", PORT, 0, Arrays.asList(srcAttempts)); fetcher = spy(builder.build()); @@ -103,7 +104,7 @@ public class TestFetcher { // when enabled and port does not match use http fetch. builder = new Fetcher.FetcherBuilder(fetcherCallback, null, null, - ApplicationId.newInstance(0, 1), null, "fetcherTest", conf, ENABLE_LOCAL_FETCH, HOST, PORT); + ApplicationId.newInstance(0, 1), null, "fetcherTest", conf, ENABLE_LOCAL_FETCH, HOST, PORT, false); builder.assignWork(HOST, PORT + 1, 0, Arrays.asList(srcAttempts)); fetcher = spy(builder.build()); @@ -119,7 +120,8 @@ public class TestFetcher { // When disabled use http fetch conf.setBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_OPTIMIZE_LOCAL_FETCH, false); builder = new Fetcher.FetcherBuilder(fetcherCallback, null, null, - ApplicationId.newInstance(0, 1), null, "fetcherTest", conf, DISABLE_LOCAL_FETCH, HOST, PORT); + ApplicationId.newInstance(0, 1), null, "fetcherTest", conf, DISABLE_LOCAL_FETCH, HOST, + PORT, false); builder.assignWork(HOST, PORT, 0, Arrays.asList(srcAttempts)); fetcher = spy(builder.build()); @@ -152,7 +154,7 @@ public class TestFetcher { int partition = 42; FetcherCallback callback = mock(FetcherCallback.class); Fetcher.FetcherBuilder builder = new Fetcher.FetcherBuilder(callback, null, null, - ApplicationId.newInstance(0, 1), null, "fetcherTest", conf, true, HOST, PORT); + ApplicationId.newInstance(0, 1), null, "fetcherTest", conf, true, HOST, PORT, false); builder.assignWork(HOST, PORT, partition, Arrays.asList(srcAttempts)); Fetcher fetcher = spy(builder.build()); http://git-wip-us.apache.org/repos/asf/tez/blob/9dabf947/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/TestFetcher.java ---------------------------------------------------------------------- diff --git a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/TestFetcher.java b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/TestFetcher.java index f77e9a6..385b7b0 100644 --- a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/TestFetcher.java +++ b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/TestFetcher.java @@ -18,6 +18,7 @@ package org.apache.tez.runtime.library.common.shuffle.orderedgrouped; +import static org.junit.Assert.fail; import static org.mockito.Matchers.any; import static org.mockito.Matchers.anyBoolean; import static org.mockito.Matchers.anyInt; @@ -44,6 +45,8 @@ import java.util.Arrays; import java.util.List; import com.google.common.collect.Lists; +import org.apache.tez.http.HttpConnection; +import org.apache.tez.http.HttpConnectionParams; import org.apache.tez.common.counters.TezCounter; import org.apache.tez.runtime.library.common.InputIdentifier; import org.slf4j.Logger; @@ -59,7 +62,6 @@ 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.common.shuffle.HttpConnection; import org.apache.tez.runtime.library.common.shuffle.ShuffleUtils; import org.junit.Assert; import org.junit.Test; @@ -118,7 +120,7 @@ public class TestFetcher { new FetcherOrderedGrouped(null, scheduler, merger, metrics, shuffle, null, false, 0, null, conf, false, HOST, PORT, "src vertex", mapHost, ioErrsCounter, wrongLengthErrsCounter, badIdErrsCounter, - wrongMapErrsCounter, connectionErrsCounter, wrongReduceErrsCounter); + wrongMapErrsCounter, connectionErrsCounter, wrongReduceErrsCounter, false); fetcher.call(); verify(scheduler).getMapsForHost(mapHost); @@ -146,7 +148,7 @@ public class TestFetcher { new FetcherOrderedGrouped(null, scheduler, merger, metrics, shuffle, null, false, 0, null, conf, ENABLE_LOCAL_FETCH, HOST, PORT, "src vertex", mapHost, ioErrsCounter, wrongLengthErrsCounter, badIdErrsCounter, - wrongMapErrsCounter, connectionErrsCounter, wrongReduceErrsCounter); + wrongMapErrsCounter, connectionErrsCounter, wrongReduceErrsCounter, false); // when local mode is enabled and host and port matches use local fetch FetcherOrderedGrouped spyFetcher = spy(fetcher); @@ -163,7 +165,7 @@ public class TestFetcher { new FetcherOrderedGrouped(null, scheduler, merger, metrics, shuffle, null, false, 0, null, conf, ENABLE_LOCAL_FETCH, HOST, PORT, "src vertex", mapHost, ioErrsCounter, wrongLengthErrsCounter, badIdErrsCounter, - wrongMapErrsCounter, connectionErrsCounter, wrongReduceErrsCounter); + wrongMapErrsCounter, connectionErrsCounter, wrongReduceErrsCounter, false); spyFetcher = spy(fetcher); doNothing().when(spyFetcher).setupLocalDiskFetch(mapHost); @@ -178,7 +180,7 @@ public class TestFetcher { new FetcherOrderedGrouped(null, scheduler, merger, metrics, shuffle, null, false, 0, null, conf, ENABLE_LOCAL_FETCH, HOST, PORT, "src vertex", mapHost, ioErrsCounter, wrongLengthErrsCounter, badIdErrsCounter, - wrongMapErrsCounter, connectionErrsCounter, wrongReduceErrsCounter); + wrongMapErrsCounter, connectionErrsCounter, wrongReduceErrsCounter, false); spyFetcher = spy(fetcher); doNothing().when(spyFetcher).setupLocalDiskFetch(mapHost); @@ -192,7 +194,7 @@ public class TestFetcher { fetcher = new FetcherOrderedGrouped(null, scheduler, merger, metrics, shuffle, null, false, 0, null, conf, DISABLE_LOCAL_FETCH, HOST, PORT, "src vertex", mapHost, ioErrsCounter, wrongLengthErrsCounter, badIdErrsCounter, - wrongMapErrsCounter, connectionErrsCounter, wrongReduceErrsCounter); + wrongMapErrsCounter, connectionErrsCounter, wrongReduceErrsCounter, false); spyFetcher = spy(fetcher); doNothing().when(spyFetcher).setupLocalDiskFetch(mapHost); @@ -217,7 +219,7 @@ public class TestFetcher { "http://"; + HOST + ":" + PORT + "/mapOutput?job=job_123&&reduce=1&map="); FetcherOrderedGrouped fetcher = new FetcherOrderedGrouped(null, scheduler, merger, metrics, shuffle, null, false, 0, null, conf, true, HOST, PORT, "src vertex", host, ioErrsCounter, wrongLengthErrsCounter, badIdErrsCounter, - wrongMapErrsCounter, connectionErrsCounter, wrongReduceErrsCounter); + wrongMapErrsCounter, connectionErrsCounter, wrongReduceErrsCounter, false); FetcherOrderedGrouped spyFetcher = spy(fetcher); @@ -342,6 +344,7 @@ public class TestFetcher { } @Test(timeout = 5000) + @SuppressWarnings("unchecked") public void testWithRetry() throws Exception { Configuration conf = new TezConfiguration(); conf.setInt(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_READ_TIMEOUT, 3000); @@ -355,13 +358,12 @@ public class TestFetcher { when(inputContext.getCounters()).thenReturn(new TezCounters()); when(inputContext.getSourceVertexName()).thenReturn(""); - HttpConnection.HttpConnectionParams httpConnectionParams = - ShuffleUtils.constructHttpShuffleConnectionParams(conf); + HttpConnectionParams httpConnectionParams = ShuffleUtils.getHttpConnectionParams(conf); final MapHost host = new MapHost(1, HOST + ":" + PORT, "http://"; + HOST + ":" + PORT + "/mapOutput?job=job_123&&reduce=1&map="); FetcherOrderedGrouped mockFetcher = new FetcherOrderedGrouped(null, scheduler, merger, metrics, shuffle, null, false, 0, null, conf, false, HOST, PORT, "src vertex", host, ioErrsCounter, wrongLengthErrsCounter, badIdErrsCounter, - wrongMapErrsCounter, connectionErrsCounter, wrongReduceErrsCounter); + wrongMapErrsCounter, connectionErrsCounter, wrongReduceErrsCounter, false); final FetcherOrderedGrouped fetcher = spy(mockFetcher); @@ -426,4 +428,53 @@ public class TestFetcher { } + @Test + @SuppressWarnings("unchecked") + public void testAsyncWithException() 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); + + TezCounters counters = new TezCounters(); + InputContext inputContext = mock(InputContext.class); + when(inputContext.getCounters()).thenReturn(counters); + when(inputContext.getSourceVertexName()).thenReturn(""); + + JobTokenSecretManager jobMgr = mock(JobTokenSecretManager.class); + doReturn(new byte[10]).when(jobMgr).computeHash(any(byte[].class)); + + HttpConnectionParams httpConnectionParams = ShuffleUtils.getHttpConnectionParams(conf); + final MapHost host = new MapHost(1, HOST + ":" + PORT, + "http://"; + HOST + ":" + PORT + "/mapOutput?job=job_123&&reduce=1&map="); + FetcherOrderedGrouped mockFetcher = + new FetcherOrderedGrouped(httpConnectionParams, scheduler, merger, metrics, shuffle, jobMgr, + false, 0, + null, conf, false, HOST, PORT, "src vertex", host, ioErrsCounter, + wrongLengthErrsCounter, badIdErrsCounter, + wrongMapErrsCounter, connectionErrsCounter, wrongReduceErrsCounter, true); + final FetcherOrderedGrouped fetcher = spy(mockFetcher); + fetcher.remaining = Lists.newLinkedList(); + + 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); + + try { + long currentIOErrors = ioErrsCounter.getValue(); + boolean connected = fetcher.setupConnection(host, srcAttempts); + Assert.assertTrue(connected == false); + //Ensure that counters are incremented (i.e it followed the exception codepath) + Assert.assertTrue(ioErrsCounter.getValue() > currentIOErrors); + } catch (IOException e) { + fail(); + } + } } http://git-wip-us.apache.org/repos/asf/tez/blob/9dabf947/tez-tests/src/test/java/org/apache/tez/test/TestPipelinedShuffle.java ---------------------------------------------------------------------- diff --git a/tez-tests/src/test/java/org/apache/tez/test/TestPipelinedShuffle.java b/tez-tests/src/test/java/org/apache/tez/test/TestPipelinedShuffle.java index 2a63293..25c149d 100644 --- a/tez-tests/src/test/java/org/apache/tez/test/TestPipelinedShuffle.java +++ b/tez-tests/src/test/java/org/apache/tez/test/TestPipelinedShuffle.java @@ -94,6 +94,7 @@ public class TestPipelinedShuffle { } } + //TODO: Add support for async http clients @Before public void setupTezCluster() throws Exception { //With 1 MB sort buffer and with good amount of dataset, it would spill records @@ -126,8 +127,18 @@ public class TestPipelinedShuffle { @Test public void baseTest() throws Exception { + Configuration conf = new Configuration(miniTezCluster.getConfig()); + conf.setBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_USE_ASYNC_HTTP, false); + test(conf); + + conf = new Configuration(miniTezCluster.getConfig()); + conf.setBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_USE_ASYNC_HTTP, true); + test(conf); + } + + private void test(Configuration conf) throws Exception { PipelinedShuffleJob pipelinedShuffle = new PipelinedShuffleJob(); - pipelinedShuffle.setConf(new Configuration(miniTezCluster.getConfig())); + pipelinedShuffle.setConf(conf); String[] args = new String[] { }; assertEquals(0, pipelinedShuffle.run(args)); http://git-wip-us.apache.org/repos/asf/tez/blob/9dabf947/tez-tests/src/test/java/org/apache/tez/test/TestSecureShuffle.java ---------------------------------------------------------------------- diff --git a/tez-tests/src/test/java/org/apache/tez/test/TestSecureShuffle.java b/tez-tests/src/test/java/org/apache/tez/test/TestSecureShuffle.java index 8da3f08..e3e42d3 100644 --- a/tez-tests/src/test/java/org/apache/tez/test/TestSecureShuffle.java +++ b/tez-tests/src/test/java/org/apache/tez/test/TestSecureShuffle.java @@ -36,6 +36,7 @@ import org.apache.hadoop.hdfs.MiniDFSCluster; import org.apache.hadoop.hdfs.server.namenode.EditLogFileOutputStream; import org.apache.hadoop.mapreduce.MRConfig; import org.apache.hadoop.security.ssl.KeyStoreTestUtil; +import org.apache.hadoop.yarn.conf.YarnConfiguration; import org.apache.tez.mapreduce.examples.TestOrderedWordCount; import org.apache.tez.runtime.library.api.TezRuntimeConfiguration; import org.junit.After; @@ -65,22 +66,30 @@ public class TestSecureShuffle { private boolean enableSSLInCluster; //To set ssl config in cluster private int resultWithTezSSL; //expected result with tez ssl setting private int resultWithoutTezSSL; //expected result without tez ssl setting + private boolean asyncHttp; - public TestSecureShuffle(boolean sslInCluster, int resultWithTezSSL, int resultWithoutTezSSL) { + public TestSecureShuffle(boolean sslInCluster, int resultWithTezSSL, int resultWithoutTezSSL, + boolean asyncHttp) { this.enableSSLInCluster = sslInCluster; this.resultWithTezSSL = resultWithTezSSL; this.resultWithoutTezSSL = resultWithoutTezSSL; + this.asyncHttp = asyncHttp; } - @Parameterized.Parameters(name = "test[sslInCluster:{0}, resultWithTezSSL:{1}, resultWithoutTezSSL:{2}]") + @Parameterized.Parameters(name = "test[sslInCluster:{0}, resultWithTezSSL:{1}, " + + "resultWithoutTezSSL:{2}, asyncHttp:{3}]") public static Collection<Object[]> getParameters() { Collection<Object[]> parameters = new ArrayList<Object[]>(); //enable ssl in cluster, succeed with tez-ssl enabled, fail with tez-ssl disabled - parameters.add(new Object[] { true, 0, 1 }); + parameters.add(new Object[] { true, 0, 1, false }); + + //With asyncHttp + parameters.add(new Object[] { true, 0, 1, true }); + parameters.add(new Object[] { false, 1, 0, true }); //Negative testcase - // disable ssl in cluster, fail with tez-ssl enabled, succeed with tez-ssl disabled - parameters.add(new Object[] { false, 1, 0 }); + //disable ssl in cluster, fail with tez-ssl enabled, succeed with tez-ssl disabled + parameters.add(new Object[] { false, 1, 0, false }); return parameters; } @@ -151,7 +160,7 @@ public class TestSecureShuffle { * * @throws Exception */ - @Test(timeout = 240000) + @Test(timeout = 500000) public void testSecureShuffle() throws Exception { //With tez-ssl setting miniTezCluster.getConfig().setBoolean(
