[SPARK-53927][BUILD][DSTREAM] Upgrade kinesis client and fix kinesis integration tests

### What changes were proposed in this pull request?
Upgrade kinesis client and AWS Java SDK to fix Kinesis integration tests.

Kinesis client is upgraded from `1.12.0` to `1.15.3` (latest on 1.x)
AWS Java SDK is upgraded from `1.11.655` to `1.12.681` (the one used by Kinesis client)
AWS Kinesis producer library (used in test) upgraded from `0.12.8` to `1.0.5`

### Why are the changes needed?
Existing clients are not compatible causing Kinesis integration tests to fail at runtime:
```
ENABLE_KINESIS_TESTS=1 ./build/sbt -Pkinesis-asl
...
Using endpoint URL https://kinesis.us-west-2.amazonaws.com for creating Kinesis streams for tests.
[info] WithoutAggregationKinesisBackedBlockRDDSuite:
[info] org.apache.spark.streaming.kinesis.WithoutAggregationKinesisBackedBlockRDDSuite *** ABORTED *** (1 second, 131 milliseconds)
[info]   java.lang.NoClassDefFoundError: com/fasterxml/jackson/databind/PropertyNamingStrategy$PascalCaseStrategy
[info]   at com.amazonaws.services.kinesis.AmazonKinesisClient.<clinit>(AmazonKinesisClient.java:86)
[info]   at org.apache.spark.streaming.kinesis.KinesisTestUtils.kinesisClient$lzycompute(KinesisTestUtils.scala:59)
[info]   at org.apache.spark.streaming.kinesis.KinesisTestUtils.kinesisClient(KinesisTestUtils.scala:58)
[info]   at org.apache.spark.streaming.kinesis.KinesisTestUtils.describeStream(KinesisTestUtils.scala:169)
[info]   at org.apache.spark.streaming.kinesis.KinesisTestUtils.findNonExistentStreamName(KinesisTestUtils.scala:182)
[info]   at org.apache.spark.streaming.kinesis.KinesisTestUtils.createStream(KinesisTestUtils.scala:85)
[info]   at org.apache.spark.streaming.kinesis.KinesisBackedBlockRDDTests.$anonfun$beforeAll$1(KinesisBackedBlockRDDSuite.scala:45)
[info]   at org.apache.spark.streaming.kinesis.KinesisFunSuite.runIfTestsEnabled(KinesisFunSuite.scala:41)
[info]   at org.apache.spark.streaming.kinesis.KinesisFunSuite.runIfTestsEnabled$(KinesisFunSuite.scala:39)
[info]   at org.apache.spark.streaming.kinesis.KinesisBackedBlockRDDTests.runIfTestsEnabled(KinesisBackedBlockRDDSuite.scala:26)
[info]   at org.apache.spark.streaming.kinesis.KinesisBackedBlockRDDTests.beforeAll(KinesisBackedBlockRDDSuite.scala:43)
[info]   at org.scalatest.BeforeAndAfterAll.liftedTree1$1(BeforeAndAfterAll.scala:212)
[info]   at org.scalatest.BeforeAndAfterAll.run(BeforeAndAfterAll.scala:210)
[info]   at org.scalatest.BeforeAndAfterAll.run$(BeforeAndAfterAll.scala:208)
[info]   at org.apache.spark.SparkFunSuite.run(SparkFunSuite.scala:68)
[info]   at org.scalatest.tools.Framework.org$scalatest$tools$Framework$$runSuite(Framework.scala:321)
[info]   at org.scalatest.tools.Framework$ScalaTestTask.execute(Framework.scala:517)
[info]   at sbt.ForkMain$Run.lambda$runTest$1(ForkMain.java:414)
[info]   at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
[info]   at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
[info]   at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
[info]   at java.base/java.lang.Thread.run(Thread.java:840)
[info]   Cause: java.lang.ClassNotFoundException: com.fasterxml.jackson.databind.PropertyNamingStrategy$PascalCaseStrategy
[info]   at java.base/jdk.internal.loader.BuiltinClassLoader.loadClass(BuiltinClassLoader.java:641)
[info]   at java.base/jdk.internal.loader.ClassLoaders$AppClassLoader.loadClass(ClassLoaders.java:188)
[info]   at java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:525)
[info]   at com.amazonaws.services.kinesis.AmazonKinesisClient.<clinit>(AmazonKinesisClient.java:86)
[info]   at org.apache.spark.streaming.kinesis.KinesisTestUtils.kinesisClient$lzycompute(KinesisTestUtils.scala:59)
[info]   at org.apache.spark.streaming.kinesis.KinesisTestUtils.kinesisClient(KinesisTestUtils.scala:58)
[info]   at org.apache.spark.streaming.kinesis.KinesisTestUtils.describeStream(KinesisTestUtils.scala:169)
[info]   at org.apache.spark.streaming.kinesis.KinesisTestUtils.findNonExistentStreamName(KinesisTestUtils.scala:182)
[info]   at org.apache.spark.streaming.kinesis.KinesisTestUtils.createStream(KinesisTestUtils.scala:85)
[info]   at org.apache.spark.streaming.kinesis.KinesisBackedBlockRDDTests.$anonfun$beforeAll$1(KinesisBackedBlockRDDSuite.scala:45)
[info]   at org.apache.spark.streaming.kinesis.KinesisFunSuite.runIfTestsEnabled(KinesisFunSuite.scala:41)
[info]   at org.apache.spark.streaming.kinesis.KinesisFunSuite.runIfTestsEnabled$(KinesisFunSuite.scala:39)
[info]   at org.apache.spark.streaming.kinesis.KinesisBackedBlockRDDTests.runIfTestsEnabled(KinesisBackedBlockRDDSuite.scala:26)
[info]   at org.apache.spark.streaming.kinesis.KinesisBackedBlockRDDTests.beforeAll(KinesisBackedBlockRDDSuite.scala:43)
[info]   at org.scalatest.BeforeAndAfterAll.liftedTree1$1(BeforeAndAfterAll.scala:212)
[info]   at org.scalatest.BeforeAndAfterAll.run(BeforeAndAfterAll.scala:210)
[info]   at org.scalatest.BeforeAndAfterAll.run$(BeforeAndAfterAll.scala:208)
[info]   at org.apache.spark.SparkFunSuite.run(SparkFunSuite.scala:68)
[info]   at org.scalatest.tools.Framework.org$scalatest$tools$Framework$$runSuite(Framework.scala:321)
[info]   at org.scalatest.tools.Framework$ScalaTestTask.execute(Framework.scala:517)
[info]   at sbt.ForkMain$Run.lambda$runTest$1(ForkMain.java:414)
[info]   at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
[info]   at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
[info]   at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
[info]   at java.base/java.lang.Thread.run(Thread.java:840)
[error] Uncaught exception when running org.apache.spark.streaming.kinesis.WithoutAggregationKinesisBackedBlockRDDSuite: java.lang.NoClassDefFoundError: com/fasterxml/jackson/databind/PropertyNamingStrategy$PascalCaseStrategy
```

