Merge branch 'master' into SPARK-18838

Author: Marcelo Vanzin

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

@@ -249,7 +249,7 @@ Index parent() {
      * calculated only once, avoiding redundant work when multiple child indices of the
      * same parent index exist.
      */
-    byte[] childPrefix(Object value) throws Exception {
+    byte[] childPrefix(Object value) {
       Preconditions.checkState(parent == null, "Not a parent index.");
       return buildKey(name, toParentKey(value));
     }
@@ -295,7 +295,7 @@ byte[] end(byte[] prefix) {
     }
 
     /** The key for the end marker for entries with the given value. */
-    byte[] end(byte[] prefix, Object value) throws Exception {
+    byte[] end(byte[] prefix, Object value) {
       checkParent(prefix);
       return (parent != null) ? buildKey(false, prefix, name, toKey(value), END_MARKER)
         : buildKey(name, toKey(value), END_MARKER);
@@ -313,7 +313,7 @@ byte[] entityKey(byte[] prefix, Object entity) throws Exception {
       return entityKey;
     }
 
-    private void updateCount(WriteBatch batch, byte[] key, long delta) throws Exception {
+    private void updateCount(WriteBatch batch, byte[] key, long delta) {
       long updated = getCount(key) + delta;
       if (updated > 0) {
         batch.put(key, db.serializer.serialize(updated));
@@ -431,7 +431,7 @@ void remove(
       addOrRemove(batch, entity, null, null, naturalKey, prefix);
     }
 
-    long getCount(byte[] key) throws Exception {
+    long getCount(byte[] key) {
       byte[] data = db.db().get(key);
       return data != null ? db.serializer.deserializeLong(data) : 0;
     }

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

@@ -81,7 +81,7 @@ class AuthEngine implements Closeable {
    *
    * @return A challenge to be sent the remote side.
    */
-  ClientChallenge challenge() throws GeneralSecurityException, IOException {
+  ClientChallenge challenge() throws GeneralSecurityException {
     this.authNonce = randomBytes(conf.encryptionKeyLength() / Byte.SIZE);
     SecretKeySpec authKey = generateKey(conf.keyFactoryAlgorithm(), conf.keyFactoryIterations(),
       authNonce, conf.encryptionKeyLength());
@@ -105,7 +105,7 @@ ClientChallenge challenge() throws GeneralSecurityException, IOException {
    * @return A response to be sent to the client.
    */
   ServerResponse respond(ClientChallenge clientChallenge)
-    throws GeneralSecurityException, IOException {
+    throws GeneralSecurityException {
 
     SecretKeySpec authKey = generateKey(clientChallenge.kdf, clientChallenge.iterations,
       clientChallenge.nonce, clientChallenge.keyLength);

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

@@ -17,7 +17,6 @@
 
 package org.apache.spark.network.sasl;
 
-import java.io.IOException;
 import java.util.Map;
 import javax.security.auth.callback.Callback;
 import javax.security.auth.callback.CallbackHandler;
@@ -125,7 +124,7 @@ public synchronized void dispose() {
    */
   private class ClientCallbackHandler implements CallbackHandler {
     @Override
-    public void handle(Callback[] callbacks) throws IOException, UnsupportedCallbackException {
+    public void handle(Callback[] callbacks) throws UnsupportedCallbackException {
 
       for (Callback callback : callbacks) {
         if (callback instanceof NameCallback) {

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

@@ -27,7 +27,6 @@
 import javax.security.sasl.Sasl;
 import javax.security.sasl.SaslException;
 import javax.security.sasl.SaslServer;
-import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 import java.util.Map;
 
@@ -155,7 +154,7 @@ public byte[] unwrap(byte[] data, int offset, int len) throws SaslException {
    */
   private class DigestCallbackHandler implements CallbackHandler {
     @Override
-    public void handle(Callback[] callbacks) throws IOException, UnsupportedCallbackException {
+    public void handle(Callback[] callbacks) throws UnsupportedCallbackException {
       for (Callback callback : callbacks) {
         if (callback instanceof NameCallback) {
           logger.trace("SASL server callback: setting username");

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

@@ -117,7 +117,7 @@ private void init(String hostToBind, int portToBind) {
 
     bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
       @Override
-      protected void initChannel(SocketChannel ch) throws Exception {
+      protected void initChannel(SocketChannel ch) {
         RpcHandler rpcHandler = appRpcHandler;
         for (TransportServerBootstrap bootstrap : bootstraps) {
           rpcHandler = bootstrap.doBootstrap(ch, rpcHandler);

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

@@ -121,7 +121,7 @@ private long decodeFrameSize() {
     return nextFrameSize;
   }
 
-  private ByteBuf decodeNext() throws Exception {
+  private ByteBuf decodeNext() {
     long frameSize = decodeFrameSize();
     if (frameSize == UNKNOWN_FRAME_SIZE || totalSize < frameSize) {
       return null;

common/sketch/src/main/java/org/apache/spark/util/sketch/CountMinSketchImpl.java

@@ -365,7 +365,7 @@ private void writeObject(ObjectOutputStream out) throws IOException {
     this.writeTo(out);
   }
 
-  private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
+  private void readObject(ObjectInputStream in) throws IOException {
     this.readFrom0(in);
   }
 }

core/src/main/java/org/apache/spark/memory/MemoryConsumer.java

@@ -111,8 +111,6 @@ public void freeArray(LongArray array) {
   /**
    * Allocate a memory block with at least `required` bytes.
    *
-   * Throws IOException if there is not enough memory.
-   *
    * @throws OutOfMemoryError
    */
   protected MemoryBlock allocatePage(long required) {

core/src/main/java/org/apache/spark/shuffle/sort/ShuffleExternalSorter.java

@@ -140,7 +140,7 @@ final class ShuffleExternalSorter extends MemoryConsumer {
    *                   bytes written should be counted towards shuffle spill metrics rather than
    *                   shuffle write metrics.
    */
-  private void writeSortedFile(boolean isLastFile) throws IOException {
+  private void writeSortedFile(boolean isLastFile) {
 
     final ShuffleWriteMetrics writeMetricsToUse;
 
@@ -325,7 +325,7 @@ public void cleanupResources() {
    * array and grows the array if additional space is required. If the required space cannot be
    * obtained, then the in-memory data will be spilled to disk.
    */
-  private void growPointerArrayIfNecessary() throws IOException {
+  private void growPointerArrayIfNecessary() {
     assert(inMemSorter != null);
     if (!inMemSorter.hasSpaceForAnotherRecord()) {
       long used = inMemSorter.getMemoryUsage();
@@ -406,19 +406,14 @@ public void insertRecord(Object recordBase, long recordOffset, int length, int p
    * @throws IOException
    */
   public SpillInfo[] closeAndGetSpills() throws IOException {
-    try {
-      if (inMemSorter != null) {
-        // Do not count the final file towards the spill count.
-        writeSortedFile(true);
-        freeMemory();
-        inMemSorter.free();
-        inMemSorter = null;
-      }
-      return spills.toArray(new SpillInfo[spills.size()]);
-    } catch (IOException e) {
-      cleanupResources();
-      throw e;
+    if (inMemSorter != null) {
+      // Do not count the final file towards the spill count.
+      writeSortedFile(true);
+      freeMemory();
+      inMemSorter.free();
+      inMemSorter = null;
     }
+    return spills.toArray(new SpillInfo[spills.size()]);
   }
 
 }

core/src/main/java/org/apache/spark/shuffle/sort/UnsafeShuffleWriter.java

@@ -208,7 +208,7 @@ public void write(scala.collection.Iterator<Product2<K, V>> records) throws IOEx
     }
   }
 
-  private void open() throws IOException {
+  private void open() {
     assert (sorter == null);
     sorter = new ShuffleExternalSorter(
       memoryManager,