try new fix

Author: brkyvz

extras/kinesis-asl/pom.xml

@@ -70,12 +70,6 @@
       <version>${aws.kinesis.producer.version}</version>
       <scope>test</scope>
     </dependency>
-    <dependency>
-      <groupId>com.google.protobuf</groupId>
-      <artifactId>protobuf-java</artifactId>
-      <version>2.6.1</version>
-      <scope>test</scope>
-    </dependency>
     <dependency>
       <groupId>org.mockito</groupId>
       <artifactId>mockito-core</artifactId>

extras/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisBackedBlockRDD.scala

@@ -99,7 +99,7 @@ class KinesisBackedBlockRDD[T: ClassTag](
     val blockId = partition.blockId
 
     def getBlockFromBlockManager(): Option[Iterator[T]] = {
-      logInfo(s"Read partition data of $this from block manager, block $blockId")
+      logDebug(s"Read partition data of $this from block manager, block $blockId")
       blockManager.get(blockId).map(_.data.asInstanceOf[Iterator[T]])
     }
 
@@ -213,8 +213,8 @@ class KinesisSequenceRangeIterator(
       }
     // De-aggregate records, if KPL was used in producing the records. The KCL automatically
     // handles de-aggregation during regular operation. This code path is used during recovery
-    // val recordIterator = UserRecord.deaggregate()
-    (getRecordsResult.getRecords.iterator().asScala, getRecordsResult.getNextShardIterator)
+    val recordIterator = UserRecord.deaggregate(getRecordsResult.getRecords)
+    (recordIterator.asScala.iterator, getRecordsResult.getNextShardIterator)
   }
 
   /**

extras/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisInputDStream.scala

@@ -54,7 +54,7 @@ private[kinesis] class KinesisInputDStream[T: ClassTag](
       val seqNumRanges = blockInfos.map {
         _.metadataOption.get.asInstanceOf[SequenceNumberRanges] }.toArray
       val isBlockIdValid = blockInfos.map { _.isBlockIdValid() }.toArray
-      logInfo(s"Creating KinesisBackedBlockRDD for $time with ${seqNumRanges.length} " +
+      logDebug(s"Creating KinesisBackedBlockRDD for $time with ${seqNumRanges.length} " +
           s"seq number ranges: ${seqNumRanges.mkString(", ")} ")
       new KinesisBackedBlockRDD(
         context.sc, regionName, endpointUrl, blockIds, seqNumRanges,

extras/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisReceiver.scala

@@ -206,7 +206,6 @@ private[kinesis] class KinesisReceiver[T](
       val dataIterator = records.iterator().asScala.map(messageHandler)
       val metadata = SequenceNumberRange(streamName, shardId,
         records.get(0).getSequenceNumber(), records.get(records.size() - 1).getSequenceNumber())
-      logInfo(s"blockGenerator.multipleDataWithCallback: iterator: $dataIterator, meta: $metadata")
       blockGenerator.addMultipleDataWithCallback(dataIterator, metadata)
     }
   }
@@ -231,7 +230,7 @@ private[kinesis] class KinesisReceiver[T](
   private def finalizeRangesForCurrentBlock(blockId: StreamBlockId): Unit = {
     blockIdToSeqNumRanges(blockId) = SequenceNumberRanges(seqNumRangesInCurrentBlock.toArray)
     seqNumRangesInCurrentBlock.clear()
-    logInfo(s"Generated block $blockId has $blockIdToSeqNumRanges")
+    logDebug(s"Generated block $blockId has $blockIdToSeqNumRanges")
   }
 
   /** Store the block along with its associated ranges */

extras/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisRecordProcessor.scala

@@ -70,11 +70,10 @@ private[kinesis] class KinesisRecordProcessor[T](
    *   in the DStream
    */
   override def processRecords(batch: List[Record], checkpointer: IRecordProcessorCheckpointer) {
-    logInfo(s"Received batch: $batch. Is receiver alive: ${!receiver.isStopped()}")
     if (!receiver.isStopped()) {
       try {
         receiver.addRecords(shardId, batch)
-        logInfo(s"Stored: Worker $workerId stored ${batch.size} records for shardId $shardId")
+        logDebug(s"Stored: Worker $workerId stored ${batch.size} records for shardId $shardId")
 
         /*
          *

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

@@ -78,8 +78,8 @@ abstract class KinesisStreamTests(aggregateTestData: Boolean) extends KinesisFun
     if (testUtils != null) {
       // Delete the Kinesis stream as well as the DynamoDB table generated by
       // Kinesis Client Library when consuming the stream
-      // testUtils.deleteStream()
-      // testUtils.deleteDynamoDBTable(appName)
+      testUtils.deleteStream()
+      testUtils.deleteDynamoDBTable(appName)
     }
   }