diff --git a/common/network-common/src/main/java/org/apache/spark/network/client/TransportClientFactory.java b/common/network-common/src/main/java/org/apache/spark/network/client/TransportClientFactory.java index e895f13f45458e4a2d13caa877636cb4a47e542e..cb10edff659f183981fa9009592a312532b88c80 100644 --- a/common/network-common/src/main/java/org/apache/spark/network/client/TransportClientFactory.java +++ b/common/network-common/src/main/java/org/apache/spark/network/client/TransportClientFactory.java @@ -100,8 +100,10 @@ public class TransportClientFactory implements Closeable { IOMode ioMode = IOMode.valueOf(conf.ioMode()); this.socketChannelClass = NettyUtils.getClientChannelClass(ioMode); - // TODO: Make thread pool name configurable. - this.workerGroup = NettyUtils.createEventLoop(ioMode, conf.clientThreads(), "shuffle-client"); + this.workerGroup = NettyUtils.createEventLoop( + ioMode, + conf.clientThreads(), + conf.getModuleName() + "-client"); this.pooledAllocator = NettyUtils.createPooledByteBufAllocator( conf.preferDirectBufs(), false /* allowCache */, conf.clientThreads()); } diff --git a/common/network-common/src/main/java/org/apache/spark/network/server/TransportChannelHandler.java b/common/network-common/src/main/java/org/apache/spark/network/server/TransportChannelHandler.java index c33848c8406c19a63d664e72cfb89b31548f76c9..c6ccae18b5e06cdd23c2c6de3f748fab825f66f2 100644 --- a/common/network-common/src/main/java/org/apache/spark/network/server/TransportChannelHandler.java +++ b/common/network-common/src/main/java/org/apache/spark/network/server/TransportChannelHandler.java @@ -88,14 +88,14 @@ public class TransportChannelHandler extends SimpleChannelInboundHandler<Message try { requestHandler.channelActive(); } catch (RuntimeException e) { - logger.error("Exception from request handler while registering channel", e); + logger.error("Exception from request handler while channel is active", e); } try { responseHandler.channelActive(); } catch (RuntimeException e) { - logger.error("Exception from response handler while registering channel", e); + logger.error("Exception from response handler while channel is active", e); } - super.channelRegistered(ctx); + super.channelActive(ctx); } @Override @@ -103,14 +103,14 @@ public class TransportChannelHandler extends SimpleChannelInboundHandler<Message try { requestHandler.channelInactive(); } catch (RuntimeException e) { - logger.error("Exception from request handler while unregistering channel", e); + logger.error("Exception from request handler while channel is inactive", e); } try { responseHandler.channelInactive(); } catch (RuntimeException e) { - logger.error("Exception from response handler while unregistering channel", e); + logger.error("Exception from response handler while channel is inactive", e); } - super.channelUnregistered(ctx); + super.channelInactive(ctx); } @Override diff --git a/common/network-common/src/main/java/org/apache/spark/network/server/TransportServer.java b/common/network-common/src/main/java/org/apache/spark/network/server/TransportServer.java index 0d7a677820d3584897e299eb66dc4d3879af4005..047c5f3f1f094fd7963f8bbb962dfb2eb127cd9e 100644 --- a/common/network-common/src/main/java/org/apache/spark/network/server/TransportServer.java +++ b/common/network-common/src/main/java/org/apache/spark/network/server/TransportServer.java @@ -89,7 +89,7 @@ public class TransportServer implements Closeable { IOMode ioMode = IOMode.valueOf(conf.ioMode()); EventLoopGroup bossGroup = - NettyUtils.createEventLoop(ioMode, conf.serverThreads(), "shuffle-server"); + NettyUtils.createEventLoop(ioMode, conf.serverThreads(), conf.getModuleName() + "-server"); EventLoopGroup workerGroup = bossGroup; PooledByteBufAllocator allocator = NettyUtils.createPooledByteBufAllocator( diff --git a/common/network-common/src/main/java/org/apache/spark/network/util/TransportConf.java b/common/network-common/src/main/java/org/apache/spark/network/util/TransportConf.java index 223d6d88de1595612727381e5d6d9bb88d1f0bdd..6a557fa75d06425d9022b6097a1c6e1daa72be15 100644 --- a/common/network-common/src/main/java/org/apache/spark/network/util/TransportConf.java +++ b/common/network-common/src/main/java/org/apache/spark/network/util/TransportConf.java @@ -70,6 +70,10 @@ public class TransportConf { return "spark." + module + "." + suffix; } + public String getModuleName() { + return module; + } + /** IO mode: nio or epoll */ public String ioMode() { return conf.get(SPARK_NETWORK_IO_MODE_KEY, "NIO").toUpperCase(); } diff --git a/core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala b/core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala index b720aaee7c7e9ef82a56cbedb13bd585a3dc1c5a..f8906117638b3c8a009b971c9035863a75cfddcc 100644 --- a/core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala +++ b/core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala @@ -315,6 +315,10 @@ final class ShuffleBlockFetcherIterator( * Throws a FetchFailedException if the next block could not be fetched. */ override def next(): (BlockId, InputStream) = { + if (!hasNext) { + throw new NoSuchElementException + } + numBlocksProcessed += 1 var result: FetchResult = null