[SPARK-13692][CORE][SQL] Fix trivial Coverity/Checkstyle defects

## What changes were proposed in this pull request?

This issue fixes the following potential bugs and Java coding style detected by Coverity and Checkstyle.

- Implement both null and type checking in equals functions.
- Fix wrong type casting logic in SimpleJavaBean2.equals.
- Add `implement Cloneable` to `UTF8String` and `SortedIterator`.
- Remove dereferencing before null check in `AbstractBytesToBytesMapSuite`.
- Fix coding style: Add '{}' to single `for` statement in mllib examples.
- Remove unused imports in `ColumnarBatch` and `JavaKinesisStreamSuite`.
- Remove unused fields in `ChunkFetchIntegrationSuite`.
- Add `stop()` to prevent resource leak.

Please note that the last two checkstyle errors exist on newly added commits after [SPARK-13583](https://issues.apache.org/jira/browse/SPARK-13583).

## How was this patch tested?

manual via `./dev/lint-java` and Coverity site.

Author: Dongjoon Hyun <[email protected]>

Closes apache#11530 from dongjoon-hyun/SPARK-13692.

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

@@ -64,8 +64,6 @@ public class ChunkFetchIntegrationSuite {
   static ManagedBuffer bufferChunk;
   static ManagedBuffer fileChunk;
 
-  private TransportConf transportConf;
-
   @BeforeClass
   public static void setUp() throws Exception {
     int bufSize = 100000;

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

@@ -55,7 +55,7 @@ public class RequestTimeoutIntegrationSuite {
   private TransportConf conf;
 
   // A large timeout that "shouldn't happen", for the sake of faulty tests not hanging forever.
-  private final int FOREVER = 60 * 1000;
+  private static final int FOREVER = 60 * 1000;
 
   @Before
   public void setUp() throws Exception {

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

@@ -122,6 +122,8 @@ public void run() {
     for (TransportClient client : clients) {
       client.close();
     }
+
+    factory.close();
   }
 
   @Test

common/unsafe/src/main/java/org/apache/spark/unsafe/types/UTF8String.java

@@ -44,7 +44,8 @@
  * <p>
  * Note: This is not designed for general use cases, should not be used outside SQL.
  */
-public final class UTF8String implements Comparable<UTF8String>, Externalizable, KryoSerializable {
+public final class UTF8String implements Comparable<UTF8String>, Externalizable, KryoSerializable,
+  Cloneable {
 
   // These are only updated by readExternal() or read()
   @Nonnull

core/src/main/java/org/apache/spark/util/collection/unsafe/sort/UnsafeInMemorySorter.java

@@ -172,7 +172,7 @@ public void insertRecord(long recordPointer, long keyPrefix) {
     pos++;
   }
 
-  public final class SortedIterator extends UnsafeSorterIterator {
+  public final class SortedIterator extends UnsafeSorterIterator implements Cloneable {
 
     private final int numRecords;
     private int position;

core/src/test/java/org/apache/spark/unsafe/map/AbstractBytesToBytesMapSuite.java

@@ -64,7 +64,7 @@ public abstract class AbstractBytesToBytesMapSuite {
 
   private TestMemoryManager memoryManager;
   private TaskMemoryManager taskMemoryManager;
-  private final long PAGE_SIZE_BYTES = 1L << 26; // 64 megabytes
+  private static final long PAGE_SIZE_BYTES = 1L << 26; // 64 megabytes
 
   final LinkedList<File> spillFilesCreated = new LinkedList<File>();
   File tempDir;
@@ -131,8 +131,8 @@ public void tearDown() {
     Utils.deleteRecursively(tempDir);
     tempDir = null;
 
-    Assert.assertEquals(0L, taskMemoryManager.cleanUpAllAllocatedMemory());
     if (taskMemoryManager != null) {
+      Assert.assertEquals(0L, taskMemoryManager.cleanUpAllAllocatedMemory());
       long leakedMemory = taskMemoryManager.getMemoryConsumptionForThisTask();
       taskMemoryManager = null;
       Assert.assertEquals(0L, leakedMemory);

examples/src/main/java/org/apache/spark/examples/ml/JavaCountVectorizerExample.java

@@ -65,5 +65,7 @@ public static void main(String[] args) {
 
     cvModel.transform(df).show();
     // $example off$
+
+    jsc.stop();
   }
 }

examples/src/main/java/org/apache/spark/examples/ml/JavaDecisionTreeClassificationExample.java

@@ -95,5 +95,7 @@ public static void main(String[] args) {
       (DecisionTreeClassificationModel) (model.stages()[2]);
     System.out.println("Learned classification tree model:\n" + treeModel.toDebugString());
     // $example off$
+
+    jsc.stop();
   }
 }

examples/src/main/java/org/apache/spark/examples/ml/JavaDecisionTreeRegressionExample.java

@@ -83,5 +83,7 @@ public static void main(String[] args) {
       (DecisionTreeRegressionModel) (model.stages()[1]);
     System.out.println("Learned regression tree model:\n" + treeModel.toDebugString());
     // $example off$
+
+    jsc.stop();
   }
 }

examples/src/main/java/org/apache/spark/examples/ml/JavaSQLTransformerExample.java

@@ -55,5 +55,7 @@ public static void main(String[] args) {
 
     sqlTrans.transform(df).show();
     // $example off$
+
+    jsc.stop();
   }
 }

examples/src/main/java/org/apache/spark/examples/ml/JavaWord2VecExample.java

@@ -63,5 +63,7 @@ public static void main(String[] args) {
       System.out.println(r);
     }
     // $example off$
+
+    jsc.stop();
   }
 }

examples/src/main/java/org/apache/spark/examples/mllib/JavaAssociationRulesExample.java

@@ -52,5 +52,7 @@ public static void main(String[] args) {
         rule.javaAntecedent() + " => " + rule.javaConsequent() + ", " + rule.confidence());
     }
     // $example off$
+
+    sc.stop();
   }
 }

examples/src/main/java/org/apache/spark/examples/mllib/JavaGaussianMixtureExample.java

@@ -44,8 +44,9 @@ public static void main(String[] args) {
         public Vector call(String s) {
           String[] sarray = s.trim().split(" ");
           double[] values = new double[sarray.length];
-          for (int i = 0; i < sarray.length; i++)
+          for (int i = 0; i < sarray.length; i++) {
             values[i] = Double.parseDouble(sarray[i]);
+          }
           return Vectors.dense(values);
         }
       }

examples/src/main/java/org/apache/spark/examples/mllib/JavaGradientBoostingClassificationExample.java

@@ -87,6 +87,8 @@ public Boolean call(Tuple2<Double, Double> pl) {
     GradientBoostedTreesModel sameModel = GradientBoostedTreesModel.load(jsc.sc(),
       "target/tmp/myGradientBoostingClassificationModel");
     // $example off$
+
+    jsc.stop();
   }
 
 }

examples/src/main/java/org/apache/spark/examples/mllib/JavaGradientBoostingRegressionExample.java

@@ -92,5 +92,7 @@ public Double call(Double a, Double b) {
     GradientBoostedTreesModel sameModel = GradientBoostedTreesModel.load(jsc.sc(),
       "target/tmp/myGradientBoostingRegressionModel");
     // $example off$
+
+    jsc.stop();
   }
 }

examples/src/main/java/org/apache/spark/examples/mllib/JavaIsotonicRegressionExample.java

@@ -82,5 +82,7 @@ public Object call(Tuple2<Double, Double> pl) {
     model.save(jsc.sc(), "target/tmp/myIsotonicRegressionModel");
     IsotonicRegressionModel sameModel = IsotonicRegressionModel.load(jsc.sc(), "target/tmp/myIsotonicRegressionModel");
     // $example off$
+
+    jsc.stop();
   }
 }

