Revert Kinesis changes

Author: srowen

external/kinesis-asl/src/main/java/org/apache/spark/examples/streaming/JavaKinesisWordCountASL.java

@@ -14,31 +14,31 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-
 package org.apache.spark.examples.streaming;
 
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Iterator;
 import java.util.List;
 import java.util.regex.Pattern;
 
-import scala.reflect.ClassTag$;
-import scala.Tuple2;
-
-import com.amazonaws.auth.DefaultAWSCredentialsProviderChain;
-import com.amazonaws.client.builder.AwsClientBuilder;
-import com.amazonaws.services.kinesis.AmazonKinesis;
-import com.amazonaws.services.kinesis.AmazonKinesisClientBuilder;
-import org.apache.spark.streaming.kinesis.KinesisInitialPositions;
-
 import org.apache.spark.SparkConf;
+import org.apache.spark.api.java.function.FlatMapFunction;
+import org.apache.spark.api.java.function.Function2;
+import org.apache.spark.api.java.function.PairFunction;
 import org.apache.spark.storage.StorageLevel;
 import org.apache.spark.streaming.Duration;
 import org.apache.spark.streaming.api.java.JavaDStream;
 import org.apache.spark.streaming.api.java.JavaPairDStream;
 import org.apache.spark.streaming.api.java.JavaStreamingContext;
