[SPARK-13423][WIP][CORE][SQL][STREAMING] Static analysis fixes for 2.x

## What changes were proposed in this pull request?

Make some cross-cutting code improvements according to static analysis. These are individually up for discussion since they exist in separate commits that can be reverted. The changes are broadly:

- Inner class should be static
- Mismatched hashCode/equals
- Overflow in compareTo
- Unchecked warnings
- Misuse of assert, vs junit.assert
- get(a) + getOrElse(b) -> getOrElse(a,b)
- Array/String .size -> .length (occasionally, -> .isEmpty / .nonEmpty) to avoid implicit conversions
- Dead code
- tailrec
- exists(_ == ) -> contains find + nonEmpty -> exists filter + size -> count
- reduce(_+_) -> sum map + flatten -> map

The most controversial may be .size -> .length simply because of its size. It is intended to avoid implicits that might be expensive in some places.

## How was the this patch tested?

Existing Jenkins unit tests.

Author: Sean Owen <sowen@cloudera.com>

Closes #11292 from srowen/SPARK-13423.

9 files changed, 26 insertions, 17 deletions

diff --git a/common/network-common/src/test/java/org/apache/spark/network/ChunkFetchIntegrationSuite.java b/common/network-common/src/test/java/org/apache/spark/network/ChunkFetchIntegrationSuite.java
index 70c849d60e..d17e986e17 100644
--- a/common/network-common/src/test/java/org/apache/spark/network/ChunkFetchIntegrationSuite.java
+++ b/common/network-common/src/test/java/org/apache/spark/network/ChunkFetchIntegrationSuite.java
@@ -132,7 +132,7 @@ public class ChunkFetchIntegrationSuite {
     testFile.delete();
   }
 