examples/src/main/java/org/apache/spark/examples/mllib/JavaKMeansExample.java

@@ -44,8 +44,9 @@ public static void main(String[] args) {
         public Vector call(String s) {
           String[] sarray = s.split(" ");
           double[] values = new double[sarray.length];
-          for (int i = 0; i < sarray.length; i++)
+          for (int i = 0; i < sarray.length; i++) {
             values[i] = Double.parseDouble(sarray[i]);
+          }
           return Vectors.dense(values);
         }
       }

examples/src/main/java/org/apache/spark/examples/mllib/JavaLatentDirichletAllocationExample.java

@@ -49,8 +49,9 @@ public static void main(String[] args) {
         public Vector call(String s) {
           String[] sarray = s.trim().split(" ");
           double[] values = new double[sarray.length];
-          for (int i = 0; i < sarray.length; i++)
+          for (int i = 0; i < sarray.length; i++) {
             values[i] = Double.parseDouble(sarray[i]);
+          }
           return Vectors.dense(values);
         }
       }

examples/src/main/java/org/apache/spark/examples/mllib/JavaMultiLabelClassificationMetricsExample.java

@@ -74,5 +74,7 @@ public static void main(String[] args) {
     // Subset accuracy
     System.out.format("Subset accuracy = %f\n", metrics.subsetAccuracy());
     // $example off$
+
+    sc.stop();
   }
 }

