Merge branch 'master' of github.com:apache/spark into SPARK-23937

Author: mgaido91

assembly/README

@@ -9,4 +9,4 @@ This module is off by default. To activate it specify the profile in the command
 
 If you need to build an assembly for a different version of Hadoop the
 hadoop-version system property needs to be set as in this example:
-  -Dhadoop.version=2.7.3
+  -Dhadoop.version=2.7.7

common/kvstore/src/main/java/org/apache/spark/util/kvstore/LevelDB.java

@@ -234,7 +234,7 @@ public void close() throws IOException {
    * Closes the given iterator if the DB is still open. Trying to close a JNI LevelDB handle
    * with a closed DB can cause JVM crashes, so this ensures that situation does not happen.
    */
-  void closeIterator(LevelDBIterator it) throws IOException {
+  void closeIterator(LevelDBIterator<?> it) throws IOException {
     synchronized (this._db) {
       DB _db = this._db.get();
       if (_db != null) {

common/network-common/src/main/java/org/apache/spark/network/client/TransportClient.java

@@ -318,7 +318,7 @@ private class StdChannelListener
     }
 
     @Override
-    public void operationComplete(Future future) throws Exception {
+    public void operationComplete(Future<? super Void> future) throws Exception {
       if (future.isSuccess()) {
         if (logger.isTraceEnabled()) {
           long timeTaken = System.currentTimeMillis() - startTime;

common/network-common/src/main/java/org/apache/spark/network/client/TransportResponseHandler.java

@@ -212,8 +212,8 @@ public void handle(ResponseMessage message) throws Exception {
       if (entry != null) {
         StreamCallback callback = entry.getValue();
         if (resp.byteCount > 0) {
-          StreamInterceptor interceptor = new StreamInterceptor(this, resp.streamId, resp.byteCount,
-            callback);
+          StreamInterceptor<ResponseMessage> interceptor = new StreamInterceptor<>(
+            this, resp.streamId, resp.byteCount, callback);
           try {
             TransportFrameDecoder frameDecoder = (TransportFrameDecoder)
               channel.pipeline().get(TransportFrameDecoder.HANDLER_NAME);

common/network-common/src/main/java/org/apache/spark/network/crypto/TransportCipher.java

@@ -267,7 +267,7 @@ private void encryptMore() throws IOException {
         int copied = byteRawChannel.write(buf.nioBuffer());
         buf.skipBytes(copied);
       } else {
-        region.transferTo(byteRawChannel, region.transfered());
+        region.transferTo(byteRawChannel, region.transferred());
       }
       cos.write(byteRawChannel.getData(), 0, byteRawChannel.length());
       cos.flush();

common/network-common/src/main/java/org/apache/spark/network/sasl/SaslEncryption.java

@@ -301,7 +301,7 @@ private void nextChunk() throws IOException {
         int copied = byteChannel.write(buf.nioBuffer());
         buf.skipBytes(copied);
       } else {
-        region.transferTo(byteChannel, region.transfered());
+        region.transferTo(byteChannel, region.transferred());
       }
 
       byte[] encrypted = backend.wrap(byteChannel.getData(), 0, byteChannel.length());

common/network-common/src/main/java/org/apache/spark/network/server/TransportRequestHandler.java

@@ -252,8 +252,8 @@ public String getID() {
         }
       };
       if (req.bodyByteCount > 0) {
-        StreamInterceptor interceptor = new StreamInterceptor(this, wrappedCallback.getID(),
-          req.bodyByteCount, wrappedCallback);
+        StreamInterceptor<RequestMessage> interceptor = new StreamInterceptor<>(
+          this, wrappedCallback.getID(), req.bodyByteCount, wrappedCallback);
         frameDecoder.setInterceptor(interceptor);
       } else {
         wrappedCallback.onComplete(wrappedCallback.getID());

common/network-common/src/main/java/org/apache/spark/network/server/TransportServer.java

@@ -148,11 +148,11 @@ public void close() {
       channelFuture.channel().close().awaitUninterruptibly(10, TimeUnit.SECONDS);
       channelFuture = null;
     }
-    if (bootstrap != null && bootstrap.group() != null) {
-      bootstrap.group().shutdownGracefully();
+    if (bootstrap != null && bootstrap.config().group() != null) {
+      bootstrap.config().group().shutdownGracefully();
     }
-    if (bootstrap != null && bootstrap.childGroup() != null) {
-      bootstrap.childGroup().shutdownGracefully();
+    if (bootstrap != null && bootstrap.config().childGroup() != null) {
+      bootstrap.config().childGroup().shutdownGracefully();
     }
     bootstrap = null;
   }

common/network-common/src/main/java/org/apache/spark/network/util/NettyUtils.java

@@ -17,7 +17,6 @@
 
 package org.apache.spark.network.util;
 
-import java.lang.reflect.Field;
 import java.util.concurrent.ThreadFactory;
 
 import io.netty.buffer.PooledByteBufAllocator;
@@ -111,24 +110,14 @@ public static PooledByteBufAllocator createPooledByteBufAllocator(
     }
     return new PooledByteBufAllocator(
       allowDirectBufs && PlatformDependent.directBufferPreferred(),
-      Math.min(getPrivateStaticField("DEFAULT_NUM_HEAP_ARENA"), numCores),
-      Math.min(getPrivateStaticField("DEFAULT_NUM_DIRECT_ARENA"), allowDirectBufs ? numCores : 0),
-      getPrivateStaticField("DEFAULT_PAGE_SIZE"),
-      getPrivateStaticField("DEFAULT_MAX_ORDER"),
-      allowCache ? getPrivateStaticField("DEFAULT_TINY_CACHE_SIZE") : 0,
-      allowCache ? getPrivateStaticField("DEFAULT_SMALL_CACHE_SIZE") : 0,
-      allowCache ? getPrivateStaticField("DEFAULT_NORMAL_CACHE_SIZE") : 0
+      Math.min(PooledByteBufAllocator.defaultNumHeapArena(), numCores),
+      Math.min(PooledByteBufAllocator.defaultNumDirectArena(), allowDirectBufs ? numCores : 0),
+      PooledByteBufAllocator.defaultPageSize(),
+      PooledByteBufAllocator.defaultMaxOrder(),
+      allowCache ? PooledByteBufAllocator.defaultTinyCacheSize() : 0,
+      allowCache ? PooledByteBufAllocator.defaultSmallCacheSize() : 0,
+      allowCache ? PooledByteBufAllocator.defaultNormalCacheSize() : 0,
+      allowCache ? PooledByteBufAllocator.defaultUseCacheForAllThreads() : false
     );
   }
-
-  /** Used to get defaults from Netty's private static fields. */
-  private static int getPrivateStaticField(String name) {
-    try {
-      Field f = PooledByteBufAllocator.DEFAULT.getClass().getDeclaredField(name);
-      f.setAccessible(true);
-      return f.getInt(null);
-    } catch (Exception e) {
-      throw new RuntimeException(e);
-    }
-  }
 }

common/network-common/src/test/java/org/apache/spark/network/ProtocolSuite.java

@@ -116,8 +116,8 @@ public void encode(ChannelHandlerContext ctx, FileRegion in, List<Object> out)
       throws Exception {
 
       ByteArrayWritableChannel channel = new ByteArrayWritableChannel(Ints.checkedCast(in.count()));
-      while (in.transfered() < in.count()) {
-        in.transferTo(channel, in.transfered());
+      while (in.transferred() < in.count()) {
+        in.transferTo(channel, in.transferred());
       }
       out.add(Unpooled.wrappedBuffer(channel.getData()));
     }