-  class FetchResult {
+  static class FetchResult {
     public Set<Integer> successChunks;
     public Set<Integer> failedChunks;
     public List<ManagedBuffer> buffers;
diff --git a/common/network-common/src/test/java/org/apache/spark/network/RequestTimeoutIntegrationSuite.java b/common/network-common/src/test/java/org/apache/spark/network/RequestTimeoutIntegrationSuite.java
index f9b5bf96d6..e2d026c66f 100644
--- a/common/network-common/src/test/java/org/apache/spark/network/RequestTimeoutIntegrationSuite.java
+++ b/common/network-common/src/test/java/org/apache/spark/network/RequestTimeoutIntegrationSuite.java
@@ -124,8 +124,8 @@ public class RequestTimeoutIntegrationSuite {
     synchronized (callback1) {
       client.sendRpc(ByteBuffer.allocate(0), callback1);
       callback1.wait(4 * 1000);
-      assert (callback1.failure != null);
-      assert (callback1.failure instanceof IOException);
+      assertNotNull(callback1.failure);
+      assertTrue(callback1.failure instanceof IOException);
     }
     semaphore.release();
   }
@@ -167,8 +167,8 @@ public class RequestTimeoutIntegrationSuite {
     synchronized (callback0) {
       client0.sendRpc(ByteBuffer.allocate(0), callback0);
       callback0.wait(FOREVER);
-      assert (callback0.failure instanceof IOException);
-      assert (!client0.isActive());
+      assertTrue(callback0.failure instanceof IOException);
+      assertFalse(client0.isActive());
     }
 
     // Increment the semaphore and the second request should succeed quickly.
@@ -236,7 +236,7 @@ public class RequestTimeoutIntegrationSuite {
 
     synchronized (callback1) {
       // failed at same time as previous
-      assert (callback0.failure instanceof IOException);
+      assertTrue(callback0.failure instanceof IOException);
     }
   }
 
@@ -244,7 +244,7 @@ public class RequestTimeoutIntegrationSuite {
    * Callback which sets 'success' or 'failure' on completion.
    * Additionally notifies all waiters on this callback when invoked.
    */
-  class TestCallback implements RpcResponseCallback, ChunkReceivedCallback {
+  static class TestCallback implements RpcResponseCallback, ChunkReceivedCallback {
 
     int successLength = -1;
     Throwable failure;
diff --git a/common/network-common/src/test/java/org/apache/spark/network/RpcIntegrationSuite.java b/common/network-common/src/test/java/org/apache/spark/network/RpcIntegrationSuite.java
index 9e9be98c14..a7a99f3bfc 100644
--- a/common/network-common/src/test/java/org/apache/spark/network/RpcIntegrationSuite.java
+++ b/common/network-common/src/test/java/org/apache/spark/network/RpcIntegrationSuite.java
@@ -91,7 +91,7 @@ public class RpcIntegrationSuite {
     clientFactory.close();
   }
 
-  class RpcResult {
+  static class RpcResult {
     public Set<String> successMessages;
     public Set<String> errorMessages;
   }
diff --git a/common/network-common/src/test/java/org/apache/spark/network/TransportClientFactorySuite.java b/common/network-common/src/test/java/org/apache/spark/network/TransportClientFactorySuite.java
index dac7d4a5b0..9a89dd114f 100644
--- a/common/network-common/src/test/java/org/apache/spark/network/TransportClientFactorySuite.java
+++ b/common/network-common/src/test/java/org/apache/spark/network/TransportClientFactorySuite.java
@@ -27,6 +27,7 @@ import java.util.concurrent.atomic.AtomicInteger;
 
 import com.google.common.collect.Maps;
 import org.junit.After;
+import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 
@@ -95,7 +96,7 @@ public class TransportClientFactorySuite {
           try {
             TransportClient client =
               factory.createClient(TestUtils.getLocalHost(), server1.getPort());
-            assert (client.isActive());
+            assertTrue(client.isActive());
             clients.add(client);
           } catch (IOException e) {
             failed.incrementAndGet();
@@ -115,8 +116,8 @@ public class TransportClientFactorySuite {
       attempts[i].join();
     }
 
-    assert(failed.get() == 0);
-    assert(clients.size() == maxConnections);
+    Assert.assertEquals(0, failed.get());
+    Assert.assertEquals(clients.size(), maxConnections);
 
     for (TransportClient client : clients) {
       client.close();
diff --git a/common/network-common/src/test/java/org/apache/spark/network/protocol/MessageWithHeaderSuite.java b/common/network-common/src/test/java/org/apache/spark/network/protocol/MessageWithHeaderSuite.java
index fbbe4b7014..b341c5681e 100644
--- a/common/network-common/src/test/java/org/apache/spark/network/protocol/MessageWithHeaderSuite.java
+++ b/common/network-common/src/test/java/org/apache/spark/network/protocol/MessageWithHeaderSuite.java
@@ -65,7 +65,7 @@ public class MessageWithHeaderSuite {
     assertEquals(42, result.readLong());
     assertEquals(84, result.readLong());
 
-    assert(msg.release());
+    assertTrue(msg.release());
     assertEquals(0, bodyPassedToNettyManagedBuffer.refCnt());
     assertEquals(0, header.refCnt());
   }
@@ -77,7 +77,7 @@ public class MessageWithHeaderSuite {
     ByteBuf body = (ByteBuf) managedBuf.convertToNetty();
     assertEquals(2, body.refCnt());
     MessageWithHeader msg = new MessageWithHeader(managedBuf, header, body, body.readableBytes());
-    assert(msg.release());
+    assertTrue(msg.release());
     Mockito.verify(managedBuf, Mockito.times(1)).release();
     assertEquals(0, body.refCnt());
   }
@@ -94,7 +94,7 @@ public class MessageWithHeaderSuite {
     for (long i = 0; i < 8; i++) {
       assertEquals(i, result.readLong());
     }
-    assert(msg.release());
+    assertTrue(msg.release());
   }
 
   private ByteBuf doWrite(MessageWithHeader msg, int minExpectedWrites) throws Exception {
diff --git a/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/protocol/mesos/RegisterDriver.java b/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/protocol/mesos/RegisterDriver.java
index 94a61d6caa..eeb0019411 100644
--- a/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/protocol/mesos/RegisterDriver.java
+++ b/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/protocol/mesos/RegisterDriver.java
@@ -56,6 +56,14 @@ public class RegisterDriver extends BlockTransferMessage {
     return Objects.hashCode(appId);
   }
 
+  @Override
+  public boolean equals(Object o) {
+    if (!(o instanceof RegisterDriver)) {
+      return false;
+    }
+    return Objects.equal(appId, ((RegisterDriver) o).appId);
+  }
+
   public static RegisterDriver decode(ByteBuf buf) {
     String appId = Encoders.Strings.decode(buf);
     return new RegisterDriver(appId);
diff --git a/common/network-shuffle/src/test/java/org/apache/spark/network/shuffle/ExternalShuffleIntegrationSuite.java b/common/network-shuffle/src/test/java/org/apache/spark/network/shuffle/ExternalShuffleIntegrationSuite.java
index 5e706bf401..ecbbe7bfa3 100644
--- a/common/network-shuffle/src/test/java/org/apache/spark/network/shuffle/ExternalShuffleIntegrationSuite.java
+++ b/common/network-shuffle/src/test/java/org/apache/spark/network/shuffle/ExternalShuffleIntegrationSuite.java
@@ -109,7 +109,7 @@ public class ExternalShuffleIntegrationSuite {
     handler.applicationRemoved(APP_ID, false /* cleanupLocalDirs */);
   }
 
-  class FetchResult {
+  static class FetchResult {
     public Set<String> successBlocks;
     public Set<String> failedBlocks;
     public List<ManagedBuffer> buffers;
diff --git a/common/network-shuffle/src/test/java/org/apache/spark/network/shuffle/RetryingBlockFetcherSuite.java b/common/network-shuffle/src/test/java/org/apache/spark/network/shuffle/RetryingBlockFetcherSuite.java
index 3a6ef0d3f8..91882e3b3b 100644
--- a/common/network-shuffle/src/test/java/org/apache/spark/network/shuffle/RetryingBlockFetcherSuite.java
+++ b/common/network-shuffle/src/test/java/org/apache/spark/network/shuffle/RetryingBlockFetcherSuite.java
@@ -305,7 +305,7 @@ public class RetryingBlockFetcherSuite {
       }
     }
 
-    assert stub != null;
+    assertNotNull(stub);
     stub.when(fetchStarter).createAndStart((String[]) any(), (BlockFetchingListener) anyObject());
     String[] blockIdArray = blockIds.toArray(new String[blockIds.size()]);
     new RetryingBlockFetcher(conf, fetchStarter, blockIdArray, listener).start();
diff --git a/common/unsafe/src/test/scala/org/apache/spark/unsafe/types/UTF8StringPropertyCheckSuite.scala b/common/unsafe/src/test/scala/org/apache/spark/unsafe/types/UTF8StringPropertyCheckSuite.scala
index b3bbd68827..8a6b9e3e45 100644
--- a/common/unsafe/src/test/scala/org/apache/spark/unsafe/types/UTF8StringPropertyCheckSuite.scala
+++ b/common/unsafe/src/test/scala/org/apache/spark/unsafe/types/UTF8StringPropertyCheckSuite.scala
@@ -193,7 +193,7 @@ class UTF8StringPropertyCheckSuite extends FunSuite with GeneratorDrivenProperty
 
   test("concat") {
     def concat(orgin: Seq[String]): String =
-      if (orgin.exists(_ == null)) null else orgin.mkString
+      if (orgin.contains(null)) null else orgin.mkString
 
     forAll { (inputs: Seq[String]) =>
       assert(UTF8String.concat(inputs.map(toUTF8): _*) === toUTF8(inputs.mkString))