examples/src/main/java/org/apache/spark/examples/mllib/JavaNaiveBayesExample.java

@@ -60,5 +60,7 @@ public Boolean call(Tuple2<Double, Double> pl) {
     model.save(jsc.sc(), "target/tmp/myNaiveBayesModel");
     NaiveBayesModel sameModel = NaiveBayesModel.load(jsc.sc(), "target/tmp/myNaiveBayesModel");
     // $example off$
+
+    jsc.stop();
   }
 }

examples/src/main/java/org/apache/spark/examples/mllib/JavaPrefixSpanExample.java

@@ -51,5 +51,7 @@ public static void main(String[] args) {
       System.out.println(freqSeq.javaSequence() + ", " + freqSeq.freq());
     }
     // $example off$
+
+    sc.stop();
   }
 }

examples/src/main/java/org/apache/spark/examples/mllib/JavaRandomForestClassificationExample.java

@@ -85,5 +85,7 @@ public Boolean call(Tuple2<Double, Double> pl) {
     RandomForestModel sameModel = RandomForestModel.load(jsc.sc(),
       "target/tmp/myRandomForestClassificationModel");
     // $example off$
+
+    jsc.stop();
   }
 }

examples/src/main/java/org/apache/spark/examples/mllib/JavaRandomForestRegressionExample.java

@@ -91,5 +91,7 @@ public Double call(Double a, Double b) {
     RandomForestModel sameModel = RandomForestModel.load(jsc.sc(),
       "target/tmp/myRandomForestRegressionModel");
     // $example off$
+
+    jsc.stop();
   }
 }

examples/src/main/java/org/apache/spark/examples/mllib/JavaRankingMetricsExample.java

@@ -181,5 +181,7 @@ public Tuple2<Tuple2<Integer, Integer>, Object> call(Rating r) {
     // R-squared
     System.out.format("R-squared = %f\n", regressionMetrics.r2());
     // $example off$
+
+    sc.stop();
   }
 }

examples/src/main/java/org/apache/spark/examples/mllib/JavaRecommendationExample.java

@@ -93,5 +93,7 @@ public Object call(Tuple2<Double, Double> pair) {
     MatrixFactorizationModel sameModel = MatrixFactorizationModel.load(jsc.sc(),
       "target/tmp/myCollaborativeFilter");
     // $example off$
+
+    jsc.stop();
   }
 }

examples/src/main/java/org/apache/spark/examples/mllib/JavaRegressionMetricsExample.java