-import org.apache.spark.streaming.kinesis.KinesisInputDStream;
+import org.apache.spark.streaming.kinesis.KinesisUtils;
+
+import scala.Tuple2;
+
+import com.amazonaws.auth.DefaultAWSCredentialsProviderChain;
+import com.amazonaws.services.kinesis.AmazonKinesisClient;
+import com.amazonaws.services.kinesis.clientlibrary.lib.worker.InitialPositionInStream;
 
 /**
  * Consumes messages from a Amazon Kinesis streams and does wordcount.
@@ -105,22 +105,25 @@ public static void main(String[] args) throws Exception {
     String endpointUrl = args[2];
 
     // Create a Kinesis client in order to determine the number of shards for the given stream
-    AmazonKinesis kinesisClient = AmazonKinesisClientBuilder.standard()
-        .withCredentials(new DefaultAWSCredentialsProviderChain())
-        .withEndpointConfiguration(
-            new AwsClientBuilder.EndpointConfiguration(endpointUrl, "us-east-1"))
-        .build();
+    AmazonKinesisClient kinesisClient =
+        new AmazonKinesisClient(new DefaultAWSCredentialsProviderChain());
+    kinesisClient.setEndpoint(endpointUrl);
+    int numShards =
+        kinesisClient.describeStream(streamName).getStreamDescription().getShards().size();
+
 
     // In this example, we're going to create 1 Kinesis Receiver/input DStream for each shard.
     // This is not a necessity; if there are less receivers/DStreams than the number of shards,
     // then the shards will be automatically distributed among the receivers and each receiver
     // will receive data from multiple shards.
-    int numStreams =
-        kinesisClient.describeStream(streamName).getStreamDescription().getShards().size();
+    int numStreams = numShards;
 
-    // Spark Streaming batch interval. Same as Kinesis checkpoint interval.
+    // Spark Streaming batch interval
     Duration batchInterval = new Duration(2000);
 
+    // Kinesis checkpoint interval.  Same as batchInterval for this example.
+    Duration kinesisCheckpointInterval = batchInterval;
+
     // Get the region name from the endpoint URL to save Kinesis Client Library metadata in
     // DynamoDB of the same region as the Kinesis stream
     String regionName = KinesisExampleUtils.getRegionNameByEndpoint(endpointUrl);
@@ -133,35 +136,46 @@ public static void main(String[] args) throws Exception {
     List<JavaDStream<byte[]>> streamsList = new ArrayList<>(numStreams);
     for (int i = 0; i < numStreams; i++) {
       streamsList.add(
-          JavaDStream.fromDStream(KinesisInputDStream.builder()
-              .streamingContext(jssc)
-              .checkpointAppName(kinesisAppName).streamName(streamName)
-              .endpointUrl(endpointUrl).regionName(regionName)
-              .initialPosition(new KinesisInitialPositions.Latest())
-              .checkpointInterval(batchInterval).storageLevel(StorageLevel.MEMORY_AND_DISK_2())
-              .build(), ClassTag$.MODULE$.apply(byte[].class)));
+          KinesisUtils.createStream(jssc, kinesisAppName, streamName, endpointUrl, regionName,
+              InitialPositionInStream.LATEST, kinesisCheckpointInterval,
+              StorageLevel.MEMORY_AND_DISK_2())
+      );
     }
 
     // Union all the streams if there is more than 1 stream
     JavaDStream<byte[]> unionStreams;
     if (streamsList.size() > 1) {
-      @SuppressWarnings("unchecked")
-      JavaDStream<byte[]>[] array = (JavaDStream<byte[]>[]) streamsList.toArray();
-      unionStreams = jssc.union(array);
+      unionStreams = jssc.union(streamsList.toArray(new JavaDStream[0]));
     } else {
       // Otherwise, just use the 1 stream
       unionStreams = streamsList.get(0);
     }
 
     // Convert each line of Array[Byte] to String, and split into words
-    JavaDStream<String> words = unionStreams.flatMap(line -> {
-      String s = new String(line, StandardCharsets.UTF_8);
-      return Arrays.asList(WORD_SEPARATOR.split(s)).iterator();
+    JavaDStream<String> words = unionStreams.flatMap(new FlatMapFunction<byte[], String>() {
+      @Override
+      public Iterator<String> call(byte[] line) {
+        String s = new String(line, StandardCharsets.UTF_8);
+        return Arrays.asList(WORD_SEPARATOR.split(s)).iterator();
+      }
     });
 
     // Map each word to a (word, 1) tuple so we can reduce by key to count the words
-    JavaPairDStream<String, Integer> wordCounts =
-        words.mapToPair(s -> new Tuple2<>(s, 1)).reduceByKey((i1, i2) -> i1 + i2);
+    JavaPairDStream<String, Integer> wordCounts = words.mapToPair(
+        new PairFunction<String, String, Integer>() {
+          @Override
+          public Tuple2<String, Integer> call(String s) {
+            return new Tuple2<>(s, 1);
+          }
+        }
+    ).reduceByKey(
+        new Function2<Integer, Integer, Integer>() {
+          @Override
+          public Integer call(Integer i1, Integer i2) {
+            return i1 + i2;
+          }
+        }
+    );
 
     // Print the first 10 wordCounts
     wordCounts.print();

external/kinesis-asl/src/main/scala/org/apache/spark/examples/streaming/KinesisWordCountASL.scala

@@ -23,8 +23,7 @@ import java.nio.ByteBuffer
 import scala.util.Random
 
 import com.amazonaws.auth.DefaultAWSCredentialsProviderChain
-import com.amazonaws.client.builder.AwsClientBuilder.EndpointConfiguration
-import com.amazonaws.services.kinesis.AmazonKinesisClientBuilder
+import com.amazonaws.services.kinesis.AmazonKinesisClient
 import com.amazonaws.services.kinesis.model.PutRecordRequest
 import org.apache.log4j.{Level, Logger}
 
@@ -101,14 +100,12 @@ object KinesisWordCountASL extends Logging {
 
     // Determine the number of shards from the stream using the low-level Kinesis Client
     // from the AWS Java SDK.
-    val provider = new DefaultAWSCredentialsProviderChain()
-    require(provider.getCredentials != null,
+    val credentials = new DefaultAWSCredentialsProviderChain().getCredentials()
+    require(credentials != null,
       "No AWS credentials found. Please specify credentials using one of the methods specified " +
         "in http://docs.aws.amazon.com/AWSSdkDocsJava/latest/DeveloperGuide/credentials.html")
-    val kinesisClient = AmazonKinesisClientBuilder.standard()
-      .withCredentials(provider)
-      .withEndpointConfiguration(new EndpointConfiguration(endpointUrl, "us-east-1"))
-      .build()
+    val kinesisClient = new AmazonKinesisClient(credentials)
+    kinesisClient.setEndpoint(endpointUrl)
     val numShards = kinesisClient.describeStream(streamName).getStreamDescription().getShards().size
 
 
@@ -223,10 +220,8 @@ object KinesisWordProducerASL {
     val totals = scala.collection.mutable.Map[String, Int]()
 
     // Create the low-level Kinesis Client from the AWS Java SDK.
-    val kinesisClient = AmazonKinesisClientBuilder.standard()
-      .withCredentials(new DefaultAWSCredentialsProviderChain())
-      .withEndpointConfiguration(new EndpointConfiguration(endpoint, "us-east-1"))
-      .build()
+    val kinesisClient = new AmazonKinesisClient(new DefaultAWSCredentialsProviderChain())
+    kinesisClient.setEndpoint(endpoint)
 
     println(s"Putting records onto stream $stream and endpoint $endpoint at a rate of" +
         s" $recordsPerSecond records per second and $wordsPerRecord words per record")

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

@@ -21,9 +21,8 @@ import scala.collection.JavaConverters._
 import scala.reflect.ClassTag
 import scala.util.control.NonFatal
 
-import com.amazonaws.auth.AWSCredentialsProvider
-import com.amazonaws.client.builder.AwsClientBuilder.EndpointConfiguration
-import com.amazonaws.services.kinesis.AmazonKinesisClientBuilder
+import com.amazonaws.auth.AWSCredentials
+import com.amazonaws.services.kinesis.AmazonKinesisClient
 import com.amazonaws.services.kinesis.clientlibrary.types.UserRecord
 import com.amazonaws.services.kinesis.model._
 
@@ -110,8 +109,9 @@ class KinesisBackedBlockRDD[T: ClassTag](
     }
 
     def getBlockFromKinesis(): Iterator[T] = {
+      val credentials = kinesisCreds.provider.getCredentials
       partition.seqNumberRanges.ranges.iterator.flatMap { range =>
-        new KinesisSequenceRangeIterator(kinesisCreds.provider, endpointUrl, regionName,
+        new KinesisSequenceRangeIterator(credentials, endpointUrl, regionName,
           range, kinesisReadConfigs).map(messageHandler)
       }
     }
@@ -131,17 +131,13 @@ class KinesisBackedBlockRDD[T: ClassTag](
  */
 private[kinesis]
 class KinesisSequenceRangeIterator(
-    credentialsProvider: AWSCredentialsProvider,
+    credentials: AWSCredentials,
     endpointUrl: String,
     regionId: String,
     range: SequenceNumberRange,
     kinesisReadConfigs: KinesisReadConfigurations) extends NextIterator[Record] with Logging {
 
-  private val client = AmazonKinesisClientBuilder.standard().
-    withCredentials(credentialsProvider).
-    withEndpointConfiguration(new EndpointConfiguration(endpointUrl, regionId)).
-    build()
-
+  private val client = new AmazonKinesisClient(credentials)
   private val streamName = range.streamName
   private val shardId = range.shardId
   // AWS limits to maximum of 10k records per get call
@@ -151,6 +147,8 @@ class KinesisSequenceRangeIterator(
   private var lastSeqNumber: String = null
   private var internalIterator: Iterator[Record] = null
 
+  client.setEndpoint(endpointUrl)
+
   override protected def getNext(): Record = {
     var nextRecord: Record = null
     if (toSeqNumberReceived) {
@@ -218,7 +216,7 @@ class KinesisSequenceRangeIterator(
       shardIterator: String,
       recordCount: Int): (Iterator[Record], String) = {
     val getRecordsRequest = new GetRecordsRequest
-    getRecordsRequest.setRequestCredentialsProvider(credentialsProvider)
+    getRecordsRequest.setRequestCredentials(credentials)
     getRecordsRequest.setShardIterator(shardIterator)
     getRecordsRequest.setLimit(Math.min(recordCount, this.maxGetRecordsLimit))
     val getRecordsResult = retryOrTimeout[GetRecordsResult](
@@ -239,7 +237,7 @@ class KinesisSequenceRangeIterator(
       iteratorType: ShardIteratorType,
       sequenceNumber: String): String = {
     val getShardIteratorRequest = new GetShardIteratorRequest
-    getShardIteratorRequest.setRequestCredentialsProvider(credentialsProvider)
+    getShardIteratorRequest.setRequestCredentials(credentials)
     getShardIteratorRequest.setStreamName(streamName)
     getShardIteratorRequest.setShardId(shardId)
     getShardIteratorRequest.setShardIteratorType(iteratorType.toString)

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

@@ -47,6 +47,8 @@ private[kinesis] class KinesisInputDStream[T: ClassTag](
     val cloudWatchCreds: Option[SparkAWSCredentials]
   ) extends ReceiverInputDStream[T](_ssc) {
 
+  import KinesisReadConfigurations._
+
   private[streaming]
   override def createBlockRDD(time: Time, blockInfos: Seq[ReceivedBlockInfo]): RDD[T] = {
 
@@ -235,7 +237,7 @@ object KinesisInputDStream {
 
     /**
      * Sets the [[SparkAWSCredentials]] to use for authenticating to the AWS Kinesis
-     * endpoint. Defaults to `DefaultCredentialsProvider` if no custom value is specified.
+     * endpoint. Defaults to [[DefaultCredentialsProvider]] if no custom value is specified.
      *
      * @param credentials [[SparkAWSCredentials]] to use for Kinesis authentication
      */

external/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisTestUtils.scala

@@ -26,11 +26,11 @@ import scala.collection.mutable
 import scala.collection.mutable.ArrayBuffer
 import scala.util.{Failure, Random, Success, Try}
 
-import com.amazonaws.auth.{AWSCredentialsProvider, DefaultAWSCredentialsProviderChain}
-import com.amazonaws.client.builder.AwsClientBuilder.EndpointConfiguration
+import com.amazonaws.auth.{AWSCredentials, DefaultAWSCredentialsProviderChain}
 import com.amazonaws.regions.RegionUtils
-import com.amazonaws.services.dynamodbv2.AmazonDynamoDBClientBuilder
-import com.amazonaws.services.kinesis.{AmazonKinesis, AmazonKinesisClientBuilder}
+import com.amazonaws.services.dynamodbv2.AmazonDynamoDBClient
+import com.amazonaws.services.dynamodbv2.document.DynamoDB
+import com.amazonaws.services.kinesis.{AmazonKinesis, AmazonKinesisClient}
 import com.amazonaws.services.kinesis.model._
 
 import org.apache.spark.internal.Logging
@@ -55,18 +55,15 @@ private[kinesis] class KinesisTestUtils(streamShardCount: Int = 2) extends Loggi
   private var _streamName: String = _
 
   protected lazy val kinesisClient = {
-    AmazonKinesisClientBuilder.standard().
-      withCredentials(KinesisTestUtils.getAWSCredentials()).
-      withEndpointConfiguration(new EndpointConfiguration(
-        endpointUrl, KinesisTestUtils.getRegionNameByEndpoint(endpointUrl))).
-      build()
+    val client = new AmazonKinesisClient(KinesisTestUtils.getAWSCredentials())
+    client.setEndpoint(endpointUrl)
+    client
   }
 
   private lazy val dynamoDB = {
-    AmazonDynamoDBClientBuilder.standard().
-      withCredentials(new DefaultAWSCredentialsProviderChain()).
-      withRegion(regionName).
-      build()
+    val dynamoDBClient = new AmazonDynamoDBClient(new DefaultAWSCredentialsProviderChain())
+    dynamoDBClient.setRegion(RegionUtils.getRegion(regionName))
+    new DynamoDB(dynamoDBClient)
   }
 
   protected def getProducer(aggregate: Boolean): KinesisDataGenerator = {
@@ -133,7 +130,7 @@ private[kinesis] class KinesisTestUtils(streamShardCount: Int = 2) extends Loggi
     val producer = getProducer(aggregate)
     val shardIdToSeqNumbers = producer.sendData(streamName, testData)
     logInfo(s"Pushed $testData:\n\t ${shardIdToSeqNumbers.mkString("\n\t")}")
-    shardIdToSeqNumbers
+    shardIdToSeqNumbers.toMap
   }
 
   /**
@@ -156,7 +153,9 @@ private[kinesis] class KinesisTestUtils(streamShardCount: Int = 2) extends Loggi
 
   def deleteDynamoDBTable(tableName: String): Unit = {
     try {
-      dynamoDB.deleteTable(tableName)
+      val table = dynamoDB.getTable(tableName)
+      table.delete()
+      table.waitForDelete()
     } catch {
       case e: Exception =>
         logWarning(s"Could not delete DynamoDB table $tableName")
@@ -249,13 +248,12 @@ private[kinesis] object KinesisTestUtils {
     Try { new DefaultAWSCredentialsProviderChain().getCredentials() }.isSuccess
   }
 
-  def getAWSCredentials(): AWSCredentialsProvider = {
+  def getAWSCredentials(): AWSCredentials = {
     assert(shouldRunTests,
       "Kinesis test not enabled, should not attempt to get AWS credentials")
-    val provider = new DefaultAWSCredentialsProviderChain()
-    Try { provider.getCredentials() } match {
-      case Success(_) => provider
-      case Failure(_) =>
+    Try { new DefaultAWSCredentialsProviderChain().getCredentials() } match {
+      case Success(cred) => cred
+      case Failure(e) =>
         throw new Exception(
           s"""
              |Kinesis tests enabled using environment variable $envVarNameForEnablingTests
@@ -274,7 +272,7 @@ private[kinesis] trait KinesisDataGenerator {
 }
 
 private[kinesis] class SimpleDataGenerator(
-    client: AmazonKinesis) extends KinesisDataGenerator {
+    client: AmazonKinesisClient) extends KinesisDataGenerator {
   override def sendData(streamName: String, data: Seq[Int]): Map[String, Seq[(Int, String)]] = {
     val shardIdToSeqNumbers = new mutable.HashMap[String, ArrayBuffer[(Int, String)]]()
     data.foreach { num =>