fixed tests and formatting, fixed a bug with JavaKinesisWordCount during

union of streams

Author: cfregly

examples/src/main/java/org/apache/spark/examples/streaming/JavaKinesisWordCountASL.java

@@ -136,7 +136,7 @@ public static void main(String[] args) {
 
         /** Create the same number of Kinesis DStreams/Receivers as Kinesis stream's shards */
         List<JavaDStream<byte[]>> streamsList = new ArrayList<JavaDStream<byte[]>>(numStreams);
-        for (int i = 0; i < streamsList.size(); i++) {
+        for (int i = 0; i < numStreams; i++) {
         	streamsList.add(
                 KinesisUtils.createStream(jssc, streamName, endpointUrl, checkpointInterval, 
                 InitialPositionInStream.LATEST, StorageLevel.MEMORY_AND_DISK_2())

examples/src/main/scala/org/apache/spark/examples/streaming/KinesisWordCountASL.scala

@@ -94,7 +94,8 @@ object KinesisWordCountASL extends Logging {
     /** Determine the number of shards from the stream */
     val kinesisClient = new AmazonKinesisClient(new DefaultAWSCredentialsProviderChain())
     kinesisClient.setEndpoint(endpointUrl)
-    val numShards = kinesisClient.describeStream(streamName).getStreamDescription().getShards().size()
+    val numShards = kinesisClient.describeStream(streamName).getStreamDescription().getShards()
+      .size()
 
     /** In this example, we're going to create 1 Kinesis Worker/Receiver/DStream for each shard. */
     val numStreams = numShards
@@ -108,7 +109,8 @@ object KinesisWordCountASL extends Logging {
     /** Setup the and SparkConfig and StreamingContext */
     /** Spark Streaming batch interval */
     val batchInterval = Milliseconds(2000)    
-    val sparkConfig = new SparkConf().setAppName("KinesisWordCount").setMaster(s"local[$numSparkThreads]")
+    val sparkConfig = new SparkConf().setAppName("KinesisWordCount")
+      .setMaster(s"local[$numSparkThreads]")
     val ssc = new StreamingContext(sparkConfig, batchInterval)
     /** Setup the checkpoint directory used by Spark Streaming */
     ssc.checkpoint("/tmp/checkpoint");

extras/kinesis-asl/src/test/scala/org/apache/spark/streaming/kinesis/KinesisReceiverSuite.scala

@@ -46,16 +46,7 @@ import com.amazonaws.services.kinesis.model.Record
  */
 class KinesisReceiverSuite extends TestSuiteBase with Matchers with BeforeAndAfter
     with EasyMockSugar {
-  
-  test("kinesis input stream") {
-    val ssc = new StreamingContext(master, framework, batchDuration)
-    // Tests the API, does not actually test data receiving
-    val kinesisStream = KinesisUtils.createStream(ssc, "mySparkStream",
-      "https://kinesis.us-west-2.amazonaws.com", Seconds(2),
-      InitialPositionInStream.LATEST, StorageLevel.MEMORY_AND_DISK_2);
-    ssc.stop()
-  }
-  
+
   val app = "TestKinesisReceiver"
   val stream = "mySparkStream"
   val endpoint = "endpoint-url"
@@ -74,14 +65,23 @@ class KinesisReceiverSuite extends TestSuiteBase with Matchers with BeforeAndAft
   var checkpointStateMock: KinesisCheckpointState = _
   var currentClockMock: Clock = _
 
-  before {
+  override def beforeFunction() = {
     receiverMock = mock[KinesisReceiver]
     checkpointerMock = mock[IRecordProcessorCheckpointer]
     checkpointClockMock = mock[ManualClock]
     checkpointStateMock = mock[KinesisCheckpointState]
     currentClockMock = mock[Clock]
   }
 
+  test("kinesis utils api") {
+    val ssc = new StreamingContext(master, framework, batchDuration)
+    // Tests the API, does not actually test data receiving
+    val kinesisStream = KinesisUtils.createStream(ssc, "mySparkStream",
+      "https://kinesis.us-west-2.amazonaws.com", Seconds(2),
+      InitialPositionInStream.LATEST, StorageLevel.MEMORY_AND_DISK_2);
+    ssc.stop()
+  }
+
   test("process records including store and checkpoint") {
     val expectedCheckpointIntervalMillis = 10
     expecting {