[SPARK-13823][SPARK-13397][SPARK-13395][CORE] More warnings, StandardCharset follow up

## What changes were proposed in this pull request?

Follow up to https://github.com/apache/spark/pull/11657

- Also update `String.getBytes("UTF-8")` to use `StandardCharsets.UTF_8`
- And fix one last new Coverity warning that turned up (use of unguarded `wait()` replaced by simpler/more robust `java.util.concurrent` classes in tests)
- And while we're here cleaning up Coverity warnings, just fix about 15 more build warnings

## How was this patch tested?

Jenkins tests

Author: Sean Owen <sowen@cloudera.com>

Closes #11725 from srowen/SPARK-13823.2.

2 files changed, 44 insertions, 63 deletions

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 c0ff9dc5f5..dd0171d1d1 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
@@ -36,6 +36,7 @@ import static org.junit.Assert.*;
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.*;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
 
@@ -93,7 +94,7 @@ public class RequestTimeoutIntegrationSuite {
           ByteBuffer message,
           RpcResponseCallback callback) {
         try {
-          semaphore.tryAcquire(FOREVER, TimeUnit.MILLISECONDS);
+          semaphore.acquire();
           callback.onSuccess(ByteBuffer.allocate(responseSize));
         } catch (InterruptedException e) {
           // do nothing
@@ -113,20 +114,17 @@ public class RequestTimeoutIntegrationSuite {
 
     // First completes quickly (semaphore starts at 1).
     TestCallback callback0 = new TestCallback();
-    synchronized (callback0) {
-      client.sendRpc(ByteBuffer.allocate(0), callback0);
-      callback0.wait(FOREVER);
-      assertEquals(responseSize, callback0.successLength);
-    }
+    client.sendRpc(ByteBuffer.allocate(0), callback0);
+    callback0.latch.await();
+    assertEquals(responseSize, callback0.successLength);
 
     // Second times out after 2 seconds, with slack. Must be IOException.
     TestCallback callback1 = new TestCallback();
-    synchronized (callback1) {
-      client.sendRpc(ByteBuffer.allocate(0), callback1);
-      callback1.wait(4 * 1000);
-      assertNotNull(callback1.failure);
-      assertTrue(callback1.failure instanceof IOException);
-    }
+    client.sendRpc(ByteBuffer.allocate(0), callback1);
+    callback1.latch.await(4, TimeUnit.SECONDS);
+    assertNotNull(callback1.failure);
+    assertTrue(callback1.failure instanceof IOException);
+
     semaphore.release();
   }
 
@@ -143,7 +141,7 @@ public class RequestTimeoutIntegrationSuite {
           ByteBuffer message,
           RpcResponseCallback callback) {
         try {
-          semaphore.tryAcquire(FOREVER, TimeUnit.MILLISECONDS);
+          semaphore.acquire();
           callback.onSuccess(ByteBuffer.allocate(responseSize));
         } catch (InterruptedException e) {
           // do nothing
@@ -164,24 +162,20 @@ public class RequestTimeoutIntegrationSuite {
     TransportClient client0 =
       clientFactory.createClient(TestUtils.getLocalHost(), server.getPort());
     TestCallback callback0 = new TestCallback();
-    synchronized (callback0) {
-      client0.sendRpc(ByteBuffer.allocate(0), callback0);
-      callback0.wait(FOREVER);
-      assertTrue(callback0.failure instanceof IOException);
-      assertFalse(client0.isActive());
-    }
+    client0.sendRpc(ByteBuffer.allocate(0), callback0);
+    callback0.latch.await();
+    assertTrue(callback0.failure instanceof IOException);
+    assertFalse(client0.isActive());
 
     // Increment the semaphore and the second request should succeed quickly.
     semaphore.release(2);
     TransportClient client1 =
       clientFactory.createClient(TestUtils.getLocalHost(), server.getPort());
     TestCallback callback1 = new TestCallback();
-    synchronized (callback1) {
-      client1.sendRpc(ByteBuffer.allocate(0), callback1);
-      callback1.wait(FOREVER);
-      assertEquals(responseSize, callback1.successLength);
-      assertNull(callback1.failure);
-    }
+    client1.sendRpc(ByteBuffer.allocate(0), callback1);
+    callback1.latch.await();
+    assertEquals(responseSize, callback1.successLength);
+    assertNull(callback1.failure);
   }
 
   // The timeout is relative to the LAST request sent, which is kinda weird, but still.
@@ -226,18 +220,14 @@ public class RequestTimeoutIntegrationSuite {
     client.fetchChunk(0, 1, callback1);
     Uninterruptibles.sleepUninterruptibly(1200, TimeUnit.MILLISECONDS);
 
-    synchronized (callback0) {
-      // not complete yet, but should complete soon
-      assertEquals(-1, callback0.successLength);
-      assertNull(callback0.failure);
-      callback0.wait(2 * 1000);
-      assertTrue(callback0.failure instanceof IOException);
-    }
+    // not complete yet, but should complete soon
+    assertEquals(-1, callback0.successLength);
+    assertNull(callback0.failure);
+    callback0.latch.await(2, TimeUnit.SECONDS);
+    assertTrue(callback0.failure instanceof IOException);
 
-    synchronized (callback1) {
-      // failed at same time as previous
-      assertTrue(callback0.failure instanceof IOException);
-    }
+    // failed at same time as previous
+    assertTrue(callback1.failure instanceof IOException);
   }
 
   /**
@@ -248,41 +238,35 @@ public class RequestTimeoutIntegrationSuite {
 
     int successLength = -1;
     Throwable failure;
+    final CountDownLatch latch = new CountDownLatch(1);
 
     @Override
     public void onSuccess(ByteBuffer response) {
-      synchronized(this) {
-        successLength = response.remaining();
-        this.notifyAll();
-      }
+      successLength = response.remaining();
+      latch.countDown();
     }
 
     @Override
     public void onFailure(Throwable e) {
-      synchronized(this) {
-        failure = e;
-        this.notifyAll();
-      }
+      failure = e;
+      latch.countDown();
     }
 
     @Override
     public void onSuccess(int chunkIndex, ManagedBuffer buffer) {
-      synchronized(this) {
-        try {
-          successLength = buffer.nioByteBuffer().remaining();
-          this.notifyAll();
-        } catch (IOException e) {
-          // weird
-        }
+      try {
+        successLength = buffer.nioByteBuffer().remaining();
+      } catch (IOException e) {
+        // weird
+      } finally {
+        latch.countDown();
       }
     }
 
     @Override
     public void onFailure(int chunkIndex, Throwable e) {
-      synchronized(this) {
-        failure = e;
-        this.notifyAll();
-      }
+      failure = e;
+      latch.countDown();
     }
   }
 }
diff --git a/common/network-common/src/test/java/org/apache/spark/network/sasl/SparkSaslSuite.java b/common/network-common/src/test/java/org/apache/spark/network/sasl/SparkSaslSuite.java
index 045773317a..45cc03df43 100644
--- a/common/network-common/src/test/java/org/apache/spark/network/sasl/SparkSaslSuite.java
+++ b/common/network-common/src/test/java/org/apache/spark/network/sasl/SparkSaslSuite.java
@@ -26,6 +26,7 @@ import java.nio.ByteBuffer;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Random;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
@@ -276,7 +277,7 @@ public class SparkSaslSuite {
 
       ctx = new SaslTestCtx(rpcHandler, true, false);
 
-      final Object lock = new Object();
+      final CountDownLatch lock = new CountDownLatch(1);
 
       ChunkReceivedCallback callback = mock(ChunkReceivedCallback.class);
       doAnswer(new Answer<Void>() {
@@ -284,17 +285,13 @@ public class SparkSaslSuite {
           public Void answer(InvocationOnMock invocation) {
             response.set((ManagedBuffer) invocation.getArguments()[1]);
             response.get().retain();
-            synchronized (lock) {
-              lock.notifyAll();
-            }
+            lock.countDown();
             return null;
           }
         }).when(callback).onSuccess(anyInt(), any(ManagedBuffer.class));
 
-      synchronized (lock) {
-        ctx.client.fetchChunk(0, 0, callback);
-        lock.wait(10 * 1000);
-      }
+      ctx.client.fetchChunk(0, 0, callback);
+      lock.await(10, TimeUnit.SECONDS);
 
       verify(callback, times(1)).onSuccess(anyInt(), any(ManagedBuffer.class));
       verify(callback, never()).onFailure(anyInt(), any(Throwable.class));