### Does this PR introduce _any_ user-facing change?
No, only minor version upgrade for the Kinesis and AWS Java SDK libraries

### How was this patch tested?
```
ENABLE_KINESIS_TESTS=1 ./build/sbt -Pkinesis-asl
```
and
```
ENABLE_KINESIS_TESTS=1 build/mvn test -Pkinesis-asl -pl connector/kinesis-asl
```

### Was this patch authored or co-authored using generative AI tooling?
No

Closes #52630 from vrozov/SPARK-53927.

Authored-by: Vlad Rozov <vrozov@amazon.com>
Signed-off-by: Kousuke Saruta <sarutak@apache.org>

Author: vrozov

connector/kinesis-asl/pom.xml

@@ -64,10 +64,16 @@
       <version>${aws.java.sdk.version}</version>
     </dependency>
     <dependency>
-      <groupId>com.amazonaws</groupId>
+      <groupId>software.amazon.kinesis</groupId>
       <artifactId>amazon-kinesis-producer</artifactId>
       <version>${aws.kinesis.producer.version}</version>
       <scope>test</scope>
+      <exclusions>
+        <exclusion>
+          <groupId>com.kjetland</groupId>
+          <artifactId>mbknor-jackson-jsonschema_2.12</artifactId>
+        </exclusion>
+      </exclusions>
     </dependency>
     <!-- manage this up explicitly to match Spark; com.amazonaws:aws-java-sdk-pom specifies
       2.6.7 but says we can manage it up -->

connector/kinesis-asl/src/test/scala/org/apache/spark/streaming/kinesis/KPLBasedKinesisTestUtils.scala

@@ -18,15 +18,14 @@ package org.apache.spark.streaming.kinesis
 
 import java.nio.ByteBuffer
 import java.nio.charset.StandardCharsets
+import java.util.concurrent.{Executors, TimeUnit}
 
 import scala.collection.mutable
 import scala.collection.mutable.ArrayBuffer
 
-import com.amazonaws.services.kinesis.producer.{KinesisProducer => KPLProducer,
-  KinesisProducerConfiguration, UserRecordResult}
 import com.google.common.util.concurrent.{FutureCallback, Futures}