@@ -88,5 +88,7 @@ public Tuple2<Object, Object> call(LabeledPoint point) {
     LinearRegressionModel sameModel = LinearRegressionModel.load(sc.sc(),
       "target/tmp/LogisticRegressionModel");
     // $example off$
+
+    sc.stop();
   }
 }

examples/src/main/java/org/apache/spark/examples/mllib/JavaSVDExample.java

@@ -40,6 +40,7 @@ public class JavaSVDExample {
   public static void main(String[] args) {
     SparkConf conf = new SparkConf().setAppName("SVD Example");
     SparkContext sc = new SparkContext(conf);
+    JavaSparkContext jsc = JavaSparkContext.fromSparkContext(sc);
 
     // $example on$
     double[][] array = {{1.12, 2.05, 3.12}, {5.56, 6.28, 8.94}, {10.2, 8.0, 20.5}};
@@ -48,7 +49,7 @@ public static void main(String[] args) {
       Vector currentRow = Vectors.dense(array[i]);
       rowsList.add(currentRow);
     }
-    JavaRDD<Vector> rows = JavaSparkContext.fromSparkContext(sc).parallelize(rowsList);
+    JavaRDD<Vector> rows = jsc.parallelize(rowsList);
 
     // Create a RowMatrix from JavaRDD<Vector>.
     RowMatrix mat = new RowMatrix(rows.rdd());
@@ -66,5 +67,7 @@ public static void main(String[] args) {
     }
     System.out.println("Singular values are: " + s);
     System.out.println("V factor is:\n" + V);
+
+    jsc.stop();
   }
 }

examples/src/main/java/org/apache/spark/examples/mllib/JavaSimpleFPGrowth.java

@@ -67,5 +67,7 @@ public List<String> call(String line) {
         rule.javaAntecedent() + " => " + rule.javaConsequent() + ", " + rule.confidence());
     }
     // $example off$
+
+    sc.stop();
   }
 }

extras/kinesis-asl/src/test/java/org/apache/spark/streaming/kinesis/JavaKinesisStreamSuite.java

@@ -28,8 +28,6 @@
 
 import com.amazonaws.services.kinesis.clientlibrary.lib.worker.InitialPositionInStream;
 
-import java.nio.ByteBuffer;
-
 /**
  * Demonstrate the use of the KinesisUtils Java API
  */

sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/ColumnarBatch.java

@@ -22,7 +22,6 @@
 import org.apache.commons.lang.NotImplementedException;
 
 import org.apache.spark.memory.MemoryMode;
-import org.apache.spark.sql.Column;
 import org.apache.spark.sql.catalyst.InternalRow;
 import org.apache.spark.sql.catalyst.expressions.GenericMutableRow;
 import org.apache.spark.sql.catalyst.expressions.UnsafeRow;

sql/core/src/test/java/test/org/apache/spark/sql/JavaDatasetSuite.java

@@ -465,6 +465,9 @@ public static class KryoSerializable {
 
     @Override
     public boolean equals(Object other) {
+      if (this == other) return true;
+      if (other == null || getClass() != other.getClass()) return false;
+
       return this.value.equals(((KryoSerializable) other).value);
     }
 
@@ -483,6 +486,9 @@ public static class JavaSerializable implements Serializable {
 
     @Override
     public boolean equals(Object other) {
+      if (this == other) return true;
+      if (other == null || getClass() != other.getClass()) return false;
+
       return this.value.equals(((JavaSerializable) other).value);
     }
 
@@ -631,7 +637,7 @@ public boolean equals(Object o) {
       if (this == o) return true;
       if (o == null || getClass() != o.getClass()) return false;
 
-      SimpleJavaBean that = (SimpleJavaBean) o;
+      SimpleJavaBean2 that = (SimpleJavaBean2) o;
 
       if (!a.equals(that.a)) return false;
       if (!b.equals(that.b)) return false;