You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-issues@hadoop.apache.org by GitBox <gi...@apache.org> on 2023/01/18 16:35:44 UTC
[GitHub] [hadoop] tomicooler opened a new pull request, #5311: MAPREDUCE-7431. ShuffleHandler refactor and fix after Netty4 upgrade.
tomicooler opened a new pull request, #5311:
URL: https://github.com/apache/hadoop/pull/5311
Change-Id: Ifedad2fae1ddd8f22623b5d44875b20a3b3fd318
### Description of PR
Work in Progress. No review is needed yet, I'll update when I'm ready with my manual end-to-end tests.
### How was this patch tested?
In progress...
### For code changes:
- [x] Does the title or this PR starts with the corresponding JIRA issue id (e.g. 'HADOOP-17799. Your PR title ...')?
- [ ] Object storage: have the integration tests been executed and the endpoint declared according to the connector-specific documentation?
- [ ] If adding new dependencies to the code, are these dependencies licensed in a way that is compatible for inclusion under [ASF 2.0](http://www.apache.org/legal/resolved.html#category-a)?
I added ch.qos.logback for just a test dependency. LGPL.
- [ ] If applicable, have you updated the `LICENSE`, `LICENSE-binary`, `NOTICE-binary` files?
--
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: common-issues-unsubscribe@hadoop.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: common-issues-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-issues-help@hadoop.apache.org
[GitHub] [hadoop] brumi1024 merged pull request #5311: MAPREDUCE-7431. ShuffleHandler refactor and fix after Netty4 upgrade.
Posted by "brumi1024 (via GitHub)" <gi...@apache.org>.
brumi1024 merged PR #5311:
URL: https://github.com/apache/hadoop/pull/5311
--
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: common-issues-unsubscribe@hadoop.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: common-issues-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-issues-help@hadoop.apache.org
[GitHub] [hadoop] tomicooler commented on a diff in pull request #5311: MAPREDUCE-7431. ShuffleHandler refactor and fix after Netty4 upgrade.
Posted by GitBox <gi...@apache.org>.
tomicooler commented on code in PR #5311:
URL: https://github.com/apache/hadoop/pull/5311#discussion_r1081262415
##########
hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-shuffle/src/main/java/org/apache/hadoop/mapred/ShuffleChannelHandler.java:
##########
@@ -0,0 +1,715 @@
+/*
+ * 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.hadoop.mapred;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.handler.codec.TooLongFrameException;
+import io.netty.handler.codec.http.DefaultFullHttpResponse;
+import io.netty.handler.codec.http.DefaultHttpResponse;
+import io.netty.handler.codec.http.FullHttpRequest;
+import io.netty.handler.codec.http.FullHttpResponse;
+import io.netty.handler.codec.http.HttpRequest;
+import io.netty.handler.codec.http.HttpResponse;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.netty.handler.codec.http.HttpUtil;
+import io.netty.handler.codec.http.LastHttpContent;
+import io.netty.handler.codec.http.QueryStringDecoder;
+import io.netty.handler.ssl.SslHandler;
+import io.netty.util.CharsetUtil;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.RandomAccessFile;
+import java.net.URL;
+import java.nio.channels.ClosedChannelException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import javax.crypto.SecretKey;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.DataOutputBuffer;
+import org.apache.hadoop.io.SecureIOUtils;
+import org.apache.hadoop.mapreduce.security.SecureShuffleUtils;
+import org.apache.hadoop.mapreduce.task.reduce.ShuffleHeader;
+import org.apache.hadoop.thirdparty.com.google.common.base.Charsets;
+import org.eclipse.jetty.http.HttpHeader;
+
+import static io.netty.buffer.Unpooled.wrappedBuffer;
+import static io.netty.handler.codec.http.HttpHeaderNames.CONTENT_TYPE;
+import static io.netty.handler.codec.http.HttpMethod.GET;
+import static io.netty.handler.codec.http.HttpResponseStatus.BAD_REQUEST;
+import static io.netty.handler.codec.http.HttpResponseStatus.FORBIDDEN;
+import static io.netty.handler.codec.http.HttpResponseStatus.INTERNAL_SERVER_ERROR;
+import static io.netty.handler.codec.http.HttpResponseStatus.METHOD_NOT_ALLOWED;
+import static io.netty.handler.codec.http.HttpResponseStatus.OK;
+import static io.netty.handler.codec.http.HttpResponseStatus.UNAUTHORIZED;
+import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1;
+import static org.apache.hadoop.mapred.ShuffleHandler.AUDITLOG;
+import static org.apache.hadoop.mapred.ShuffleHandler.CONNECTION_CLOSE;
+import static org.apache.hadoop.mapred.ShuffleHandler.FETCH_RETRY_DELAY;
+import static org.apache.hadoop.mapred.ShuffleHandler.IGNORABLE_ERROR_MESSAGE;
+import static org.apache.hadoop.mapred.ShuffleHandler.RETRY_AFTER_HEADER;
+import static org.apache.hadoop.mapred.ShuffleHandler.TIMEOUT_HANDLER;
+import static org.apache.hadoop.mapred.ShuffleHandler.TOO_MANY_REQ_STATUS;
+import static org.apache.hadoop.mapred.ShuffleHandler.LOG;
+
+/**
+ * ShuffleChannelHandler verifies the map request then servers the attempts in a http stream.
+ * Before each attempt a serialised ShuffleHeader object is written with the details.
+ *
+ * <pre>
+ * Example Request
+ * ===================
+ * GET /mapOutput?job=job_1111111111111_0001&reduce=0&
+ * map=attempt_1111111111111_0001_m_000001_0,
+ * attempt_1111111111111_0002_m_000002_0,
+ * attempt_1111111111111_0003_m_000003_0 HTTP/1.1
+ * name: mapreduce
+ * version: 1.0.0
+ * UrlHash: 9zS++qE0/7/D2l1Rg0TqRoSguAk=
+ *
+ * Example Response
+ * ===================
+ * HTTP/1.1 200 OK
+ * ReplyHash: GcuojWkAxXUyhZHPnwoV/MW2tGA=
+ * name: mapreduce
+ * version: 1.0.0
+ * connection: close
+ * content-length: 138
+ *
+ * +--------+-------------------------------------------------+----------------+
+ * |00000000| 25 61 74 74 65 6d 70 74 5f 31 31 31 31 31 31 31 |%attempt_1111111|
+ * |00000010| 31 31 31 31 31 31 5f 30 30 30 31 5f 6d 5f 30 30 |111111_0001_m_00|
+ * |00000020| 30 30 30 31 5f 30 05 0a 00 |0001_0... |
+ * +--------+-------------------------------------------------+----------------+
+ * |00000000| 61 61 61 61 61 |aaaaa |
+ * +--------+-------------------------------------------------+----------------+
+ * |00000000| 25 61 74 74 65 6d 70 74 5f 31 31 31 31 31 31 31 |%attempt_1111111|
+ * |00000010| 31 31 31 31 31 31 5f 30 30 30 32 5f 6d 5f 30 30 |111111_0002_m_00|
+ * |00000020| 30 30 30 32 5f 30 05 0a 00 |0002_0... |
+ * +--------+-------------------------------------------------+----------------+
+ * |00000000| 62 62 62 62 62 |bbbbb |
+ * +--------+-------------------------------------------------+----------------+
+ * |00000000| 25 61 74 74 65 6d 70 74 5f 31 31 31 31 31 31 31 |%attempt_1111111|
+ * |00000010| 31 31 31 31 31 31 5f 30 30 30 33 5f 6d 5f 30 30 |111111_0003_m_00|
+ * |00000020| 30 30 30 33 5f 30 05 0a 00 |0003_0... |
+ * +--------+-------------------------------------------------+----------------+
+ * |00000000| 63 63 63 63 63 |ccccc |
+ * +--------+-------------------------------------------------+----------------+
+ * </pre>
+ */
+public class ShuffleChannelHandler extends SimpleChannelInboundHandler<FullHttpRequest> {
+ private final ShuffleChannelHandlerContext handlerCtx;
+
+ ShuffleChannelHandler(ShuffleChannelHandlerContext ctx) {
+ handlerCtx = ctx;
+ }
+
+ private List<String> splitMaps(List<String> mapq) {
+ if (null == mapq) {
+ return null;
+ }
+ final List<String> ret = new ArrayList<>();
+ for (String s : mapq) {
+ Collections.addAll(ret, s.split(","));
+ }
+ return ret;
+ }
+
+ @Override
+ public void channelActive(ChannelHandlerContext ctx)
+ throws Exception {
+ LOG.debug("Executing channelActive; channel='{}'", ctx.channel().id());
+ int numConnections = handlerCtx.activeConnections.incrementAndGet();
+ if ((handlerCtx.maxShuffleConnections > 0) &&
+ (numConnections > handlerCtx.maxShuffleConnections)) {
+ LOG.info(String.format("Current number of shuffle connections (%d) is " +
+ "greater than the max allowed shuffle connections (%d)",
+ handlerCtx.allChannels.size(), handlerCtx.maxShuffleConnections));
+
+ Map<String, String> headers = new HashMap<>(1);
+ // notify fetchers to backoff for a while before closing the connection
+ // if the shuffle connection limit is hit. Fetchers are expected to
+ // handle this notification gracefully, that is, not treating this as a
+ // fetch failure.
+ headers.put(RETRY_AFTER_HEADER, String.valueOf(FETCH_RETRY_DELAY));
+ sendError(ctx, "", TOO_MANY_REQ_STATUS, headers);
+ } else {
+ super.channelActive(ctx);
+ handlerCtx.allChannels.add(ctx.channel());
+ LOG.debug("Added channel: {}, channel id: {}. Accepted number of connections={}",
+ ctx.channel(), ctx.channel().id(), handlerCtx.activeConnections.get());
+ }
+ }
+
+ @Override
+ public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+ LOG.debug("Executing channelInactive; channel='{}'", ctx.channel().id());
+ super.channelInactive(ctx);
+ int noOfConnections = handlerCtx.activeConnections.decrementAndGet();
+ LOG.debug("New value of Accepted number of connections={}", noOfConnections);
+ }
+
+ @Override
+ public void channelRead0(ChannelHandlerContext ctx, FullHttpRequest request) {
+ Channel channel = ctx.channel();
+ LOG.debug("Received HTTP request: {}, channel='{}'", request, channel.id());
+
+ if (request.method() != GET) {
+ sendError(ctx, METHOD_NOT_ALLOWED);
+ return;
+ }
+ // Check whether the shuffle version is compatible
+ String shuffleVersion = ShuffleHeader.DEFAULT_HTTP_HEADER_VERSION;
+ String httpHeaderName = ShuffleHeader.DEFAULT_HTTP_HEADER_NAME;
+ if (request.headers() != null) {
+ shuffleVersion = request.headers().get(ShuffleHeader.HTTP_HEADER_VERSION);
+ httpHeaderName = request.headers().get(ShuffleHeader.HTTP_HEADER_NAME);
+ LOG.debug("Received from request header: ShuffleVersion={} header name={}, channel id: {}",
+ shuffleVersion, httpHeaderName, channel.id());
+ }
+ if (request.headers() == null ||
+ !ShuffleHeader.DEFAULT_HTTP_HEADER_NAME.equals(httpHeaderName) ||
+ !ShuffleHeader.DEFAULT_HTTP_HEADER_VERSION.equals(shuffleVersion)) {
+ sendError(ctx, "Incompatible shuffle request version", BAD_REQUEST);
+ return;
+ }
+ final Map<String, List<String>> q =
+ new QueryStringDecoder(request.uri()).parameters();
+
+ final List<String> keepAliveList = q.get("keepAlive");
+ boolean keepAliveParam = false;
+ if (keepAliveList != null && keepAliveList.size() == 1) {
+ keepAliveParam = Boolean.parseBoolean(keepAliveList.get(0));
+ if (LOG.isDebugEnabled()) {
Review Comment:
I guess it was used like this to perform better when DEBUG log is not activated. (The log parameters are not evaluated). This is just copy-pasta code from the original solution, no change here.
##########
hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-shuffle/src/main/java/org/apache/hadoop/mapred/ShuffleChannelHandler.java:
##########
@@ -0,0 +1,715 @@
+/*
+ * 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.hadoop.mapred;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.handler.codec.TooLongFrameException;
+import io.netty.handler.codec.http.DefaultFullHttpResponse;
+import io.netty.handler.codec.http.DefaultHttpResponse;
+import io.netty.handler.codec.http.FullHttpRequest;
+import io.netty.handler.codec.http.FullHttpResponse;
+import io.netty.handler.codec.http.HttpRequest;
+import io.netty.handler.codec.http.HttpResponse;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.netty.handler.codec.http.HttpUtil;
+import io.netty.handler.codec.http.LastHttpContent;
+import io.netty.handler.codec.http.QueryStringDecoder;
+import io.netty.handler.ssl.SslHandler;
+import io.netty.util.CharsetUtil;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.RandomAccessFile;
+import java.net.URL;
+import java.nio.channels.ClosedChannelException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import javax.crypto.SecretKey;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.DataOutputBuffer;
+import org.apache.hadoop.io.SecureIOUtils;
+import org.apache.hadoop.mapreduce.security.SecureShuffleUtils;
+import org.apache.hadoop.mapreduce.task.reduce.ShuffleHeader;
+import org.apache.hadoop.thirdparty.com.google.common.base.Charsets;
+import org.eclipse.jetty.http.HttpHeader;
+
+import static io.netty.buffer.Unpooled.wrappedBuffer;
+import static io.netty.handler.codec.http.HttpHeaderNames.CONTENT_TYPE;
+import static io.netty.handler.codec.http.HttpMethod.GET;
+import static io.netty.handler.codec.http.HttpResponseStatus.BAD_REQUEST;
+import static io.netty.handler.codec.http.HttpResponseStatus.FORBIDDEN;
+import static io.netty.handler.codec.http.HttpResponseStatus.INTERNAL_SERVER_ERROR;
+import static io.netty.handler.codec.http.HttpResponseStatus.METHOD_NOT_ALLOWED;
+import static io.netty.handler.codec.http.HttpResponseStatus.OK;
+import static io.netty.handler.codec.http.HttpResponseStatus.UNAUTHORIZED;
+import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1;
+import static org.apache.hadoop.mapred.ShuffleHandler.AUDITLOG;
+import static org.apache.hadoop.mapred.ShuffleHandler.CONNECTION_CLOSE;
+import static org.apache.hadoop.mapred.ShuffleHandler.FETCH_RETRY_DELAY;
+import static org.apache.hadoop.mapred.ShuffleHandler.IGNORABLE_ERROR_MESSAGE;
+import static org.apache.hadoop.mapred.ShuffleHandler.RETRY_AFTER_HEADER;
+import static org.apache.hadoop.mapred.ShuffleHandler.TIMEOUT_HANDLER;
+import static org.apache.hadoop.mapred.ShuffleHandler.TOO_MANY_REQ_STATUS;
+import static org.apache.hadoop.mapred.ShuffleHandler.LOG;
+
+/**
+ * ShuffleChannelHandler verifies the map request then servers the attempts in a http stream.
+ * Before each attempt a serialised ShuffleHeader object is written with the details.
+ *
+ * <pre>
+ * Example Request
+ * ===================
+ * GET /mapOutput?job=job_1111111111111_0001&reduce=0&
+ * map=attempt_1111111111111_0001_m_000001_0,
+ * attempt_1111111111111_0002_m_000002_0,
+ * attempt_1111111111111_0003_m_000003_0 HTTP/1.1
+ * name: mapreduce
+ * version: 1.0.0
+ * UrlHash: 9zS++qE0/7/D2l1Rg0TqRoSguAk=
+ *
+ * Example Response
+ * ===================
+ * HTTP/1.1 200 OK
+ * ReplyHash: GcuojWkAxXUyhZHPnwoV/MW2tGA=
+ * name: mapreduce
+ * version: 1.0.0
+ * connection: close
+ * content-length: 138
+ *
+ * +--------+-------------------------------------------------+----------------+
+ * |00000000| 25 61 74 74 65 6d 70 74 5f 31 31 31 31 31 31 31 |%attempt_1111111|
+ * |00000010| 31 31 31 31 31 31 5f 30 30 30 31 5f 6d 5f 30 30 |111111_0001_m_00|
+ * |00000020| 30 30 30 31 5f 30 05 0a 00 |0001_0... |
+ * +--------+-------------------------------------------------+----------------+
+ * |00000000| 61 61 61 61 61 |aaaaa |
+ * +--------+-------------------------------------------------+----------------+
+ * |00000000| 25 61 74 74 65 6d 70 74 5f 31 31 31 31 31 31 31 |%attempt_1111111|
+ * |00000010| 31 31 31 31 31 31 5f 30 30 30 32 5f 6d 5f 30 30 |111111_0002_m_00|
+ * |00000020| 30 30 30 32 5f 30 05 0a 00 |0002_0... |
+ * +--------+-------------------------------------------------+----------------+
+ * |00000000| 62 62 62 62 62 |bbbbb |
+ * +--------+-------------------------------------------------+----------------+
+ * |00000000| 25 61 74 74 65 6d 70 74 5f 31 31 31 31 31 31 31 |%attempt_1111111|
+ * |00000010| 31 31 31 31 31 31 5f 30 30 30 33 5f 6d 5f 30 30 |111111_0003_m_00|
+ * |00000020| 30 30 30 33 5f 30 05 0a 00 |0003_0... |
+ * +--------+-------------------------------------------------+----------------+
+ * |00000000| 63 63 63 63 63 |ccccc |
+ * +--------+-------------------------------------------------+----------------+
+ * </pre>
+ */
+public class ShuffleChannelHandler extends SimpleChannelInboundHandler<FullHttpRequest> {
+ private final ShuffleChannelHandlerContext handlerCtx;
+
+ ShuffleChannelHandler(ShuffleChannelHandlerContext ctx) {
+ handlerCtx = ctx;
+ }
+
+ private List<String> splitMaps(List<String> mapq) {
+ if (null == mapq) {
+ return null;
+ }
+ final List<String> ret = new ArrayList<>();
+ for (String s : mapq) {
+ Collections.addAll(ret, s.split(","));
+ }
+ return ret;
+ }
+
+ @Override
+ public void channelActive(ChannelHandlerContext ctx)
+ throws Exception {
+ LOG.debug("Executing channelActive; channel='{}'", ctx.channel().id());
+ int numConnections = handlerCtx.activeConnections.incrementAndGet();
+ if ((handlerCtx.maxShuffleConnections > 0) &&
+ (numConnections > handlerCtx.maxShuffleConnections)) {
+ LOG.info(String.format("Current number of shuffle connections (%d) is " +
+ "greater than the max allowed shuffle connections (%d)",
+ handlerCtx.allChannels.size(), handlerCtx.maxShuffleConnections));
+
+ Map<String, String> headers = new HashMap<>(1);
+ // notify fetchers to backoff for a while before closing the connection
+ // if the shuffle connection limit is hit. Fetchers are expected to
+ // handle this notification gracefully, that is, not treating this as a
+ // fetch failure.
+ headers.put(RETRY_AFTER_HEADER, String.valueOf(FETCH_RETRY_DELAY));
+ sendError(ctx, "", TOO_MANY_REQ_STATUS, headers);
+ } else {
+ super.channelActive(ctx);
+ handlerCtx.allChannels.add(ctx.channel());
+ LOG.debug("Added channel: {}, channel id: {}. Accepted number of connections={}",
+ ctx.channel(), ctx.channel().id(), handlerCtx.activeConnections.get());
+ }
+ }
+
+ @Override
+ public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+ LOG.debug("Executing channelInactive; channel='{}'", ctx.channel().id());
+ super.channelInactive(ctx);
+ int noOfConnections = handlerCtx.activeConnections.decrementAndGet();
+ LOG.debug("New value of Accepted number of connections={}", noOfConnections);
+ }
+
+ @Override
+ public void channelRead0(ChannelHandlerContext ctx, FullHttpRequest request) {
+ Channel channel = ctx.channel();
+ LOG.debug("Received HTTP request: {}, channel='{}'", request, channel.id());
+
+ if (request.method() != GET) {
+ sendError(ctx, METHOD_NOT_ALLOWED);
+ return;
+ }
+ // Check whether the shuffle version is compatible
+ String shuffleVersion = ShuffleHeader.DEFAULT_HTTP_HEADER_VERSION;
+ String httpHeaderName = ShuffleHeader.DEFAULT_HTTP_HEADER_NAME;
+ if (request.headers() != null) {
+ shuffleVersion = request.headers().get(ShuffleHeader.HTTP_HEADER_VERSION);
+ httpHeaderName = request.headers().get(ShuffleHeader.HTTP_HEADER_NAME);
+ LOG.debug("Received from request header: ShuffleVersion={} header name={}, channel id: {}",
+ shuffleVersion, httpHeaderName, channel.id());
+ }
+ if (request.headers() == null ||
+ !ShuffleHeader.DEFAULT_HTTP_HEADER_NAME.equals(httpHeaderName) ||
+ !ShuffleHeader.DEFAULT_HTTP_HEADER_VERSION.equals(shuffleVersion)) {
+ sendError(ctx, "Incompatible shuffle request version", BAD_REQUEST);
+ return;
+ }
+ final Map<String, List<String>> q =
+ new QueryStringDecoder(request.uri()).parameters();
+
+ final List<String> keepAliveList = q.get("keepAlive");
+ boolean keepAliveParam = false;
+ if (keepAliveList != null && keepAliveList.size() == 1) {
+ keepAliveParam = Boolean.parseBoolean(keepAliveList.get(0));
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("KeepAliveParam: {} : {}, channel id: {}",
+ keepAliveList, keepAliveParam, channel.id());
+ }
+ }
+ final List<String> mapIds = splitMaps(q.get("map"));
+ final List<String> reduceQ = q.get("reduce");
+ final List<String> jobQ = q.get("job");
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("RECV: " + request.uri() +
+ "\n mapId: " + mapIds +
+ "\n reduceId: " + reduceQ +
+ "\n jobId: " + jobQ +
+ "\n keepAlive: " + keepAliveParam +
+ "\n channel id: " + channel.id());
+ }
+
+ if (mapIds == null || reduceQ == null || jobQ == null) {
+ sendError(ctx, "Required param job, map and reduce", BAD_REQUEST);
+ return;
+ }
+ if (reduceQ.size() != 1 || jobQ.size() != 1) {
+ sendError(ctx, "Too many job/reduce parameters", BAD_REQUEST);
+ return;
+ }
+
+ int reduceId;
+ String jobId;
+ try {
+ reduceId = Integer.parseInt(reduceQ.get(0));
+ jobId = jobQ.get(0);
+ } catch (NumberFormatException e) {
+ sendError(ctx, "Bad reduce parameter", BAD_REQUEST);
+ return;
+ } catch (IllegalArgumentException e) {
+ sendError(ctx, "Bad job parameter", BAD_REQUEST);
+ return;
+ }
+ final String reqUri = request.uri();
+ if (null == reqUri) {
+ // TODO? add upstream?
+ sendError(ctx, FORBIDDEN);
+ return;
+ }
+ HttpResponse response = new DefaultHttpResponse(HTTP_1_1, OK);
+ try {
+ verifyRequest(jobId, ctx, request, response,
+ new URL("http", "", handlerCtx.port, reqUri));
+ } catch (IOException e) {
+ LOG.warn("Shuffle failure ", e);
+ sendError(ctx, e.getMessage(), UNAUTHORIZED);
+ return;
+ }
+
+ Map<String, MapOutputInfo> mapOutputInfoMap = new HashMap<>();
+ ChannelPipeline pipeline = channel.pipeline();
+ ShuffleHandler.TimeoutHandler timeoutHandler =
+ (ShuffleHandler.TimeoutHandler)pipeline.get(TIMEOUT_HANDLER);
+ timeoutHandler.setEnabledTimeout(false);
+ String user = handlerCtx.userRsrc.get(jobId);
+
+ try {
+ populateHeaders(mapIds, jobId, user, reduceId,
+ response, keepAliveParam, mapOutputInfoMap);
+ } catch(IOException e) {
+ LOG.error("Shuffle error while populating headers. Channel id: " + channel.id(), e);
+ sendError(ctx, getErrorMessage(e), INTERNAL_SERVER_ERROR);
+ return;
+ }
+
+ channel.write(response);
+
+ //Initialize one ReduceContext object per channelRead call
+ boolean keepAlive = keepAliveParam || handlerCtx.connectionKeepAliveEnabled;
+ ReduceContext reduceContext = new ReduceContext(mapIds, reduceId, ctx,
+ user, mapOutputInfoMap, jobId, keepAlive);
+
+ sendMap(reduceContext);
+ }
+
+ /**
+ * Calls sendMapOutput for the mapId pointed by ReduceContext.mapsToSend
+ * and increments it. This method is first called by messageReceived()
+ * maxSessionOpenFiles times and then on the completion of every
+ * sendMapOutput operation. This limits the number of open files on a node,
+ * which can get really large(exhausting file descriptors on the NM) if all
+ * sendMapOutputs are called in one go, as was done previous to this change.
+ * @param reduceContext used to call sendMapOutput with correct params.
+ */
+ public void sendMap(ReduceContext reduceContext) {
+ LOG.trace("Executing sendMap; channel='{}'", reduceContext.ctx.channel().id());
+ if (reduceContext.getMapsToSend().get() <
+ reduceContext.getMapIds().size()) {
+ int nextIndex = reduceContext.getMapsToSend().getAndIncrement();
+ String mapId = reduceContext.getMapIds().get(nextIndex);
+
+ try {
+ MapOutputInfo info = reduceContext.getInfoMap().get(mapId);
+ if (info == null) {
+ info = getMapOutputInfo(mapId, reduceContext.getReduceId(),
+ reduceContext.getJobId(), reduceContext.getUser());
+ }
+ LOG.trace("Calling sendMapOutput; channel='{}'", reduceContext.ctx.channel().id());
+ ChannelFuture nextMap = sendMapOutput(
+ reduceContext.getCtx().channel(),
+ reduceContext.getUser(), mapId,
+ reduceContext.getReduceId(), info);
+ nextMap.addListener(new ReduceMapFileCount(this, reduceContext));
+ } catch (IOException e) {
+ LOG.error("Shuffle error: {}; channel={}", e, reduceContext.ctx.channel().id());
+
+ // It is not possible to sendError, the success HttpResponse has been already sent
+ reduceContext.ctx.channel().close();
+ }
+ }
+ }
+
+ private String getErrorMessage(Throwable t) {
+ StringBuilder sb = new StringBuilder(t.getMessage());
+ while (t.getCause() != null) {
+ sb.append(t.getCause().getMessage());
+ t = t.getCause();
+ }
+ return sb.toString();
+ }
+
+ protected MapOutputInfo getMapOutputInfo(String mapId, int reduce, String jobId, String user)
+ throws IOException {
+ ShuffleHandler.AttemptPathInfo pathInfo;
+ try {
+ ShuffleHandler.AttemptPathIdentifier identifier = new ShuffleHandler.AttemptPathIdentifier(
+ jobId, user, mapId);
+ pathInfo = handlerCtx.pathCache.get(identifier);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Retrieved pathInfo for " + identifier +
+ " check for corresponding loaded messages to determine whether" +
+ " it was loaded or cached");
+ }
+ } catch (ExecutionException e) {
+ if (e.getCause() instanceof IOException) {
+ throw (IOException) e.getCause();
+ } else {
+ throw new RuntimeException(e.getCause());
+ }
+ }
+
+ IndexRecord info =
+ handlerCtx.indexCache.getIndexInformation(mapId, reduce, pathInfo.indexPath, user);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("getMapOutputInfo: jobId=" + jobId + ", mapId=" + mapId +
+ ",dataFile=" + pathInfo.dataPath + ", indexFile=" +
+ pathInfo.indexPath);
+ LOG.debug("getMapOutputInfo: startOffset={}, partLength={} rawLength={}",
+ info.startOffset, info.partLength, info.rawLength);
+ }
+
+ return new MapOutputInfo(pathInfo.dataPath, info);
+ }
+
+ protected void populateHeaders(List<String> mapIds, String jobId,
+ String user, int reduce, HttpResponse response,
+ boolean keepAliveParam,
+ Map<String, MapOutputInfo> mapOutputInfoMap)
+ throws IOException {
+
+ long contentLength = 0;
+ for (String mapId : mapIds) {
+ MapOutputInfo outputInfo = getMapOutputInfo(mapId, reduce, jobId, user);
+ if (mapOutputInfoMap.size() < handlerCtx.mapOutputMetaInfoCacheSize) {
+ mapOutputInfoMap.put(mapId, outputInfo);
+ }
+
+ ShuffleHeader header =
+ new ShuffleHeader(mapId, outputInfo.indexRecord.partLength,
+ outputInfo.indexRecord.rawLength, reduce);
+ DataOutputBuffer dob = new DataOutputBuffer();
+ header.write(dob);
+ contentLength += outputInfo.indexRecord.partLength;
+ contentLength += dob.getLength();
+
+ // verify file access to data file to send an actually correct http error
+ final File spillFile = new File(outputInfo.mapOutputFileName.toString());
+ RandomAccessFile r = SecureIOUtils.openForRandomRead(spillFile, "r", user, null);
+ r.close();
+ }
+
+ // Now set the response headers.
+ setResponseHeaders(response, keepAliveParam, contentLength);
+
+ // this audit log is disabled by default,
+ // to turn it on please enable this audit log
+ // on log4j.properties by uncommenting the setting
+ if (AUDITLOG.isDebugEnabled()) {
+ StringBuilder sb = new StringBuilder("shuffle for ");
+ sb.append(jobId).append(" reducer ").append(reduce);
+ sb.append(" length ").append(contentLength);
+ if (AUDITLOG.isTraceEnabled()) {
+ // For trace level logging, append the list of mappers
+ sb.append(" mappers: ").append(mapIds);
+ AUDITLOG.trace(sb.toString());
+ } else {
+ AUDITLOG.debug(sb.toString());
+ }
+ }
+ }
+
+ protected void setResponseHeaders(HttpResponse response,
+ boolean keepAliveParam, long contentLength) {
+ if (!handlerCtx.connectionKeepAliveEnabled && !keepAliveParam) {
+ response.headers().set(HttpHeader.CONNECTION.asString(), CONNECTION_CLOSE);
+ } else {
+ response.headers().set(HttpHeader.CONNECTION.asString(),
+ HttpHeader.KEEP_ALIVE.asString());
+ response.headers().set(HttpHeader.KEEP_ALIVE.asString(),
+ "timeout=" + handlerCtx.connectionKeepAliveTimeOut);
+ }
+
+ // Content length must be set (https://www.rfc-editor.org/rfc/rfc7230#section-3.3.3)
+ HttpUtil.setContentLength(response, contentLength);
+ }
+
+ @SuppressWarnings("checkstyle:VisibilityModifier")
+ static class MapOutputInfo {
+ final Path mapOutputFileName;
+ final IndexRecord indexRecord;
+
+ MapOutputInfo(Path mapOutputFileName, IndexRecord indexRecord) {
+ this.mapOutputFileName = mapOutputFileName;
+ this.indexRecord = indexRecord;
+ }
+ }
+
+ protected void verifyRequest(String appid, ChannelHandlerContext ctx,
+ HttpRequest request, HttpResponse response, URL requestUri)
+ throws IOException {
+ SecretKey tokenSecret = handlerCtx.secretManager.retrieveTokenSecret(appid);
+ if (null == tokenSecret) {
+ LOG.info("Request for unknown token {}, channel id: {}", appid, ctx.channel().id());
+ throw new IOException("Could not find jobid");
+ }
+ // encrypting URL
+ String encryptedURL = SecureShuffleUtils.buildMsgFrom(requestUri);
+ // hash from the fetcher
+ String urlHashStr =
+ request.headers().get(SecureShuffleUtils.HTTP_HEADER_URL_HASH);
+ if (urlHashStr == null) {
+ LOG.info("Missing header hash for {}, channel id: {}", appid, ctx.channel().id());
+ throw new IOException("fetcher cannot be authenticated");
+ }
+ if (LOG.isDebugEnabled()) {
+ int len = urlHashStr.length();
+ LOG.debug("Verifying request. encryptedURL:{}, hash:{}, channel id: " +
+ "{}", encryptedURL,
+ urlHashStr.substring(len - len / 2, len - 1), ctx.channel().id());
+ }
+ // verify - throws exception
+ SecureShuffleUtils.verifyReply(urlHashStr, encryptedURL, tokenSecret);
+ // verification passed - encode the reply
+ String reply = SecureShuffleUtils.generateHash(urlHashStr.getBytes(Charsets.UTF_8),
+ tokenSecret);
+ response.headers().set(
+ SecureShuffleUtils.HTTP_HEADER_REPLY_URL_HASH, reply);
+ // Put shuffle version into http header
+ response.headers().set(ShuffleHeader.HTTP_HEADER_NAME,
+ ShuffleHeader.DEFAULT_HTTP_HEADER_NAME);
+ response.headers().set(ShuffleHeader.HTTP_HEADER_VERSION,
+ ShuffleHeader.DEFAULT_HTTP_HEADER_VERSION);
+ if (LOG.isDebugEnabled()) {
+ int len = reply.length();
+ LOG.debug("Fetcher request verified. " +
+ "encryptedURL: {}, reply: {}, channel id: {}",
+ encryptedURL, reply.substring(len - len / 2, len - 1),
+ ctx.channel().id());
+ }
+ }
+
+ public static ByteBuf shuffleHeaderToBytes(ShuffleHeader header) throws IOException {
+ final DataOutputBuffer dob = new DataOutputBuffer();
+ header.write(dob);
+ return wrappedBuffer(dob.getData(), 0, dob.getLength());
+ }
+
+ protected ChannelFuture sendMapOutput(Channel ch, String user, String mapId, int reduce,
+ MapOutputInfo mapOutputInfo)
+ throws IOException {
+ final IndexRecord info = mapOutputInfo.indexRecord;
+ ch.write(shuffleHeaderToBytes(
+ new ShuffleHeader(mapId, info.partLength, info.rawLength, reduce)));
+ final File spillFile =
+ new File(mapOutputInfo.mapOutputFileName.toString());
+ RandomAccessFile spill = SecureIOUtils.openForRandomRead(spillFile, "r", user, null);
Review Comment:
The FadvisedFileRegion/FadvisedChunkedFile closes the file.
(The openForRandomRead() could be moved to each Fadvised() constructor to make it a bit harder to make a mistake with it, but the current code can't throw and exception between openForRandomRead and giving it to Fadvised constructors.)
##########
hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-shuffle/src/test/resources/log4j.properties:
##########
@@ -17,5 +17,5 @@ log4j.threshold=ALL
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d{ISO8601} %-5p [%t] %c{2} (%F:%M(%L)) - %m%n
-log4j.logger.io.netty=INFO
-log4j.logger.org.apache.hadoop.mapred=INFO
\ No newline at end of file
+log4j.logger.io.netty=TRACE
Review Comment:
It doesn't log that much at all.
##########
hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-shuffle/src/main/java/org/apache/hadoop/mapred/ShuffleChannelHandler.java:
##########
@@ -0,0 +1,715 @@
+/*
+ * 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.hadoop.mapred;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.handler.codec.TooLongFrameException;
+import io.netty.handler.codec.http.DefaultFullHttpResponse;
+import io.netty.handler.codec.http.DefaultHttpResponse;
+import io.netty.handler.codec.http.FullHttpRequest;
+import io.netty.handler.codec.http.FullHttpResponse;
+import io.netty.handler.codec.http.HttpRequest;
+import io.netty.handler.codec.http.HttpResponse;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.netty.handler.codec.http.HttpUtil;
+import io.netty.handler.codec.http.LastHttpContent;
+import io.netty.handler.codec.http.QueryStringDecoder;
+import io.netty.handler.ssl.SslHandler;
+import io.netty.util.CharsetUtil;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.RandomAccessFile;
+import java.net.URL;
+import java.nio.channels.ClosedChannelException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import javax.crypto.SecretKey;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.DataOutputBuffer;
+import org.apache.hadoop.io.SecureIOUtils;
+import org.apache.hadoop.mapreduce.security.SecureShuffleUtils;
+import org.apache.hadoop.mapreduce.task.reduce.ShuffleHeader;
+import org.apache.hadoop.thirdparty.com.google.common.base.Charsets;
+import org.eclipse.jetty.http.HttpHeader;
+
+import static io.netty.buffer.Unpooled.wrappedBuffer;
+import static io.netty.handler.codec.http.HttpHeaderNames.CONTENT_TYPE;
+import static io.netty.handler.codec.http.HttpMethod.GET;
+import static io.netty.handler.codec.http.HttpResponseStatus.BAD_REQUEST;
+import static io.netty.handler.codec.http.HttpResponseStatus.FORBIDDEN;
+import static io.netty.handler.codec.http.HttpResponseStatus.INTERNAL_SERVER_ERROR;
+import static io.netty.handler.codec.http.HttpResponseStatus.METHOD_NOT_ALLOWED;
+import static io.netty.handler.codec.http.HttpResponseStatus.OK;
+import static io.netty.handler.codec.http.HttpResponseStatus.UNAUTHORIZED;
+import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1;
+import static org.apache.hadoop.mapred.ShuffleHandler.AUDITLOG;
+import static org.apache.hadoop.mapred.ShuffleHandler.CONNECTION_CLOSE;
+import static org.apache.hadoop.mapred.ShuffleHandler.FETCH_RETRY_DELAY;
+import static org.apache.hadoop.mapred.ShuffleHandler.IGNORABLE_ERROR_MESSAGE;
+import static org.apache.hadoop.mapred.ShuffleHandler.RETRY_AFTER_HEADER;
+import static org.apache.hadoop.mapred.ShuffleHandler.TIMEOUT_HANDLER;
+import static org.apache.hadoop.mapred.ShuffleHandler.TOO_MANY_REQ_STATUS;
+import static org.apache.hadoop.mapred.ShuffleHandler.LOG;
+
+/**
+ * ShuffleChannelHandler verifies the map request then servers the attempts in a http stream.
+ * Before each attempt a serialised ShuffleHeader object is written with the details.
+ *
+ * <pre>
+ * Example Request
+ * ===================
+ * GET /mapOutput?job=job_1111111111111_0001&reduce=0&
+ * map=attempt_1111111111111_0001_m_000001_0,
+ * attempt_1111111111111_0002_m_000002_0,
+ * attempt_1111111111111_0003_m_000003_0 HTTP/1.1
+ * name: mapreduce
+ * version: 1.0.0
+ * UrlHash: 9zS++qE0/7/D2l1Rg0TqRoSguAk=
+ *
+ * Example Response
+ * ===================
+ * HTTP/1.1 200 OK
+ * ReplyHash: GcuojWkAxXUyhZHPnwoV/MW2tGA=
+ * name: mapreduce
+ * version: 1.0.0
+ * connection: close
+ * content-length: 138
+ *
+ * +--------+-------------------------------------------------+----------------+
+ * |00000000| 25 61 74 74 65 6d 70 74 5f 31 31 31 31 31 31 31 |%attempt_1111111|
+ * |00000010| 31 31 31 31 31 31 5f 30 30 30 31 5f 6d 5f 30 30 |111111_0001_m_00|
+ * |00000020| 30 30 30 31 5f 30 05 0a 00 |0001_0... |
+ * +--------+-------------------------------------------------+----------------+
+ * |00000000| 61 61 61 61 61 |aaaaa |
+ * +--------+-------------------------------------------------+----------------+
+ * |00000000| 25 61 74 74 65 6d 70 74 5f 31 31 31 31 31 31 31 |%attempt_1111111|
+ * |00000010| 31 31 31 31 31 31 5f 30 30 30 32 5f 6d 5f 30 30 |111111_0002_m_00|
+ * |00000020| 30 30 30 32 5f 30 05 0a 00 |0002_0... |
+ * +--------+-------------------------------------------------+----------------+
+ * |00000000| 62 62 62 62 62 |bbbbb |
+ * +--------+-------------------------------------------------+----------------+
+ * |00000000| 25 61 74 74 65 6d 70 74 5f 31 31 31 31 31 31 31 |%attempt_1111111|
+ * |00000010| 31 31 31 31 31 31 5f 30 30 30 33 5f 6d 5f 30 30 |111111_0003_m_00|
+ * |00000020| 30 30 30 33 5f 30 05 0a 00 |0003_0... |
+ * +--------+-------------------------------------------------+----------------+
+ * |00000000| 63 63 63 63 63 |ccccc |
+ * +--------+-------------------------------------------------+----------------+
+ * </pre>
+ */
+public class ShuffleChannelHandler extends SimpleChannelInboundHandler<FullHttpRequest> {
+ private final ShuffleChannelHandlerContext handlerCtx;
+
+ ShuffleChannelHandler(ShuffleChannelHandlerContext ctx) {
+ handlerCtx = ctx;
+ }
+
+ private List<String> splitMaps(List<String> mapq) {
+ if (null == mapq) {
+ return null;
+ }
+ final List<String> ret = new ArrayList<>();
+ for (String s : mapq) {
+ Collections.addAll(ret, s.split(","));
+ }
+ return ret;
+ }
+
+ @Override
+ public void channelActive(ChannelHandlerContext ctx)
+ throws Exception {
+ LOG.debug("Executing channelActive; channel='{}'", ctx.channel().id());
+ int numConnections = handlerCtx.activeConnections.incrementAndGet();
+ if ((handlerCtx.maxShuffleConnections > 0) &&
+ (numConnections > handlerCtx.maxShuffleConnections)) {
+ LOG.info(String.format("Current number of shuffle connections (%d) is " +
+ "greater than the max allowed shuffle connections (%d)",
+ handlerCtx.allChannels.size(), handlerCtx.maxShuffleConnections));
+
+ Map<String, String> headers = new HashMap<>(1);
+ // notify fetchers to backoff for a while before closing the connection
+ // if the shuffle connection limit is hit. Fetchers are expected to
+ // handle this notification gracefully, that is, not treating this as a
+ // fetch failure.
+ headers.put(RETRY_AFTER_HEADER, String.valueOf(FETCH_RETRY_DELAY));
+ sendError(ctx, "", TOO_MANY_REQ_STATUS, headers);
+ } else {
+ super.channelActive(ctx);
+ handlerCtx.allChannels.add(ctx.channel());
+ LOG.debug("Added channel: {}, channel id: {}. Accepted number of connections={}",
+ ctx.channel(), ctx.channel().id(), handlerCtx.activeConnections.get());
+ }
+ }
+
+ @Override
+ public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+ LOG.debug("Executing channelInactive; channel='{}'", ctx.channel().id());
+ super.channelInactive(ctx);
+ int noOfConnections = handlerCtx.activeConnections.decrementAndGet();
+ LOG.debug("New value of Accepted number of connections={}", noOfConnections);
+ }
+
+ @Override
+ public void channelRead0(ChannelHandlerContext ctx, FullHttpRequest request) {
+ Channel channel = ctx.channel();
+ LOG.debug("Received HTTP request: {}, channel='{}'", request, channel.id());
+
+ if (request.method() != GET) {
+ sendError(ctx, METHOD_NOT_ALLOWED);
+ return;
+ }
+ // Check whether the shuffle version is compatible
+ String shuffleVersion = ShuffleHeader.DEFAULT_HTTP_HEADER_VERSION;
+ String httpHeaderName = ShuffleHeader.DEFAULT_HTTP_HEADER_NAME;
+ if (request.headers() != null) {
+ shuffleVersion = request.headers().get(ShuffleHeader.HTTP_HEADER_VERSION);
+ httpHeaderName = request.headers().get(ShuffleHeader.HTTP_HEADER_NAME);
+ LOG.debug("Received from request header: ShuffleVersion={} header name={}, channel id: {}",
+ shuffleVersion, httpHeaderName, channel.id());
+ }
+ if (request.headers() == null ||
+ !ShuffleHeader.DEFAULT_HTTP_HEADER_NAME.equals(httpHeaderName) ||
+ !ShuffleHeader.DEFAULT_HTTP_HEADER_VERSION.equals(shuffleVersion)) {
+ sendError(ctx, "Incompatible shuffle request version", BAD_REQUEST);
+ return;
+ }
+ final Map<String, List<String>> q =
+ new QueryStringDecoder(request.uri()).parameters();
+
+ final List<String> keepAliveList = q.get("keepAlive");
+ boolean keepAliveParam = false;
+ if (keepAliveList != null && keepAliveList.size() == 1) {
+ keepAliveParam = Boolean.parseBoolean(keepAliveList.get(0));
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("KeepAliveParam: {} : {}, channel id: {}",
+ keepAliveList, keepAliveParam, channel.id());
+ }
+ }
+ final List<String> mapIds = splitMaps(q.get("map"));
+ final List<String> reduceQ = q.get("reduce");
+ final List<String> jobQ = q.get("job");
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("RECV: " + request.uri() +
+ "\n mapId: " + mapIds +
+ "\n reduceId: " + reduceQ +
+ "\n jobId: " + jobQ +
+ "\n keepAlive: " + keepAliveParam +
+ "\n channel id: " + channel.id());
+ }
+
+ if (mapIds == null || reduceQ == null || jobQ == null) {
+ sendError(ctx, "Required param job, map and reduce", BAD_REQUEST);
Review Comment:
No change here from my side.
--
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: common-issues-unsubscribe@hadoop.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: common-issues-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-issues-help@hadoop.apache.org
[GitHub] [hadoop] K0K0V0K commented on a diff in pull request #5311: MAPREDUCE-7431. ShuffleHandler refactor and fix after Netty4 upgrade.
Posted by GitBox <gi...@apache.org>.
K0K0V0K commented on code in PR #5311:
URL: https://github.com/apache/hadoop/pull/5311#discussion_r1081143805
##########
hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-shuffle/src/main/java/org/apache/hadoop/mapred/ShuffleChannelHandler.java:
##########
@@ -0,0 +1,715 @@
+/*
+ * 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.hadoop.mapred;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.handler.codec.TooLongFrameException;
+import io.netty.handler.codec.http.DefaultFullHttpResponse;
+import io.netty.handler.codec.http.DefaultHttpResponse;
+import io.netty.handler.codec.http.FullHttpRequest;
+import io.netty.handler.codec.http.FullHttpResponse;
+import io.netty.handler.codec.http.HttpRequest;
+import io.netty.handler.codec.http.HttpResponse;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.netty.handler.codec.http.HttpUtil;
+import io.netty.handler.codec.http.LastHttpContent;
+import io.netty.handler.codec.http.QueryStringDecoder;
+import io.netty.handler.ssl.SslHandler;
+import io.netty.util.CharsetUtil;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.RandomAccessFile;
+import java.net.URL;
+import java.nio.channels.ClosedChannelException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import javax.crypto.SecretKey;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.DataOutputBuffer;
+import org.apache.hadoop.io.SecureIOUtils;
+import org.apache.hadoop.mapreduce.security.SecureShuffleUtils;
+import org.apache.hadoop.mapreduce.task.reduce.ShuffleHeader;
+import org.apache.hadoop.thirdparty.com.google.common.base.Charsets;
+import org.eclipse.jetty.http.HttpHeader;
+
+import static io.netty.buffer.Unpooled.wrappedBuffer;
+import static io.netty.handler.codec.http.HttpHeaderNames.CONTENT_TYPE;
+import static io.netty.handler.codec.http.HttpMethod.GET;
+import static io.netty.handler.codec.http.HttpResponseStatus.BAD_REQUEST;
+import static io.netty.handler.codec.http.HttpResponseStatus.FORBIDDEN;
+import static io.netty.handler.codec.http.HttpResponseStatus.INTERNAL_SERVER_ERROR;
+import static io.netty.handler.codec.http.HttpResponseStatus.METHOD_NOT_ALLOWED;
+import static io.netty.handler.codec.http.HttpResponseStatus.OK;
+import static io.netty.handler.codec.http.HttpResponseStatus.UNAUTHORIZED;
+import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1;
+import static org.apache.hadoop.mapred.ShuffleHandler.AUDITLOG;
+import static org.apache.hadoop.mapred.ShuffleHandler.CONNECTION_CLOSE;
+import static org.apache.hadoop.mapred.ShuffleHandler.FETCH_RETRY_DELAY;
+import static org.apache.hadoop.mapred.ShuffleHandler.IGNORABLE_ERROR_MESSAGE;
+import static org.apache.hadoop.mapred.ShuffleHandler.RETRY_AFTER_HEADER;
+import static org.apache.hadoop.mapred.ShuffleHandler.TIMEOUT_HANDLER;
+import static org.apache.hadoop.mapred.ShuffleHandler.TOO_MANY_REQ_STATUS;
+import static org.apache.hadoop.mapred.ShuffleHandler.LOG;
+
+/**
+ * ShuffleChannelHandler verifies the map request then servers the attempts in a http stream.
+ * Before each attempt a serialised ShuffleHeader object is written with the details.
+ *
+ * <pre>
+ * Example Request
+ * ===================
+ * GET /mapOutput?job=job_1111111111111_0001&reduce=0&
+ * map=attempt_1111111111111_0001_m_000001_0,
+ * attempt_1111111111111_0002_m_000002_0,
+ * attempt_1111111111111_0003_m_000003_0 HTTP/1.1
+ * name: mapreduce
+ * version: 1.0.0
+ * UrlHash: 9zS++qE0/7/D2l1Rg0TqRoSguAk=
+ *
+ * Example Response
+ * ===================
+ * HTTP/1.1 200 OK
+ * ReplyHash: GcuojWkAxXUyhZHPnwoV/MW2tGA=
+ * name: mapreduce
+ * version: 1.0.0
+ * connection: close
+ * content-length: 138
+ *
+ * +--------+-------------------------------------------------+----------------+
+ * |00000000| 25 61 74 74 65 6d 70 74 5f 31 31 31 31 31 31 31 |%attempt_1111111|
+ * |00000010| 31 31 31 31 31 31 5f 30 30 30 31 5f 6d 5f 30 30 |111111_0001_m_00|
+ * |00000020| 30 30 30 31 5f 30 05 0a 00 |0001_0... |
+ * +--------+-------------------------------------------------+----------------+
+ * |00000000| 61 61 61 61 61 |aaaaa |
+ * +--------+-------------------------------------------------+----------------+
+ * |00000000| 25 61 74 74 65 6d 70 74 5f 31 31 31 31 31 31 31 |%attempt_1111111|
+ * |00000010| 31 31 31 31 31 31 5f 30 30 30 32 5f 6d 5f 30 30 |111111_0002_m_00|
+ * |00000020| 30 30 30 32 5f 30 05 0a 00 |0002_0... |
+ * +--------+-------------------------------------------------+----------------+
+ * |00000000| 62 62 62 62 62 |bbbbb |
+ * +--------+-------------------------------------------------+----------------+
+ * |00000000| 25 61 74 74 65 6d 70 74 5f 31 31 31 31 31 31 31 |%attempt_1111111|
+ * |00000010| 31 31 31 31 31 31 5f 30 30 30 33 5f 6d 5f 30 30 |111111_0003_m_00|
+ * |00000020| 30 30 30 33 5f 30 05 0a 00 |0003_0... |
+ * +--------+-------------------------------------------------+----------------+
+ * |00000000| 63 63 63 63 63 |ccccc |
+ * +--------+-------------------------------------------------+----------------+
+ * </pre>
+ */
+public class ShuffleChannelHandler extends SimpleChannelInboundHandler<FullHttpRequest> {
+ private final ShuffleChannelHandlerContext handlerCtx;
+
+ ShuffleChannelHandler(ShuffleChannelHandlerContext ctx) {
+ handlerCtx = ctx;
+ }
+
+ private List<String> splitMaps(List<String> mapq) {
+ if (null == mapq) {
+ return null;
+ }
+ final List<String> ret = new ArrayList<>();
+ for (String s : mapq) {
+ Collections.addAll(ret, s.split(","));
+ }
+ return ret;
+ }
+
+ @Override
+ public void channelActive(ChannelHandlerContext ctx)
+ throws Exception {
+ LOG.debug("Executing channelActive; channel='{}'", ctx.channel().id());
+ int numConnections = handlerCtx.activeConnections.incrementAndGet();
+ if ((handlerCtx.maxShuffleConnections > 0) &&
+ (numConnections > handlerCtx.maxShuffleConnections)) {
+ LOG.info(String.format("Current number of shuffle connections (%d) is " +
+ "greater than the max allowed shuffle connections (%d)",
+ handlerCtx.allChannels.size(), handlerCtx.maxShuffleConnections));
+
+ Map<String, String> headers = new HashMap<>(1);
+ // notify fetchers to backoff for a while before closing the connection
+ // if the shuffle connection limit is hit. Fetchers are expected to
+ // handle this notification gracefully, that is, not treating this as a
+ // fetch failure.
+ headers.put(RETRY_AFTER_HEADER, String.valueOf(FETCH_RETRY_DELAY));
+ sendError(ctx, "", TOO_MANY_REQ_STATUS, headers);
+ } else {
+ super.channelActive(ctx);
+ handlerCtx.allChannels.add(ctx.channel());
+ LOG.debug("Added channel: {}, channel id: {}. Accepted number of connections={}",
+ ctx.channel(), ctx.channel().id(), handlerCtx.activeConnections.get());
+ }
+ }
+
+ @Override
+ public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+ LOG.debug("Executing channelInactive; channel='{}'", ctx.channel().id());
+ super.channelInactive(ctx);
+ int noOfConnections = handlerCtx.activeConnections.decrementAndGet();
+ LOG.debug("New value of Accepted number of connections={}", noOfConnections);
+ }
+
+ @Override
+ public void channelRead0(ChannelHandlerContext ctx, FullHttpRequest request) {
+ Channel channel = ctx.channel();
+ LOG.debug("Received HTTP request: {}, channel='{}'", request, channel.id());
+
+ if (request.method() != GET) {
+ sendError(ctx, METHOD_NOT_ALLOWED);
+ return;
+ }
+ // Check whether the shuffle version is compatible
+ String shuffleVersion = ShuffleHeader.DEFAULT_HTTP_HEADER_VERSION;
+ String httpHeaderName = ShuffleHeader.DEFAULT_HTTP_HEADER_NAME;
+ if (request.headers() != null) {
+ shuffleVersion = request.headers().get(ShuffleHeader.HTTP_HEADER_VERSION);
+ httpHeaderName = request.headers().get(ShuffleHeader.HTTP_HEADER_NAME);
+ LOG.debug("Received from request header: ShuffleVersion={} header name={}, channel id: {}",
+ shuffleVersion, httpHeaderName, channel.id());
+ }
+ if (request.headers() == null ||
+ !ShuffleHeader.DEFAULT_HTTP_HEADER_NAME.equals(httpHeaderName) ||
+ !ShuffleHeader.DEFAULT_HTTP_HEADER_VERSION.equals(shuffleVersion)) {
+ sendError(ctx, "Incompatible shuffle request version", BAD_REQUEST);
+ return;
+ }
+ final Map<String, List<String>> q =
+ new QueryStringDecoder(request.uri()).parameters();
+
+ final List<String> keepAliveList = q.get("keepAlive");
+ boolean keepAliveParam = false;
+ if (keepAliveList != null && keepAliveList.size() == 1) {
+ keepAliveParam = Boolean.parseBoolean(keepAliveList.get(0));
+ if (LOG.isDebugEnabled()) {
Review Comment:
Why this if is required here?
##########
hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-shuffle/src/test/resources/log4j.properties:
##########
@@ -17,5 +17,5 @@ log4j.threshold=ALL
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d{ISO8601} %-5p [%t] %c{2} (%F:%M(%L)) - %m%n
-log4j.logger.io.netty=INFO
-log4j.logger.org.apache.hadoop.mapred=INFO
\ No newline at end of file
+log4j.logger.io.netty=TRACE
Review Comment:
I think this can stay as INFO, to reduce the build test times
##########
hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-shuffle/src/main/java/org/apache/hadoop/mapred/ShuffleChannelHandler.java:
##########
@@ -0,0 +1,715 @@
+/*
+ * 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.hadoop.mapred;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.handler.codec.TooLongFrameException;
+import io.netty.handler.codec.http.DefaultFullHttpResponse;
+import io.netty.handler.codec.http.DefaultHttpResponse;
+import io.netty.handler.codec.http.FullHttpRequest;
+import io.netty.handler.codec.http.FullHttpResponse;
+import io.netty.handler.codec.http.HttpRequest;
+import io.netty.handler.codec.http.HttpResponse;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.netty.handler.codec.http.HttpUtil;
+import io.netty.handler.codec.http.LastHttpContent;
+import io.netty.handler.codec.http.QueryStringDecoder;
+import io.netty.handler.ssl.SslHandler;
+import io.netty.util.CharsetUtil;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.RandomAccessFile;
+import java.net.URL;
+import java.nio.channels.ClosedChannelException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import javax.crypto.SecretKey;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.DataOutputBuffer;
+import org.apache.hadoop.io.SecureIOUtils;
+import org.apache.hadoop.mapreduce.security.SecureShuffleUtils;
+import org.apache.hadoop.mapreduce.task.reduce.ShuffleHeader;
+import org.apache.hadoop.thirdparty.com.google.common.base.Charsets;
+import org.eclipse.jetty.http.HttpHeader;
+
+import static io.netty.buffer.Unpooled.wrappedBuffer;
+import static io.netty.handler.codec.http.HttpHeaderNames.CONTENT_TYPE;
+import static io.netty.handler.codec.http.HttpMethod.GET;
+import static io.netty.handler.codec.http.HttpResponseStatus.BAD_REQUEST;
+import static io.netty.handler.codec.http.HttpResponseStatus.FORBIDDEN;
+import static io.netty.handler.codec.http.HttpResponseStatus.INTERNAL_SERVER_ERROR;
+import static io.netty.handler.codec.http.HttpResponseStatus.METHOD_NOT_ALLOWED;
+import static io.netty.handler.codec.http.HttpResponseStatus.OK;
+import static io.netty.handler.codec.http.HttpResponseStatus.UNAUTHORIZED;
+import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1;
+import static org.apache.hadoop.mapred.ShuffleHandler.AUDITLOG;
+import static org.apache.hadoop.mapred.ShuffleHandler.CONNECTION_CLOSE;
+import static org.apache.hadoop.mapred.ShuffleHandler.FETCH_RETRY_DELAY;
+import static org.apache.hadoop.mapred.ShuffleHandler.IGNORABLE_ERROR_MESSAGE;
+import static org.apache.hadoop.mapred.ShuffleHandler.RETRY_AFTER_HEADER;
+import static org.apache.hadoop.mapred.ShuffleHandler.TIMEOUT_HANDLER;
+import static org.apache.hadoop.mapred.ShuffleHandler.TOO_MANY_REQ_STATUS;
+import static org.apache.hadoop.mapred.ShuffleHandler.LOG;
+
+/**
+ * ShuffleChannelHandler verifies the map request then servers the attempts in a http stream.
+ * Before each attempt a serialised ShuffleHeader object is written with the details.
+ *
+ * <pre>
+ * Example Request
+ * ===================
+ * GET /mapOutput?job=job_1111111111111_0001&reduce=0&
+ * map=attempt_1111111111111_0001_m_000001_0,
+ * attempt_1111111111111_0002_m_000002_0,
+ * attempt_1111111111111_0003_m_000003_0 HTTP/1.1
+ * name: mapreduce
+ * version: 1.0.0
+ * UrlHash: 9zS++qE0/7/D2l1Rg0TqRoSguAk=
+ *
+ * Example Response
+ * ===================
+ * HTTP/1.1 200 OK
+ * ReplyHash: GcuojWkAxXUyhZHPnwoV/MW2tGA=
+ * name: mapreduce
+ * version: 1.0.0
+ * connection: close
+ * content-length: 138
+ *
+ * +--------+-------------------------------------------------+----------------+
+ * |00000000| 25 61 74 74 65 6d 70 74 5f 31 31 31 31 31 31 31 |%attempt_1111111|
+ * |00000010| 31 31 31 31 31 31 5f 30 30 30 31 5f 6d 5f 30 30 |111111_0001_m_00|
+ * |00000020| 30 30 30 31 5f 30 05 0a 00 |0001_0... |
+ * +--------+-------------------------------------------------+----------------+
+ * |00000000| 61 61 61 61 61 |aaaaa |
+ * +--------+-------------------------------------------------+----------------+
+ * |00000000| 25 61 74 74 65 6d 70 74 5f 31 31 31 31 31 31 31 |%attempt_1111111|
+ * |00000010| 31 31 31 31 31 31 5f 30 30 30 32 5f 6d 5f 30 30 |111111_0002_m_00|
+ * |00000020| 30 30 30 32 5f 30 05 0a 00 |0002_0... |
+ * +--------+-------------------------------------------------+----------------+
+ * |00000000| 62 62 62 62 62 |bbbbb |
+ * +--------+-------------------------------------------------+----------------+
+ * |00000000| 25 61 74 74 65 6d 70 74 5f 31 31 31 31 31 31 31 |%attempt_1111111|
+ * |00000010| 31 31 31 31 31 31 5f 30 30 30 33 5f 6d 5f 30 30 |111111_0003_m_00|
+ * |00000020| 30 30 30 33 5f 30 05 0a 00 |0003_0... |
+ * +--------+-------------------------------------------------+----------------+
+ * |00000000| 63 63 63 63 63 |ccccc |
+ * +--------+-------------------------------------------------+----------------+
+ * </pre>
+ */
+public class ShuffleChannelHandler extends SimpleChannelInboundHandler<FullHttpRequest> {
+ private final ShuffleChannelHandlerContext handlerCtx;
+
+ ShuffleChannelHandler(ShuffleChannelHandlerContext ctx) {
+ handlerCtx = ctx;
+ }
+
+ private List<String> splitMaps(List<String> mapq) {
+ if (null == mapq) {
+ return null;
+ }
+ final List<String> ret = new ArrayList<>();
+ for (String s : mapq) {
+ Collections.addAll(ret, s.split(","));
+ }
+ return ret;
+ }
+
+ @Override
+ public void channelActive(ChannelHandlerContext ctx)
+ throws Exception {
+ LOG.debug("Executing channelActive; channel='{}'", ctx.channel().id());
+ int numConnections = handlerCtx.activeConnections.incrementAndGet();
+ if ((handlerCtx.maxShuffleConnections > 0) &&
+ (numConnections > handlerCtx.maxShuffleConnections)) {
+ LOG.info(String.format("Current number of shuffle connections (%d) is " +
+ "greater than the max allowed shuffle connections (%d)",
+ handlerCtx.allChannels.size(), handlerCtx.maxShuffleConnections));
+
+ Map<String, String> headers = new HashMap<>(1);
+ // notify fetchers to backoff for a while before closing the connection
+ // if the shuffle connection limit is hit. Fetchers are expected to
+ // handle this notification gracefully, that is, not treating this as a
+ // fetch failure.
+ headers.put(RETRY_AFTER_HEADER, String.valueOf(FETCH_RETRY_DELAY));
+ sendError(ctx, "", TOO_MANY_REQ_STATUS, headers);
+ } else {
+ super.channelActive(ctx);
+ handlerCtx.allChannels.add(ctx.channel());
+ LOG.debug("Added channel: {}, channel id: {}. Accepted number of connections={}",
+ ctx.channel(), ctx.channel().id(), handlerCtx.activeConnections.get());
+ }
+ }
+
+ @Override
+ public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+ LOG.debug("Executing channelInactive; channel='{}'", ctx.channel().id());
+ super.channelInactive(ctx);
+ int noOfConnections = handlerCtx.activeConnections.decrementAndGet();
+ LOG.debug("New value of Accepted number of connections={}", noOfConnections);
+ }
+
+ @Override
+ public void channelRead0(ChannelHandlerContext ctx, FullHttpRequest request) {
+ Channel channel = ctx.channel();
+ LOG.debug("Received HTTP request: {}, channel='{}'", request, channel.id());
+
+ if (request.method() != GET) {
+ sendError(ctx, METHOD_NOT_ALLOWED);
+ return;
+ }
+ // Check whether the shuffle version is compatible
+ String shuffleVersion = ShuffleHeader.DEFAULT_HTTP_HEADER_VERSION;
+ String httpHeaderName = ShuffleHeader.DEFAULT_HTTP_HEADER_NAME;
+ if (request.headers() != null) {
+ shuffleVersion = request.headers().get(ShuffleHeader.HTTP_HEADER_VERSION);
+ httpHeaderName = request.headers().get(ShuffleHeader.HTTP_HEADER_NAME);
+ LOG.debug("Received from request header: ShuffleVersion={} header name={}, channel id: {}",
+ shuffleVersion, httpHeaderName, channel.id());
+ }
+ if (request.headers() == null ||
+ !ShuffleHeader.DEFAULT_HTTP_HEADER_NAME.equals(httpHeaderName) ||
+ !ShuffleHeader.DEFAULT_HTTP_HEADER_VERSION.equals(shuffleVersion)) {
+ sendError(ctx, "Incompatible shuffle request version", BAD_REQUEST);
+ return;
+ }
+ final Map<String, List<String>> q =
+ new QueryStringDecoder(request.uri()).parameters();
+
+ final List<String> keepAliveList = q.get("keepAlive");
+ boolean keepAliveParam = false;
+ if (keepAliveList != null && keepAliveList.size() == 1) {
+ keepAliveParam = Boolean.parseBoolean(keepAliveList.get(0));
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("KeepAliveParam: {} : {}, channel id: {}",
+ keepAliveList, keepAliveParam, channel.id());
+ }
+ }
+ final List<String> mapIds = splitMaps(q.get("map"));
+ final List<String> reduceQ = q.get("reduce");
+ final List<String> jobQ = q.get("job");
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("RECV: " + request.uri() +
+ "\n mapId: " + mapIds +
+ "\n reduceId: " + reduceQ +
+ "\n jobId: " + jobQ +
+ "\n keepAlive: " + keepAliveParam +
+ "\n channel id: " + channel.id());
+ }
+
+ if (mapIds == null || reduceQ == null || jobQ == null) {
+ sendError(ctx, "Required param job, map and reduce", BAD_REQUEST);
Review Comment:
Maybe some error log can be added here what was null?
##########
hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-shuffle/src/main/java/org/apache/hadoop/mapred/ShuffleChannelHandler.java:
##########
@@ -0,0 +1,715 @@
+/*
+ * 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.hadoop.mapred;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.handler.codec.TooLongFrameException;
+import io.netty.handler.codec.http.DefaultFullHttpResponse;
+import io.netty.handler.codec.http.DefaultHttpResponse;
+import io.netty.handler.codec.http.FullHttpRequest;
+import io.netty.handler.codec.http.FullHttpResponse;
+import io.netty.handler.codec.http.HttpRequest;
+import io.netty.handler.codec.http.HttpResponse;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.netty.handler.codec.http.HttpUtil;
+import io.netty.handler.codec.http.LastHttpContent;
+import io.netty.handler.codec.http.QueryStringDecoder;
+import io.netty.handler.ssl.SslHandler;
+import io.netty.util.CharsetUtil;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.RandomAccessFile;
+import java.net.URL;
+import java.nio.channels.ClosedChannelException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import javax.crypto.SecretKey;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.DataOutputBuffer;
+import org.apache.hadoop.io.SecureIOUtils;
+import org.apache.hadoop.mapreduce.security.SecureShuffleUtils;
+import org.apache.hadoop.mapreduce.task.reduce.ShuffleHeader;
+import org.apache.hadoop.thirdparty.com.google.common.base.Charsets;
+import org.eclipse.jetty.http.HttpHeader;
+
+import static io.netty.buffer.Unpooled.wrappedBuffer;
+import static io.netty.handler.codec.http.HttpHeaderNames.CONTENT_TYPE;
+import static io.netty.handler.codec.http.HttpMethod.GET;
+import static io.netty.handler.codec.http.HttpResponseStatus.BAD_REQUEST;
+import static io.netty.handler.codec.http.HttpResponseStatus.FORBIDDEN;
+import static io.netty.handler.codec.http.HttpResponseStatus.INTERNAL_SERVER_ERROR;
+import static io.netty.handler.codec.http.HttpResponseStatus.METHOD_NOT_ALLOWED;
+import static io.netty.handler.codec.http.HttpResponseStatus.OK;
+import static io.netty.handler.codec.http.HttpResponseStatus.UNAUTHORIZED;
+import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1;
+import static org.apache.hadoop.mapred.ShuffleHandler.AUDITLOG;
+import static org.apache.hadoop.mapred.ShuffleHandler.CONNECTION_CLOSE;
+import static org.apache.hadoop.mapred.ShuffleHandler.FETCH_RETRY_DELAY;
+import static org.apache.hadoop.mapred.ShuffleHandler.IGNORABLE_ERROR_MESSAGE;
+import static org.apache.hadoop.mapred.ShuffleHandler.RETRY_AFTER_HEADER;
+import static org.apache.hadoop.mapred.ShuffleHandler.TIMEOUT_HANDLER;
+import static org.apache.hadoop.mapred.ShuffleHandler.TOO_MANY_REQ_STATUS;
+import static org.apache.hadoop.mapred.ShuffleHandler.LOG;
+
+/**
+ * ShuffleChannelHandler verifies the map request then servers the attempts in a http stream.
+ * Before each attempt a serialised ShuffleHeader object is written with the details.
+ *
+ * <pre>
+ * Example Request
+ * ===================
+ * GET /mapOutput?job=job_1111111111111_0001&reduce=0&
+ * map=attempt_1111111111111_0001_m_000001_0,
+ * attempt_1111111111111_0002_m_000002_0,
+ * attempt_1111111111111_0003_m_000003_0 HTTP/1.1
+ * name: mapreduce
+ * version: 1.0.0
+ * UrlHash: 9zS++qE0/7/D2l1Rg0TqRoSguAk=
+ *
+ * Example Response
+ * ===================
+ * HTTP/1.1 200 OK
+ * ReplyHash: GcuojWkAxXUyhZHPnwoV/MW2tGA=
+ * name: mapreduce
+ * version: 1.0.0
+ * connection: close
+ * content-length: 138
+ *
+ * +--------+-------------------------------------------------+----------------+
+ * |00000000| 25 61 74 74 65 6d 70 74 5f 31 31 31 31 31 31 31 |%attempt_1111111|
+ * |00000010| 31 31 31 31 31 31 5f 30 30 30 31 5f 6d 5f 30 30 |111111_0001_m_00|
+ * |00000020| 30 30 30 31 5f 30 05 0a 00 |0001_0... |
+ * +--------+-------------------------------------------------+----------------+
+ * |00000000| 61 61 61 61 61 |aaaaa |
+ * +--------+-------------------------------------------------+----------------+
+ * |00000000| 25 61 74 74 65 6d 70 74 5f 31 31 31 31 31 31 31 |%attempt_1111111|
+ * |00000010| 31 31 31 31 31 31 5f 30 30 30 32 5f 6d 5f 30 30 |111111_0002_m_00|
+ * |00000020| 30 30 30 32 5f 30 05 0a 00 |0002_0... |
+ * +--------+-------------------------------------------------+----------------+
+ * |00000000| 62 62 62 62 62 |bbbbb |
+ * +--------+-------------------------------------------------+----------------+
+ * |00000000| 25 61 74 74 65 6d 70 74 5f 31 31 31 31 31 31 31 |%attempt_1111111|
+ * |00000010| 31 31 31 31 31 31 5f 30 30 30 33 5f 6d 5f 30 30 |111111_0003_m_00|
+ * |00000020| 30 30 30 33 5f 30 05 0a 00 |0003_0... |
+ * +--------+-------------------------------------------------+----------------+
+ * |00000000| 63 63 63 63 63 |ccccc |
+ * +--------+-------------------------------------------------+----------------+
+ * </pre>
+ */
+public class ShuffleChannelHandler extends SimpleChannelInboundHandler<FullHttpRequest> {
+ private final ShuffleChannelHandlerContext handlerCtx;
+
+ ShuffleChannelHandler(ShuffleChannelHandlerContext ctx) {
+ handlerCtx = ctx;
+ }
+
+ private List<String> splitMaps(List<String> mapq) {
+ if (null == mapq) {
+ return null;
+ }
+ final List<String> ret = new ArrayList<>();
+ for (String s : mapq) {
+ Collections.addAll(ret, s.split(","));
+ }
+ return ret;
+ }
+
+ @Override
+ public void channelActive(ChannelHandlerContext ctx)
+ throws Exception {
+ LOG.debug("Executing channelActive; channel='{}'", ctx.channel().id());
+ int numConnections = handlerCtx.activeConnections.incrementAndGet();
+ if ((handlerCtx.maxShuffleConnections > 0) &&
+ (numConnections > handlerCtx.maxShuffleConnections)) {
+ LOG.info(String.format("Current number of shuffle connections (%d) is " +
+ "greater than the max allowed shuffle connections (%d)",
+ handlerCtx.allChannels.size(), handlerCtx.maxShuffleConnections));
+
+ Map<String, String> headers = new HashMap<>(1);
+ // notify fetchers to backoff for a while before closing the connection
+ // if the shuffle connection limit is hit. Fetchers are expected to
+ // handle this notification gracefully, that is, not treating this as a
+ // fetch failure.
+ headers.put(RETRY_AFTER_HEADER, String.valueOf(FETCH_RETRY_DELAY));
+ sendError(ctx, "", TOO_MANY_REQ_STATUS, headers);
+ } else {
+ super.channelActive(ctx);
+ handlerCtx.allChannels.add(ctx.channel());
+ LOG.debug("Added channel: {}, channel id: {}. Accepted number of connections={}",
+ ctx.channel(), ctx.channel().id(), handlerCtx.activeConnections.get());
+ }
+ }
+
+ @Override
+ public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+ LOG.debug("Executing channelInactive; channel='{}'", ctx.channel().id());
+ super.channelInactive(ctx);
+ int noOfConnections = handlerCtx.activeConnections.decrementAndGet();
+ LOG.debug("New value of Accepted number of connections={}", noOfConnections);
+ }
+
+ @Override
+ public void channelRead0(ChannelHandlerContext ctx, FullHttpRequest request) {
+ Channel channel = ctx.channel();
+ LOG.debug("Received HTTP request: {}, channel='{}'", request, channel.id());
+
+ if (request.method() != GET) {
+ sendError(ctx, METHOD_NOT_ALLOWED);
+ return;
+ }
+ // Check whether the shuffle version is compatible
+ String shuffleVersion = ShuffleHeader.DEFAULT_HTTP_HEADER_VERSION;
+ String httpHeaderName = ShuffleHeader.DEFAULT_HTTP_HEADER_NAME;
+ if (request.headers() != null) {
+ shuffleVersion = request.headers().get(ShuffleHeader.HTTP_HEADER_VERSION);
+ httpHeaderName = request.headers().get(ShuffleHeader.HTTP_HEADER_NAME);
+ LOG.debug("Received from request header: ShuffleVersion={} header name={}, channel id: {}",
+ shuffleVersion, httpHeaderName, channel.id());
+ }
+ if (request.headers() == null ||
+ !ShuffleHeader.DEFAULT_HTTP_HEADER_NAME.equals(httpHeaderName) ||
+ !ShuffleHeader.DEFAULT_HTTP_HEADER_VERSION.equals(shuffleVersion)) {
+ sendError(ctx, "Incompatible shuffle request version", BAD_REQUEST);
+ return;
+ }
+ final Map<String, List<String>> q =
+ new QueryStringDecoder(request.uri()).parameters();
+
+ final List<String> keepAliveList = q.get("keepAlive");
+ boolean keepAliveParam = false;
+ if (keepAliveList != null && keepAliveList.size() == 1) {
+ keepAliveParam = Boolean.parseBoolean(keepAliveList.get(0));
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("KeepAliveParam: {} : {}, channel id: {}",
+ keepAliveList, keepAliveParam, channel.id());
+ }
+ }
+ final List<String> mapIds = splitMaps(q.get("map"));
+ final List<String> reduceQ = q.get("reduce");
+ final List<String> jobQ = q.get("job");
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("RECV: " + request.uri() +
+ "\n mapId: " + mapIds +
+ "\n reduceId: " + reduceQ +
+ "\n jobId: " + jobQ +
+ "\n keepAlive: " + keepAliveParam +
+ "\n channel id: " + channel.id());
+ }
+
+ if (mapIds == null || reduceQ == null || jobQ == null) {
+ sendError(ctx, "Required param job, map and reduce", BAD_REQUEST);
+ return;
+ }
+ if (reduceQ.size() != 1 || jobQ.size() != 1) {
+ sendError(ctx, "Too many job/reduce parameters", BAD_REQUEST);
+ return;
+ }
+
+ int reduceId;
+ String jobId;
+ try {
+ reduceId = Integer.parseInt(reduceQ.get(0));
+ jobId = jobQ.get(0);
+ } catch (NumberFormatException e) {
+ sendError(ctx, "Bad reduce parameter", BAD_REQUEST);
+ return;
+ } catch (IllegalArgumentException e) {
+ sendError(ctx, "Bad job parameter", BAD_REQUEST);
+ return;
+ }
+ final String reqUri = request.uri();
+ if (null == reqUri) {
+ // TODO? add upstream?
+ sendError(ctx, FORBIDDEN);
+ return;
+ }
+ HttpResponse response = new DefaultHttpResponse(HTTP_1_1, OK);
+ try {
+ verifyRequest(jobId, ctx, request, response,
+ new URL("http", "", handlerCtx.port, reqUri));
+ } catch (IOException e) {
+ LOG.warn("Shuffle failure ", e);
+ sendError(ctx, e.getMessage(), UNAUTHORIZED);
+ return;
+ }
+
+ Map<String, MapOutputInfo> mapOutputInfoMap = new HashMap<>();
+ ChannelPipeline pipeline = channel.pipeline();
+ ShuffleHandler.TimeoutHandler timeoutHandler =
+ (ShuffleHandler.TimeoutHandler)pipeline.get(TIMEOUT_HANDLER);
+ timeoutHandler.setEnabledTimeout(false);
+ String user = handlerCtx.userRsrc.get(jobId);
+
+ try {
+ populateHeaders(mapIds, jobId, user, reduceId,
+ response, keepAliveParam, mapOutputInfoMap);
+ } catch(IOException e) {
+ LOG.error("Shuffle error while populating headers. Channel id: " + channel.id(), e);
+ sendError(ctx, getErrorMessage(e), INTERNAL_SERVER_ERROR);
+ return;
+ }
+
+ channel.write(response);
+
+ //Initialize one ReduceContext object per channelRead call
+ boolean keepAlive = keepAliveParam || handlerCtx.connectionKeepAliveEnabled;
+ ReduceContext reduceContext = new ReduceContext(mapIds, reduceId, ctx,
+ user, mapOutputInfoMap, jobId, keepAlive);
+
+ sendMap(reduceContext);
+ }
+
+ /**
+ * Calls sendMapOutput for the mapId pointed by ReduceContext.mapsToSend
+ * and increments it. This method is first called by messageReceived()
+ * maxSessionOpenFiles times and then on the completion of every
+ * sendMapOutput operation. This limits the number of open files on a node,
+ * which can get really large(exhausting file descriptors on the NM) if all
+ * sendMapOutputs are called in one go, as was done previous to this change.
+ * @param reduceContext used to call sendMapOutput with correct params.
+ */
+ public void sendMap(ReduceContext reduceContext) {
+ LOG.trace("Executing sendMap; channel='{}'", reduceContext.ctx.channel().id());
+ if (reduceContext.getMapsToSend().get() <
+ reduceContext.getMapIds().size()) {
+ int nextIndex = reduceContext.getMapsToSend().getAndIncrement();
+ String mapId = reduceContext.getMapIds().get(nextIndex);
+
+ try {
+ MapOutputInfo info = reduceContext.getInfoMap().get(mapId);
+ if (info == null) {
+ info = getMapOutputInfo(mapId, reduceContext.getReduceId(),
+ reduceContext.getJobId(), reduceContext.getUser());
+ }
+ LOG.trace("Calling sendMapOutput; channel='{}'", reduceContext.ctx.channel().id());
+ ChannelFuture nextMap = sendMapOutput(
+ reduceContext.getCtx().channel(),
+ reduceContext.getUser(), mapId,
+ reduceContext.getReduceId(), info);
+ nextMap.addListener(new ReduceMapFileCount(this, reduceContext));
+ } catch (IOException e) {
+ LOG.error("Shuffle error: {}; channel={}", e, reduceContext.ctx.channel().id());
+
+ // It is not possible to sendError, the success HttpResponse has been already sent
+ reduceContext.ctx.channel().close();
+ }
+ }
+ }
+
+ private String getErrorMessage(Throwable t) {
+ StringBuilder sb = new StringBuilder(t.getMessage());
+ while (t.getCause() != null) {
+ sb.append(t.getCause().getMessage());
+ t = t.getCause();
+ }
+ return sb.toString();
+ }
+
+ protected MapOutputInfo getMapOutputInfo(String mapId, int reduce, String jobId, String user)
+ throws IOException {
+ ShuffleHandler.AttemptPathInfo pathInfo;
+ try {
+ ShuffleHandler.AttemptPathIdentifier identifier = new ShuffleHandler.AttemptPathIdentifier(
+ jobId, user, mapId);
+ pathInfo = handlerCtx.pathCache.get(identifier);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Retrieved pathInfo for " + identifier +
+ " check for corresponding loaded messages to determine whether" +
+ " it was loaded or cached");
+ }
+ } catch (ExecutionException e) {
+ if (e.getCause() instanceof IOException) {
+ throw (IOException) e.getCause();
+ } else {
+ throw new RuntimeException(e.getCause());
+ }
+ }
+
+ IndexRecord info =
+ handlerCtx.indexCache.getIndexInformation(mapId, reduce, pathInfo.indexPath, user);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("getMapOutputInfo: jobId=" + jobId + ", mapId=" + mapId +
+ ",dataFile=" + pathInfo.dataPath + ", indexFile=" +
+ pathInfo.indexPath);
+ LOG.debug("getMapOutputInfo: startOffset={}, partLength={} rawLength={}",
+ info.startOffset, info.partLength, info.rawLength);
+ }
+
+ return new MapOutputInfo(pathInfo.dataPath, info);
+ }
+
+ protected void populateHeaders(List<String> mapIds, String jobId,
+ String user, int reduce, HttpResponse response,
+ boolean keepAliveParam,
+ Map<String, MapOutputInfo> mapOutputInfoMap)
+ throws IOException {
+
+ long contentLength = 0;
+ for (String mapId : mapIds) {
+ MapOutputInfo outputInfo = getMapOutputInfo(mapId, reduce, jobId, user);
+ if (mapOutputInfoMap.size() < handlerCtx.mapOutputMetaInfoCacheSize) {
+ mapOutputInfoMap.put(mapId, outputInfo);
+ }
+
+ ShuffleHeader header =
+ new ShuffleHeader(mapId, outputInfo.indexRecord.partLength,
+ outputInfo.indexRecord.rawLength, reduce);
+ DataOutputBuffer dob = new DataOutputBuffer();
+ header.write(dob);
+ contentLength += outputInfo.indexRecord.partLength;
+ contentLength += dob.getLength();
+
+ // verify file access to data file to send an actually correct http error
+ final File spillFile = new File(outputInfo.mapOutputFileName.toString());
+ RandomAccessFile r = SecureIOUtils.openForRandomRead(spillFile, "r", user, null);
+ r.close();
+ }
+
+ // Now set the response headers.
+ setResponseHeaders(response, keepAliveParam, contentLength);
+
+ // this audit log is disabled by default,
+ // to turn it on please enable this audit log
+ // on log4j.properties by uncommenting the setting
+ if (AUDITLOG.isDebugEnabled()) {
+ StringBuilder sb = new StringBuilder("shuffle for ");
+ sb.append(jobId).append(" reducer ").append(reduce);
+ sb.append(" length ").append(contentLength);
+ if (AUDITLOG.isTraceEnabled()) {
+ // For trace level logging, append the list of mappers
+ sb.append(" mappers: ").append(mapIds);
+ AUDITLOG.trace(sb.toString());
+ } else {
+ AUDITLOG.debug(sb.toString());
+ }
+ }
+ }
+
+ protected void setResponseHeaders(HttpResponse response,
+ boolean keepAliveParam, long contentLength) {
+ if (!handlerCtx.connectionKeepAliveEnabled && !keepAliveParam) {
+ response.headers().set(HttpHeader.CONNECTION.asString(), CONNECTION_CLOSE);
+ } else {
+ response.headers().set(HttpHeader.CONNECTION.asString(),
+ HttpHeader.KEEP_ALIVE.asString());
+ response.headers().set(HttpHeader.KEEP_ALIVE.asString(),
+ "timeout=" + handlerCtx.connectionKeepAliveTimeOut);
+ }
+
+ // Content length must be set (https://www.rfc-editor.org/rfc/rfc7230#section-3.3.3)
+ HttpUtil.setContentLength(response, contentLength);
+ }
+
+ @SuppressWarnings("checkstyle:VisibilityModifier")
+ static class MapOutputInfo {
+ final Path mapOutputFileName;
+ final IndexRecord indexRecord;
+
+ MapOutputInfo(Path mapOutputFileName, IndexRecord indexRecord) {
+ this.mapOutputFileName = mapOutputFileName;
+ this.indexRecord = indexRecord;
+ }
+ }
+
+ protected void verifyRequest(String appid, ChannelHandlerContext ctx,
+ HttpRequest request, HttpResponse response, URL requestUri)
+ throws IOException {
+ SecretKey tokenSecret = handlerCtx.secretManager.retrieveTokenSecret(appid);
+ if (null == tokenSecret) {
+ LOG.info("Request for unknown token {}, channel id: {}", appid, ctx.channel().id());
+ throw new IOException("Could not find jobid");
+ }
+ // encrypting URL
+ String encryptedURL = SecureShuffleUtils.buildMsgFrom(requestUri);
+ // hash from the fetcher
+ String urlHashStr =
+ request.headers().get(SecureShuffleUtils.HTTP_HEADER_URL_HASH);
+ if (urlHashStr == null) {
+ LOG.info("Missing header hash for {}, channel id: {}", appid, ctx.channel().id());
+ throw new IOException("fetcher cannot be authenticated");
+ }
+ if (LOG.isDebugEnabled()) {
+ int len = urlHashStr.length();
+ LOG.debug("Verifying request. encryptedURL:{}, hash:{}, channel id: " +
+ "{}", encryptedURL,
+ urlHashStr.substring(len - len / 2, len - 1), ctx.channel().id());
+ }
+ // verify - throws exception
+ SecureShuffleUtils.verifyReply(urlHashStr, encryptedURL, tokenSecret);
+ // verification passed - encode the reply
+ String reply = SecureShuffleUtils.generateHash(urlHashStr.getBytes(Charsets.UTF_8),
+ tokenSecret);
+ response.headers().set(
+ SecureShuffleUtils.HTTP_HEADER_REPLY_URL_HASH, reply);
+ // Put shuffle version into http header
+ response.headers().set(ShuffleHeader.HTTP_HEADER_NAME,
+ ShuffleHeader.DEFAULT_HTTP_HEADER_NAME);
+ response.headers().set(ShuffleHeader.HTTP_HEADER_VERSION,
+ ShuffleHeader.DEFAULT_HTTP_HEADER_VERSION);
+ if (LOG.isDebugEnabled()) {
+ int len = reply.length();
+ LOG.debug("Fetcher request verified. " +
+ "encryptedURL: {}, reply: {}, channel id: {}",
+ encryptedURL, reply.substring(len - len / 2, len - 1),
+ ctx.channel().id());
+ }
+ }
+
+ public static ByteBuf shuffleHeaderToBytes(ShuffleHeader header) throws IOException {
+ final DataOutputBuffer dob = new DataOutputBuffer();
+ header.write(dob);
+ return wrappedBuffer(dob.getData(), 0, dob.getLength());
+ }
+
+ protected ChannelFuture sendMapOutput(Channel ch, String user, String mapId, int reduce,
+ MapOutputInfo mapOutputInfo)
+ throws IOException {
+ final IndexRecord info = mapOutputInfo.indexRecord;
+ ch.write(shuffleHeaderToBytes(
+ new ShuffleHeader(mapId, info.partLength, info.rawLength, reduce)));
+ final File spillFile =
+ new File(mapOutputInfo.mapOutputFileName.toString());
+ RandomAccessFile spill = SecureIOUtils.openForRandomRead(spillFile, "r", user, null);
Review Comment:
This should not be closed?
--
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: common-issues-unsubscribe@hadoop.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: common-issues-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-issues-help@hadoop.apache.org
[GitHub] [hadoop] hadoop-yetus commented on pull request #5311: MAPREDUCE-7431. ShuffleHandler refactor and fix after Netty4 upgrade.
Posted by GitBox <gi...@apache.org>.
hadoop-yetus commented on PR #5311:
URL: https://github.com/apache/hadoop/pull/5311#issuecomment-1396860917
:confetti_ball: **+1 overall**
| Vote | Subsystem | Runtime | Logfile | Comment |
|:----:|----------:|--------:|:--------:|:-------:|
| +0 :ok: | reexec | 0m 50s | | Docker mode activated. |
|||| _ Prechecks _ |
| +1 :green_heart: | dupname | 0m 0s | | No case conflicting files found. |
| +0 :ok: | codespell | 0m 0s | | codespell was not available. |
| +0 :ok: | detsecrets | 0m 0s | | detect-secrets was not available. |
| +0 :ok: | xmllint | 0m 0s | | xmllint was not available. |
| +1 :green_heart: | @author | 0m 0s | | The patch does not contain any @author tags. |
| +1 :green_heart: | test4tests | 0m 0s | | The patch appears to include 6 new or modified test files. |
|||| _ trunk Compile Tests _ |
| +1 :green_heart: | mvninstall | 46m 28s | | trunk passed |
| +1 :green_heart: | compile | 0m 25s | | trunk passed with JDK Ubuntu-11.0.17+8-post-Ubuntu-1ubuntu220.04 |
| +1 :green_heart: | compile | 0m 23s | | trunk passed with JDK Private Build-1.8.0_352-8u352-ga-1~20.04-b08 |
| +1 :green_heart: | checkstyle | 0m 25s | | trunk passed |
| +1 :green_heart: | mvnsite | 0m 28s | | trunk passed |
| +1 :green_heart: | javadoc | 0m 31s | | trunk passed with JDK Ubuntu-11.0.17+8-post-Ubuntu-1ubuntu220.04 |
| +1 :green_heart: | javadoc | 0m 21s | | trunk passed with JDK Private Build-1.8.0_352-8u352-ga-1~20.04-b08 |
| +1 :green_heart: | spotbugs | 0m 52s | | trunk passed |
| +1 :green_heart: | shadedclient | 26m 52s | | branch has no errors when building and testing our client artifacts. |
|||| _ Patch Compile Tests _ |
| +1 :green_heart: | mvninstall | 0m 33s | | the patch passed |
| +1 :green_heart: | compile | 0m 21s | | the patch passed with JDK Ubuntu-11.0.17+8-post-Ubuntu-1ubuntu220.04 |
| +1 :green_heart: | javac | 0m 21s | | the patch passed |
| +1 :green_heart: | compile | 0m 19s | | the patch passed with JDK Private Build-1.8.0_352-8u352-ga-1~20.04-b08 |
| +1 :green_heart: | javac | 0m 19s | | the patch passed |
| +1 :green_heart: | blanks | 0m 0s | | The patch has no blanks issues. |
| -0 :warning: | checkstyle | 0m 13s | [/results-checkstyle-hadoop-mapreduce-project_hadoop-mapreduce-client_hadoop-mapreduce-client-shuffle.txt](https://ci-hadoop.apache.org/job/hadoop-multibranch/job/PR-5311/4/artifact/out/results-checkstyle-hadoop-mapreduce-project_hadoop-mapreduce-client_hadoop-mapreduce-client-shuffle.txt) | hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-shuffle: The patch generated 1 new + 5 unchanged - 32 fixed = 6 total (was 37) |
| +1 :green_heart: | mvnsite | 0m 20s | | the patch passed |
| +1 :green_heart: | javadoc | 0m 17s | | hadoop-mapreduce-project_hadoop-mapreduce-client_hadoop-mapreduce-client-shuffle-jdkUbuntu-11.0.17+8-post-Ubuntu-1ubuntu220.04 with JDK Ubuntu-11.0.17+8-post-Ubuntu-1ubuntu220.04 generated 0 new + 0 unchanged - 3 fixed = 0 total (was 3) |
| +1 :green_heart: | javadoc | 0m 16s | | hadoop-mapreduce-project_hadoop-mapreduce-client_hadoop-mapreduce-client-shuffle-jdkPrivateBuild-1.8.0_352-8u352-ga-1~20.04-b08 with JDK Private Build-1.8.0_352-8u352-ga-1~20.04-b08 generated 0 new + 0 unchanged - 3 fixed = 0 total (was 3) |
| +1 :green_heart: | spotbugs | 0m 45s | | the patch passed |
| +1 :green_heart: | shadedclient | 27m 3s | | patch has no errors when building and testing our client artifacts. |
|||| _ Other Tests _ |
| +1 :green_heart: | unit | 0m 36s | | hadoop-mapreduce-client-shuffle in the patch passed. |
| +1 :green_heart: | asflicense | 0m 36s | | The patch does not generate ASF License warnings. |
| | | 110m 39s | | |
| Subsystem | Report/Notes |
|----------:|:-------------|
| Docker | ClientAPI=1.41 ServerAPI=1.41 base: https://ci-hadoop.apache.org/job/hadoop-multibranch/job/PR-5311/4/artifact/out/Dockerfile |
| GITHUB PR | https://github.com/apache/hadoop/pull/5311 |
| Optional Tests | dupname asflicense compile javac javadoc mvninstall mvnsite unit shadedclient codespell detsecrets xmllint spotbugs checkstyle |
| uname | Linux 2b5931453a0f 4.15.0-200-generic #211-Ubuntu SMP Thu Nov 24 18:16:04 UTC 2022 x86_64 x86_64 x86_64 GNU/Linux |
| Build tool | maven |
| Personality | dev-support/bin/hadoop.sh |
| git revision | trunk / 59a1f409101c47c76bb3078987a4cd41f4f68e63 |
| Default Java | Private Build-1.8.0_352-8u352-ga-1~20.04-b08 |
| Multi-JDK versions | /usr/lib/jvm/java-11-openjdk-amd64:Ubuntu-11.0.17+8-post-Ubuntu-1ubuntu220.04 /usr/lib/jvm/java-8-openjdk-amd64:Private Build-1.8.0_352-8u352-ga-1~20.04-b08 |
| Test Results | https://ci-hadoop.apache.org/job/hadoop-multibranch/job/PR-5311/4/testReport/ |
| Max. process+thread count | 585 (vs. ulimit of 5500) |
| modules | C: hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-shuffle U: hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-shuffle |
| Console output | https://ci-hadoop.apache.org/job/hadoop-multibranch/job/PR-5311/4/console |
| versions | git=2.25.1 maven=3.6.3 spotbugs=4.2.2 |
| Powered by | Apache Yetus 0.14.0 https://yetus.apache.org |
This message was automatically generated.
--
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: common-issues-unsubscribe@hadoop.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: common-issues-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-issues-help@hadoop.apache.org
[GitHub] [hadoop] brumi1024 commented on pull request #5311: MAPREDUCE-7431. ShuffleHandler refactor and fix after Netty4 upgrade.
Posted by "brumi1024 (via GitHub)" <gi...@apache.org>.
brumi1024 commented on PR #5311:
URL: https://github.com/apache/hadoop/pull/5311#issuecomment-1426065639
Thanks @tomicooler, the patch looks good to me, merged to trunk. Thanks @K0K0V0K for the review.
--
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: common-issues-unsubscribe@hadoop.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: common-issues-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-issues-help@hadoop.apache.org
[GitHub] [hadoop] K0K0V0K commented on pull request #5311: MAPREDUCE-7431. ShuffleHandler refactor and fix after Netty4 upgrade.
Posted by GitBox <gi...@apache.org>.
K0K0V0K commented on PR #5311:
URL: https://github.com/apache/hadoop/pull/5311#issuecomment-1397239024
+1, non-binding
--
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: common-issues-unsubscribe@hadoop.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: common-issues-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-issues-help@hadoop.apache.org
[GitHub] [hadoop] ayushtkn commented on pull request #5311: MAPREDUCE-7431. ShuffleHandler refactor and fix after Netty4 upgrade.
Posted by "ayushtkn (via GitHub)" <gi...@apache.org>.
ayushtkn commented on PR #5311:
URL: https://github.com/apache/hadoop/pull/5311#issuecomment-1439380291
the test is failing now:
https://ci-hadoop.apache.org/view/Hadoop/job/hadoop-qbt-trunk-java8-linux-x86_64/1143/testReport/junit/org.apache.hadoop.mapred/TestShuffleHandler/testMapFileAccess/
Though it passes locally for me.
Looks like something specific to jenkins:
```
Caused by: java.io.IOException: Owner 'jenkins' for path /tmp/test-shuffle-channel-handler1169959588626711767/job_1111111111111_0001/testUser/attempt_1111111111111_0001_m_000001_0/index did not match expected owner 'testUser'
```
anyone planning to fix or should we revert this?
--
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: common-issues-unsubscribe@hadoop.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: common-issues-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-issues-help@hadoop.apache.org
[GitHub] [hadoop] hadoop-yetus commented on pull request #5311: MAPREDUCE-7431. ShuffleHandler refactor and fix after Netty4 upgrade.
Posted by GitBox <gi...@apache.org>.
hadoop-yetus commented on PR #5311:
URL: https://github.com/apache/hadoop/pull/5311#issuecomment-1387525353
:broken_heart: **-1 overall**
| Vote | Subsystem | Runtime | Logfile | Comment |
|:----:|----------:|--------:|:--------:|:-------:|
| +0 :ok: | reexec | 0m 48s | | Docker mode activated. |
|||| _ Prechecks _ |
| +1 :green_heart: | dupname | 0m 0s | | No case conflicting files found. |
| +0 :ok: | codespell | 0m 1s | | codespell was not available. |
| +0 :ok: | detsecrets | 0m 1s | | detect-secrets was not available. |
| +0 :ok: | xmllint | 0m 1s | | xmllint was not available. |
| +1 :green_heart: | @author | 0m 0s | | The patch does not contain any @author tags. |
| +1 :green_heart: | test4tests | 0m 0s | | The patch appears to include 6 new or modified test files. |
|||| _ trunk Compile Tests _ |
| +1 :green_heart: | mvninstall | 48m 10s | | trunk passed |
| +1 :green_heart: | compile | 0m 26s | | trunk passed with JDK Ubuntu-11.0.17+8-post-Ubuntu-1ubuntu220.04 |
| +1 :green_heart: | compile | 0m 23s | | trunk passed with JDK Private Build-1.8.0_352-8u352-ga-1~20.04-b08 |
| +1 :green_heart: | checkstyle | 0m 24s | | trunk passed |
| +1 :green_heart: | mvnsite | 0m 29s | | trunk passed |
| +1 :green_heart: | javadoc | 0m 31s | | trunk passed with JDK Ubuntu-11.0.17+8-post-Ubuntu-1ubuntu220.04 |
| +1 :green_heart: | javadoc | 0m 20s | | trunk passed with JDK Private Build-1.8.0_352-8u352-ga-1~20.04-b08 |
| +1 :green_heart: | spotbugs | 0m 52s | | trunk passed |
| +1 :green_heart: | shadedclient | 26m 47s | | branch has no errors when building and testing our client artifacts. |
|||| _ Patch Compile Tests _ |
| +1 :green_heart: | mvninstall | 0m 31s | | the patch passed |
| +1 :green_heart: | compile | 0m 19s | | the patch passed with JDK Ubuntu-11.0.17+8-post-Ubuntu-1ubuntu220.04 |
| +1 :green_heart: | javac | 0m 19s | | the patch passed |
| +1 :green_heart: | compile | 0m 18s | | the patch passed with JDK Private Build-1.8.0_352-8u352-ga-1~20.04-b08 |
| +1 :green_heart: | javac | 0m 18s | | the patch passed |
| +1 :green_heart: | blanks | 0m 0s | | The patch has no blanks issues. |
| -0 :warning: | checkstyle | 0m 12s | [/results-checkstyle-hadoop-mapreduce-project_hadoop-mapreduce-client_hadoop-mapreduce-client-shuffle.txt](https://ci-hadoop.apache.org/job/hadoop-multibranch/job/PR-5311/1/artifact/out/results-checkstyle-hadoop-mapreduce-project_hadoop-mapreduce-client_hadoop-mapreduce-client-shuffle.txt) | hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-shuffle: The patch generated 11 new + 8 unchanged - 29 fixed = 19 total (was 37) |
| +1 :green_heart: | mvnsite | 0m 21s | | the patch passed |
| -1 :x: | javadoc | 0m 18s | [/patch-javadoc-hadoop-mapreduce-project_hadoop-mapreduce-client_hadoop-mapreduce-client-shuffle-jdkUbuntu-11.0.17+8-post-Ubuntu-1ubuntu220.04.txt](https://ci-hadoop.apache.org/job/hadoop-multibranch/job/PR-5311/1/artifact/out/patch-javadoc-hadoop-mapreduce-project_hadoop-mapreduce-client_hadoop-mapreduce-client-shuffle-jdkUbuntu-11.0.17+8-post-Ubuntu-1ubuntu220.04.txt) | hadoop-mapreduce-client-shuffle in the patch failed with JDK Ubuntu-11.0.17+8-post-Ubuntu-1ubuntu220.04. |
| -1 :x: | javadoc | 0m 16s | [/patch-javadoc-hadoop-mapreduce-project_hadoop-mapreduce-client_hadoop-mapreduce-client-shuffle-jdkPrivateBuild-1.8.0_352-8u352-ga-1~20.04-b08.txt](https://ci-hadoop.apache.org/job/hadoop-multibranch/job/PR-5311/1/artifact/out/patch-javadoc-hadoop-mapreduce-project_hadoop-mapreduce-client_hadoop-mapreduce-client-shuffle-jdkPrivateBuild-1.8.0_352-8u352-ga-1~20.04-b08.txt) | hadoop-mapreduce-client-shuffle in the patch failed with JDK Private Build-1.8.0_352-8u352-ga-1~20.04-b08. |
| +1 :green_heart: | spotbugs | 0m 47s | | the patch passed |
| +1 :green_heart: | shadedclient | 26m 18s | | patch has no errors when building and testing our client artifacts. |
|||| _ Other Tests _ |
| +1 :green_heart: | unit | 0m 35s | | hadoop-mapreduce-client-shuffle in the patch passed. |
| -1 :x: | asflicense | 0m 33s | [/results-asflicense.txt](https://ci-hadoop.apache.org/job/hadoop-multibranch/job/PR-5311/1/artifact/out/results-asflicense.txt) | The patch generated 1 ASF License warnings. |
| | | 111m 5s | | |
| Subsystem | Report/Notes |
|----------:|:-------------|
| Docker | ClientAPI=1.41 ServerAPI=1.41 base: https://ci-hadoop.apache.org/job/hadoop-multibranch/job/PR-5311/1/artifact/out/Dockerfile |
| GITHUB PR | https://github.com/apache/hadoop/pull/5311 |
| Optional Tests | dupname asflicense compile javac javadoc mvninstall mvnsite unit shadedclient codespell detsecrets xmllint spotbugs checkstyle |
| uname | Linux ebbc81ae26e4 4.15.0-200-generic #211-Ubuntu SMP Thu Nov 24 18:16:04 UTC 2022 x86_64 x86_64 x86_64 GNU/Linux |
| Build tool | maven |
| Personality | dev-support/bin/hadoop.sh |
| git revision | trunk / ea8f12ade37f0739ddb48d317bea64c4c2ba59b1 |
| Default Java | Private Build-1.8.0_352-8u352-ga-1~20.04-b08 |
| Multi-JDK versions | /usr/lib/jvm/java-11-openjdk-amd64:Ubuntu-11.0.17+8-post-Ubuntu-1ubuntu220.04 /usr/lib/jvm/java-8-openjdk-amd64:Private Build-1.8.0_352-8u352-ga-1~20.04-b08 |
| Test Results | https://ci-hadoop.apache.org/job/hadoop-multibranch/job/PR-5311/1/testReport/ |
| Max. process+thread count | 538 (vs. ulimit of 5500) |
| modules | C: hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-shuffle U: hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-shuffle |
| Console output | https://ci-hadoop.apache.org/job/hadoop-multibranch/job/PR-5311/1/console |
| versions | git=2.25.1 maven=3.6.3 spotbugs=4.2.2 |
| Powered by | Apache Yetus 0.14.0 https://yetus.apache.org |
This message was automatically generated.
--
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: common-issues-unsubscribe@hadoop.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: common-issues-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-issues-help@hadoop.apache.org
[GitHub] [hadoop] ayushtkn commented on pull request #5311: MAPREDUCE-7431. ShuffleHandler refactor and fix after Netty4 upgrade.
Posted by "ayushtkn (via GitHub)" <gi...@apache.org>.
ayushtkn commented on PR #5311:
URL: https://github.com/apache/hadoop/pull/5311#issuecomment-1439571729
Thank you Tamas for volunteering!!!
--
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: common-issues-unsubscribe@hadoop.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: common-issues-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-issues-help@hadoop.apache.org
[GitHub] [hadoop] hadoop-yetus commented on pull request #5311: MAPREDUCE-7431. ShuffleHandler refactor and fix after Netty4 upgrade.
Posted by GitBox <gi...@apache.org>.
hadoop-yetus commented on PR #5311:
URL: https://github.com/apache/hadoop/pull/5311#issuecomment-1387533352
:broken_heart: **-1 overall**
| Vote | Subsystem | Runtime | Logfile | Comment |
|:----:|----------:|--------:|:--------:|:-------:|
| +0 :ok: | reexec | 0m 53s | | Docker mode activated. |
|||| _ Prechecks _ |
| +1 :green_heart: | dupname | 0m 0s | | No case conflicting files found. |
| +0 :ok: | codespell | 0m 1s | | codespell was not available. |
| +0 :ok: | detsecrets | 0m 1s | | detect-secrets was not available. |
| +0 :ok: | xmllint | 0m 1s | | xmllint was not available. |
| +1 :green_heart: | @author | 0m 0s | | The patch does not contain any @author tags. |
| +1 :green_heart: | test4tests | 0m 0s | | The patch appears to include 6 new or modified test files. |
|||| _ trunk Compile Tests _ |
| +1 :green_heart: | mvninstall | 47m 56s | | trunk passed |
| +1 :green_heart: | compile | 0m 28s | | trunk passed with JDK Ubuntu-11.0.17+8-post-Ubuntu-1ubuntu220.04 |
| +1 :green_heart: | compile | 0m 25s | | trunk passed with JDK Private Build-1.8.0_352-8u352-ga-1~20.04-b08 |
| +1 :green_heart: | checkstyle | 0m 25s | | trunk passed |
| +1 :green_heart: | mvnsite | 0m 27s | | trunk passed |
| +1 :green_heart: | javadoc | 0m 32s | | trunk passed with JDK Ubuntu-11.0.17+8-post-Ubuntu-1ubuntu220.04 |
| +1 :green_heart: | javadoc | 0m 21s | | trunk passed with JDK Private Build-1.8.0_352-8u352-ga-1~20.04-b08 |
| +1 :green_heart: | spotbugs | 0m 52s | | trunk passed |
| +1 :green_heart: | shadedclient | 25m 51s | | branch has no errors when building and testing our client artifacts. |
|||| _ Patch Compile Tests _ |
| +1 :green_heart: | mvninstall | 0m 33s | | the patch passed |
| +1 :green_heart: | compile | 0m 22s | | the patch passed with JDK Ubuntu-11.0.17+8-post-Ubuntu-1ubuntu220.04 |
| +1 :green_heart: | javac | 0m 22s | | the patch passed |
| +1 :green_heart: | compile | 0m 18s | | the patch passed with JDK Private Build-1.8.0_352-8u352-ga-1~20.04-b08 |
| +1 :green_heart: | javac | 0m 18s | | the patch passed |
| +1 :green_heart: | blanks | 0m 0s | | The patch has no blanks issues. |
| -0 :warning: | checkstyle | 0m 11s | [/results-checkstyle-hadoop-mapreduce-project_hadoop-mapreduce-client_hadoop-mapreduce-client-shuffle.txt](https://ci-hadoop.apache.org/job/hadoop-multibranch/job/PR-5311/2/artifact/out/results-checkstyle-hadoop-mapreduce-project_hadoop-mapreduce-client_hadoop-mapreduce-client-shuffle.txt) | hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-shuffle: The patch generated 11 new + 8 unchanged - 29 fixed = 19 total (was 37) |
| +1 :green_heart: | mvnsite | 0m 22s | | the patch passed |
| -1 :x: | javadoc | 0m 17s | [/patch-javadoc-hadoop-mapreduce-project_hadoop-mapreduce-client_hadoop-mapreduce-client-shuffle-jdkUbuntu-11.0.17+8-post-Ubuntu-1ubuntu220.04.txt](https://ci-hadoop.apache.org/job/hadoop-multibranch/job/PR-5311/2/artifact/out/patch-javadoc-hadoop-mapreduce-project_hadoop-mapreduce-client_hadoop-mapreduce-client-shuffle-jdkUbuntu-11.0.17+8-post-Ubuntu-1ubuntu220.04.txt) | hadoop-mapreduce-client-shuffle in the patch failed with JDK Ubuntu-11.0.17+8-post-Ubuntu-1ubuntu220.04. |
| -1 :x: | javadoc | 0m 16s | [/patch-javadoc-hadoop-mapreduce-project_hadoop-mapreduce-client_hadoop-mapreduce-client-shuffle-jdkPrivateBuild-1.8.0_352-8u352-ga-1~20.04-b08.txt](https://ci-hadoop.apache.org/job/hadoop-multibranch/job/PR-5311/2/artifact/out/patch-javadoc-hadoop-mapreduce-project_hadoop-mapreduce-client_hadoop-mapreduce-client-shuffle-jdkPrivateBuild-1.8.0_352-8u352-ga-1~20.04-b08.txt) | hadoop-mapreduce-client-shuffle in the patch failed with JDK Private Build-1.8.0_352-8u352-ga-1~20.04-b08. |
| +1 :green_heart: | spotbugs | 0m 46s | | the patch passed |
| +1 :green_heart: | shadedclient | 26m 22s | | patch has no errors when building and testing our client artifacts. |
|||| _ Other Tests _ |
| +1 :green_heart: | unit | 0m 36s | | hadoop-mapreduce-client-shuffle in the patch passed. |
| -1 :x: | asflicense | 0m 33s | [/results-asflicense.txt](https://ci-hadoop.apache.org/job/hadoop-multibranch/job/PR-5311/2/artifact/out/results-asflicense.txt) | The patch generated 1 ASF License warnings. |
| | | 110m 18s | | |
| Subsystem | Report/Notes |
|----------:|:-------------|
| Docker | ClientAPI=1.41 ServerAPI=1.41 base: https://ci-hadoop.apache.org/job/hadoop-multibranch/job/PR-5311/2/artifact/out/Dockerfile |
| GITHUB PR | https://github.com/apache/hadoop/pull/5311 |
| Optional Tests | dupname asflicense compile javac javadoc mvninstall mvnsite unit shadedclient codespell detsecrets xmllint spotbugs checkstyle |
| uname | Linux 4edd60006a54 4.15.0-200-generic #211-Ubuntu SMP Thu Nov 24 18:16:04 UTC 2022 x86_64 x86_64 x86_64 GNU/Linux |
| Build tool | maven |
| Personality | dev-support/bin/hadoop.sh |
| git revision | trunk / 8fefb83f680483ba0f8efa0094e22ab1d6f8bf8e |
| Default Java | Private Build-1.8.0_352-8u352-ga-1~20.04-b08 |
| Multi-JDK versions | /usr/lib/jvm/java-11-openjdk-amd64:Ubuntu-11.0.17+8-post-Ubuntu-1ubuntu220.04 /usr/lib/jvm/java-8-openjdk-amd64:Private Build-1.8.0_352-8u352-ga-1~20.04-b08 |
| Test Results | https://ci-hadoop.apache.org/job/hadoop-multibranch/job/PR-5311/2/testReport/ |
| Max. process+thread count | 630 (vs. ulimit of 5500) |
| modules | C: hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-shuffle U: hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-shuffle |
| Console output | https://ci-hadoop.apache.org/job/hadoop-multibranch/job/PR-5311/2/console |
| versions | git=2.25.1 maven=3.6.3 spotbugs=4.2.2 |
| Powered by | Apache Yetus 0.14.0 https://yetus.apache.org |
This message was automatically generated.
--
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: common-issues-unsubscribe@hadoop.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: common-issues-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-issues-help@hadoop.apache.org
[GitHub] [hadoop] hadoop-yetus commented on pull request #5311: MAPREDUCE-7431. ShuffleHandler refactor and fix after Netty4 upgrade.
Posted by GitBox <gi...@apache.org>.
hadoop-yetus commented on PR #5311:
URL: https://github.com/apache/hadoop/pull/5311#issuecomment-1396671969
:broken_heart: **-1 overall**
| Vote | Subsystem | Runtime | Logfile | Comment |
|:----:|----------:|--------:|:--------:|:-------:|
| +0 :ok: | reexec | 0m 49s | | Docker mode activated. |
|||| _ Prechecks _ |
| +1 :green_heart: | dupname | 0m 1s | | No case conflicting files found. |
| +0 :ok: | codespell | 0m 0s | | codespell was not available. |
| +0 :ok: | detsecrets | 0m 0s | | detect-secrets was not available. |
| +0 :ok: | xmllint | 0m 0s | | xmllint was not available. |
| +1 :green_heart: | @author | 0m 0s | | The patch does not contain any @author tags. |
| +1 :green_heart: | test4tests | 0m 0s | | The patch appears to include 6 new or modified test files. |
|||| _ trunk Compile Tests _ |
| -1 :x: | mvninstall | 45m 0s | [/branch-mvninstall-root.txt](https://ci-hadoop.apache.org/job/hadoop-multibranch/job/PR-5311/3/artifact/out/branch-mvninstall-root.txt) | root in trunk failed. |
| +1 :green_heart: | compile | 0m 26s | | trunk passed with JDK Ubuntu-11.0.17+8-post-Ubuntu-1ubuntu220.04 |
| +1 :green_heart: | compile | 0m 24s | | trunk passed with JDK Private Build-1.8.0_352-8u352-ga-1~20.04-b08 |
| +1 :green_heart: | checkstyle | 0m 24s | | trunk passed |
| +1 :green_heart: | mvnsite | 0m 26s | | trunk passed |
| +1 :green_heart: | javadoc | 0m 34s | | trunk passed with JDK Ubuntu-11.0.17+8-post-Ubuntu-1ubuntu220.04 |
| +1 :green_heart: | javadoc | 0m 22s | | trunk passed with JDK Private Build-1.8.0_352-8u352-ga-1~20.04-b08 |
| +1 :green_heart: | spotbugs | 0m 52s | | trunk passed |
| +1 :green_heart: | shadedclient | 26m 38s | | branch has no errors when building and testing our client artifacts. |
|||| _ Patch Compile Tests _ |
| +1 :green_heart: | mvninstall | 0m 35s | | the patch passed |
| +1 :green_heart: | compile | 0m 24s | | the patch passed with JDK Ubuntu-11.0.17+8-post-Ubuntu-1ubuntu220.04 |
| +1 :green_heart: | javac | 0m 24s | | the patch passed |
| +1 :green_heart: | compile | 0m 20s | | the patch passed with JDK Private Build-1.8.0_352-8u352-ga-1~20.04-b08 |
| +1 :green_heart: | javac | 0m 20s | | the patch passed |
| +1 :green_heart: | blanks | 0m 0s | | The patch has no blanks issues. |
| -0 :warning: | checkstyle | 0m 15s | [/results-checkstyle-hadoop-mapreduce-project_hadoop-mapreduce-client_hadoop-mapreduce-client-shuffle.txt](https://ci-hadoop.apache.org/job/hadoop-multibranch/job/PR-5311/3/artifact/out/results-checkstyle-hadoop-mapreduce-project_hadoop-mapreduce-client_hadoop-mapreduce-client-shuffle.txt) | hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-shuffle: The patch generated 1 new + 5 unchanged - 32 fixed = 6 total (was 37) |
| +1 :green_heart: | mvnsite | 0m 23s | | the patch passed |
| +1 :green_heart: | javadoc | 0m 20s | | hadoop-mapreduce-project_hadoop-mapreduce-client_hadoop-mapreduce-client-shuffle-jdkUbuntu-11.0.17+8-post-Ubuntu-1ubuntu220.04 with JDK Ubuntu-11.0.17+8-post-Ubuntu-1ubuntu220.04 generated 0 new + 0 unchanged - 3 fixed = 0 total (was 3) |
| +1 :green_heart: | javadoc | 0m 18s | | hadoop-mapreduce-project_hadoop-mapreduce-client_hadoop-mapreduce-client-shuffle-jdkPrivateBuild-1.8.0_352-8u352-ga-1~20.04-b08 with JDK Private Build-1.8.0_352-8u352-ga-1~20.04-b08 generated 0 new + 0 unchanged - 3 fixed = 0 total (was 3) |
| +1 :green_heart: | spotbugs | 0m 53s | | the patch passed |
| +1 :green_heart: | shadedclient | 28m 10s | | patch has no errors when building and testing our client artifacts. |
|||| _ Other Tests _ |
| +1 :green_heart: | unit | 0m 37s | | hadoop-mapreduce-client-shuffle in the patch passed. |
| +1 :green_heart: | asflicense | 0m 33s | | The patch does not generate ASF License warnings. |
| | | 110m 42s | | |
| Subsystem | Report/Notes |
|----------:|:-------------|
| Docker | ClientAPI=1.41 ServerAPI=1.41 base: https://ci-hadoop.apache.org/job/hadoop-multibranch/job/PR-5311/3/artifact/out/Dockerfile |
| GITHUB PR | https://github.com/apache/hadoop/pull/5311 |
| Optional Tests | dupname asflicense compile javac javadoc mvninstall mvnsite unit shadedclient codespell detsecrets xmllint spotbugs checkstyle |
| uname | Linux 5c70fe9bae8f 4.15.0-200-generic #211-Ubuntu SMP Thu Nov 24 18:16:04 UTC 2022 x86_64 x86_64 x86_64 GNU/Linux |
| Build tool | maven |
| Personality | dev-support/bin/hadoop.sh |
| git revision | trunk / 6c572c8c9c7f67117069c7c67228a608dd24021a |
| Default Java | Private Build-1.8.0_352-8u352-ga-1~20.04-b08 |
| Multi-JDK versions | /usr/lib/jvm/java-11-openjdk-amd64:Ubuntu-11.0.17+8-post-Ubuntu-1ubuntu220.04 /usr/lib/jvm/java-8-openjdk-amd64:Private Build-1.8.0_352-8u352-ga-1~20.04-b08 |
| Test Results | https://ci-hadoop.apache.org/job/hadoop-multibranch/job/PR-5311/3/testReport/ |
| Max. process+thread count | 615 (vs. ulimit of 5500) |
| modules | C: hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-shuffle U: hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-shuffle |
| Console output | https://ci-hadoop.apache.org/job/hadoop-multibranch/job/PR-5311/3/console |
| versions | git=2.25.1 maven=3.6.3 spotbugs=4.2.2 |
| Powered by | Apache Yetus 0.14.0 https://yetus.apache.org |
This message was automatically generated.
--
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: common-issues-unsubscribe@hadoop.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: common-issues-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-issues-help@hadoop.apache.org
[GitHub] [hadoop] tomicooler commented on pull request #5311: MAPREDUCE-7431. ShuffleHandler refactor and fix after Netty4 upgrade.
Posted by "tomicooler (via GitHub)" <gi...@apache.org>.
tomicooler commented on PR #5311:
URL: https://github.com/apache/hadoop/pull/5311#issuecomment-1439525148
> the test is failing now: https://ci-hadoop.apache.org/view/Hadoop/job/hadoop-qbt-trunk-java8-linux-x86_64/1143/testReport/junit/org.apache.hadoop.mapred/TestShuffleHandler/testMapFileAccess/
>
> Though it passes locally for me. Looks like something specific to jenkins:
>
> ```
> Caused by: java.io.IOException: Owner 'jenkins' for path /tmp/test-shuffle-channel-handler1169959588626711767/job_1111111111111_0001/testUser/attempt_1111111111111_0001_m_000001_0/index did not match expected owner 'testUser'
> ```
>
> anyone planning to fix or should we revert this?
[MAPREDUCE-7434](https://issues.apache.org/jira/browse/MAPREDUCE-7434)
--
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: common-issues-unsubscribe@hadoop.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: common-issues-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-issues-help@hadoop.apache.org