-
-import org.apache.spark.util.ThreadUtils
+import software.amazon.kinesis.producer.{KinesisProducer => KPLProducer,
+  KinesisProducerConfiguration, UserRecordResult}
 
 private[kinesis] class KPLBasedKinesisTestUtils(streamShardCount: Int = 2)
     extends KinesisTestUtils(streamShardCount) {
@@ -53,6 +52,7 @@ private[kinesis] class KPLDataGenerator(regionName: String) extends KinesisDataG
   }
 
   override def sendData(streamName: String, data: Seq[Int]): Map[String, Seq[(Int, String)]] = {
+    val executor = Executors.newSingleThreadExecutor()
     val shardIdToSeqNumbers = new mutable.HashMap[String, ArrayBuffer[(Int, String)]]()
     data.foreach { num =>
       val str = num.toString
@@ -63,15 +63,17 @@ private[kinesis] class KPLDataGenerator(regionName: String) extends KinesisDataG
 
         override def onSuccess(result: UserRecordResult): Unit = {
           val shardId = result.getShardId
-          val seqNumber = result.getSequenceNumber()
+          val seqNumber = result.getSequenceNumber
           val sentSeqNumbers = shardIdToSeqNumbers.getOrElseUpdate(shardId,
             new ArrayBuffer[(Int, String)]())
           sentSeqNumbers += ((num, seqNumber))
         }
       }
-      Futures.addCallback(future, kinesisCallBack, ThreadUtils.sameThreadExecutorService())
+      Futures.addCallback(future, kinesisCallBack, executor)
     }
     producer.flushSync()
-    shardIdToSeqNumbers.toMap.transform((_, v) => v.toSeq)
+    executor.shutdown()
+    executor.awaitTermination(10, TimeUnit.SECONDS)
+    shardIdToSeqNumbers.toMap.transform((_, v) => v.toSeq.sortBy(_._2))
   }
 }

connector/kinesis-asl/src/test/scala/org/apache/spark/streaming/kinesis/KinesisTestUtils.scala

@@ -32,6 +32,7 @@ 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 com.amazonaws.waiters.WaiterParameters
 
 import org.apache.spark.internal.Logging
 import org.apache.spark.internal.LogKeys.{STREAM_NAME, TABLE_NAME}
@@ -61,6 +62,8 @@ private[kinesis] class KinesisTestUtils(streamShardCount: Int = 2) extends Loggi
     client
   }
 
+  private lazy val streamExistsWaiter = kinesisClient.waiters().streamExists()
+
   private lazy val dynamoDB = {
     val dynamoDBClient = new AmazonDynamoDBClient(new DefaultAWSCredentialsProviderChain())
     dynamoDBClient.setRegion(RegionUtils.getRegion(regionName))
@@ -184,18 +187,9 @@ private[kinesis] class KinesisTestUtils(streamShardCount: Int = 2) extends Loggi
   }
 
   private def waitForStreamToBeActive(streamNameToWaitFor: String): Unit = {
-    val startTimeNs = System.nanoTime()
-    while (System.nanoTime() - startTimeNs < TimeUnit.SECONDS.toNanos(createStreamTimeoutSeconds)) {
-      Thread.sleep(TimeUnit.SECONDS.toMillis(describeStreamPollTimeSeconds))
-      describeStream(streamNameToWaitFor).foreach { description =>
-        val streamStatus = description.getStreamStatus()
-        logDebug(s"\t- current state: $streamStatus\n")
-        if ("ACTIVE".equals(streamStatus)) {
-          return
-        }
-      }
-    }
-    require(false, s"Stream $streamName never became active")
+    val describeStreamRequest = new DescribeStreamRequest()
+      .withStreamName(streamNameToWaitFor)
+    streamExistsWaiter.run(new WaiterParameters(describeStreamRequest))
   }
 }
 

pom.xml

@@ -158,12 +158,12 @@
     <codahale.metrics.version>4.2.33</codahale.metrics.version>
     <!-- Should be consistent with SparkBuild.scala and docs -->
     <avro.version>1.12.0</avro.version>
-    <aws.kinesis.client.version>1.12.0</aws.kinesis.client.version>
+    <aws.kinesis.client.version>1.15.3</aws.kinesis.client.version>
     <!-- Should be consistent with Kinesis client dependency -->
-    <aws.java.sdk.version>1.11.655</aws.java.sdk.version>
+    <aws.java.sdk.version>1.12.681</aws.java.sdk.version>
     <aws.java.sdk.v2.version>2.29.52</aws.java.sdk.v2.version>
     <!-- the producer is used in tests -->
-    <aws.kinesis.producer.version>0.12.8</aws.kinesis.producer.version>
+    <aws.kinesis.producer.version>1.0.5</aws.kinesis.producer.version>
     <!-- Do not use 3.0.0: https://github.com/GoogleCloudDataproc/hadoop-connectors/issues/1114 -->
     <gcs-connector.version>hadoop3-2.2.28</gcs-connector.version>
     <analyticsaccelerator-s3.version>1.3.0</analyticsaccelerator-s3.version>