diff --git a/connector/kinesis-asl-assembly/pom.xml b/connector/kinesis-asl-assembly/pom.xml
index cd5c0393f6f8..37b960e42550 100644
--- a/connector/kinesis-asl-assembly/pom.xml
+++ b/connector/kinesis-asl-assembly/pom.xml
@@ -181,6 +181,10 @@
               <include>com.google.protobuf.**</include>
             </includes>
           </relocation>
+          <relocation>
+            <pattern>com.google.common</pattern>
+            <shadedPattern>${spark.shade.packageName}.kinesis.guava</shadedPattern>
+          </relocation>
         </relocations>
         <filters>
           <filter>
diff --git a/connector/kinesis-asl/pom.xml b/connector/kinesis-asl/pom.xml
index c70a073e7340..d207be6c02ed 100644
--- a/connector/kinesis-asl/pom.xml
+++ b/connector/kinesis-asl/pom.xml
@@ -54,14 +54,25 @@
       <scope>test</scope>
     </dependency>
     <dependency>
-      <groupId>com.amazonaws</groupId>
+      <groupId>software.amazon.kinesis</groupId>
       <artifactId>amazon-kinesis-client</artifactId>
-      <version>${aws.kinesis.client.version}</version>
+      <version>${aws.kinesis.client.v2.version}</version>
+      <exclusions>
+        <exclusion>
+          <groupId>com.amazonaws</groupId>
+          <artifactId>*</artifactId>
+        </exclusion>
+      </exclusions>
     </dependency>
     <dependency>
-      <groupId>com.amazonaws</groupId>
-      <artifactId>aws-java-sdk-sts</artifactId>
-      <version>${aws.java.sdk.version}</version>
+      <groupId>software.amazon.awssdk</groupId>
+      <artifactId>sts</artifactId>
+      <version>${aws.java.sdk.v2.version}</version>
+    </dependency>
+    <dependency>
+      <groupId>software.amazon.awssdk</groupId>
+      <artifactId>apache-client</artifactId>
+      <version>${aws.java.sdk.v2.version}</version>
     </dependency>
     <dependency>
       <groupId>com.amazonaws</groupId>
@@ -90,6 +101,11 @@
       <groupId>org.apache.spark</groupId>
       <artifactId>spark-tags_${scala.binary.version}</artifactId>
     </dependency>
+    <dependency>
+      <groupId>com.google.guava</groupId>
+      <artifactId>guava</artifactId>
+      <version>31.1-jre</version>
+    </dependency>
 
     <!--
       This spark-tags test-dep is needed even though it isn't used in this module, otherwise testing-cmds that exclude
diff --git a/connector/kinesis-asl/src/main/java/org/apache/spark/examples/streaming/JavaKinesisWordCountASL.java b/connector/kinesis-asl/src/main/java/org/apache/spark/examples/streaming/JavaKinesisWordCountASL.java
index 636af9f6c606..550833c29668 100644
--- a/connector/kinesis-asl/src/main/java/org/apache/spark/examples/streaming/JavaKinesisWordCountASL.java
+++ b/connector/kinesis-asl/src/main/java/org/apache/spark/examples/streaming/JavaKinesisWordCountASL.java
@@ -16,6 +16,7 @@
  */
 package org.apache.spark.examples.streaming;
 
+import java.net.URI;
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -38,8 +39,10 @@
 import scala.Tuple2;
 import scala.reflect.ClassTag$;
 
-import com.amazonaws.auth.DefaultAWSCredentialsProviderChain;
-import com.amazonaws.services.kinesis.AmazonKinesisClient;
+import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider;
+import software.amazon.awssdk.http.apache.ApacheHttpClient;
+import software.amazon.awssdk.services.kinesis.KinesisClient;
+import software.amazon.awssdk.services.kinesis.model.DescribeStreamRequest;
 
 /**
  * Consumes messages from a Amazon Kinesis streams and does wordcount.
@@ -66,7 +69,7 @@
  * There is a companion helper class called KinesisWordProducerASL which puts dummy data
  * onto the Kinesis stream.
  *
- * This code uses the DefaultAWSCredentialsProviderChain to find credentials
+ * This code uses the DefaultCredentialsProvider to find credentials
  * in the following order:
  *    Environment Variables - AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY
  *    Java System Properties - aws.accessKeyId and aws.secretKey
@@ -106,11 +109,18 @@ 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
-    AmazonKinesisClient kinesisClient =
-        new AmazonKinesisClient(new DefaultAWSCredentialsProviderChain());
-    kinesisClient.setEndpoint(endpointUrl);
+    KinesisClient kinesisClient =
+         KinesisClient.builder()
+                 .credentialsProvider(DefaultCredentialsProvider.create())
+                 .endpointOverride(URI.create(endpointUrl))
+                 .httpClientBuilder(ApacheHttpClient.builder())
+                 .build();
+
+    DescribeStreamRequest describeStreamRequest = DescribeStreamRequest.builder()
+            .streamName(streamName)
+            .build();
     int numShards =
-        kinesisClient.describeStream(streamName).getStreamDescription().getShards().size();
+        kinesisClient.describeStream(describeStreamRequest).streamDescription().shards().size();
 
 
     // In this example, we're going to create 1 Kinesis Receiver/input DStream for each shard.
diff --git a/connector/kinesis-asl/src/main/java/org/apache/spark/streaming/kinesis/KinesisInitialPositions.java b/connector/kinesis-asl/src/main/java/org/apache/spark/streaming/kinesis/KinesisInitialPositions.java
index b5f5ab0e9054..8d03e7d020f0 100644
--- a/connector/kinesis-asl/src/main/java/org/apache/spark/streaming/kinesis/KinesisInitialPositions.java
+++ b/connector/kinesis-asl/src/main/java/org/apache/spark/streaming/kinesis/KinesisInitialPositions.java
@@ -16,7 +16,8 @@
  */
 package org.apache.spark.streaming.kinesis;
 
-import com.amazonaws.services.kinesis.clientlibrary.lib.worker.InitialPositionInStream;
+
+import software.amazon.kinesis.common.InitialPositionInStream;
 
 import java.io.Serializable;
 import java.util.Date;
diff --git a/connector/kinesis-asl/src/main/scala/org/apache/spark/examples/streaming/KinesisExampleUtils.scala b/connector/kinesis-asl/src/main/scala/org/apache/spark/examples/streaming/KinesisExampleUtils.scala
index 2eebd6130d4d..ebe5f6c447d3 100644
--- a/connector/kinesis-asl/src/main/scala/org/apache/spark/examples/streaming/KinesisExampleUtils.scala
+++ b/connector/kinesis-asl/src/main/scala/org/apache/spark/examples/streaming/KinesisExampleUtils.scala
@@ -19,16 +19,16 @@ package org.apache.spark.examples.streaming
 
 import scala.collection.JavaConverters._
 
-import com.amazonaws.regions.RegionUtils
-import com.amazonaws.services.kinesis.AmazonKinesis
+import software.amazon.awssdk.regions.servicemetadata.KinesisServiceMetadata
 
 private[streaming] object KinesisExampleUtils {
   def getRegionNameByEndpoint(endpoint: String): String = {
     val uri = new java.net.URI(endpoint)
-    RegionUtils.getRegionsForService(AmazonKinesis.ENDPOINT_PREFIX)
+    val kinesisServiceMetadata = new KinesisServiceMetadata()
+    kinesisServiceMetadata.regions()
       .asScala
-      .find(_.getAvailableEndpoints.asScala.toSeq.contains(uri.getHost))
-      .map(_.getName)
+      .find(r => kinesisServiceMetadata.endpointFor(r).toString.equals(uri.getHost))
+      .map(_.id())
       .getOrElse(
         throw new IllegalArgumentException(s"Could not resolve region for endpoint: $endpoint"))
   }
diff --git a/connector/kinesis-asl/src/main/scala/org/apache/spark/examples/streaming/KinesisWordCountASL.scala b/connector/kinesis-asl/src/main/scala/org/apache/spark/examples/streaming/KinesisWordCountASL.scala
index 7d12af3256f1..cefe63912e0c 100644
--- a/connector/kinesis-asl/src/main/scala/org/apache/spark/examples/streaming/KinesisWordCountASL.scala
+++ b/connector/kinesis-asl/src/main/scala/org/apache/spark/examples/streaming/KinesisWordCountASL.scala
@@ -18,15 +18,18 @@
 // scalastyle:off println
 package org.apache.spark.examples.streaming
 
+import java.net.URI
 import java.nio.ByteBuffer
 
 import scala.util.Random
 
-import com.amazonaws.auth.DefaultAWSCredentialsProviderChain
-import com.amazonaws.services.kinesis.AmazonKinesisClient
-import com.amazonaws.services.kinesis.model.PutRecordRequest
 import org.apache.logging.log4j.Level
 import org.apache.logging.log4j.core.config.Configurator
+import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider
+import software.amazon.awssdk.core.SdkBytes
+import software.amazon.awssdk.http.apache.ApacheHttpClient
+import software.amazon.awssdk.services.kinesis.KinesisClient
+import software.amazon.awssdk.services.kinesis.model.{DescribeStreamRequest, PutRecordRequest}
 
 import org.apache.spark.SparkConf
 import org.apache.spark.internal.Logging
@@ -101,13 +104,18 @@ object KinesisWordCountASL extends Logging {
 
     // Determine the number of shards from the stream using the low-level Kinesis Client
     // from the AWS Java SDK.
-    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 = new AmazonKinesisClient(credentials)
-    kinesisClient.setEndpoint(endpointUrl)
-    val numShards = kinesisClient.describeStream(streamName).getStreamDescription().getShards().size
+    val kinesisClient = KinesisClient.builder()
+      .credentialsProvider(DefaultCredentialsProvider.create())
+      .endpointOverride(URI.create(endpointUrl))
+      .httpClientBuilder(ApacheHttpClient.builder())
+      .build()
+    val describeStreamRequest = DescribeStreamRequest.builder()
+      .streamName(streamName)
+      .build()
+    val numShards = kinesisClient.describeStream(describeStreamRequest)
+      .streamDescription()
+      .shards()
+      .size
 
 
     // In this example, we're going to create 1 Kinesis Receiver/input DStream for each shard.
@@ -221,8 +229,11 @@ object KinesisWordProducerASL {
     val totals = scala.collection.mutable.Map[String, Int]()
 
     // Create the low-level Kinesis Client from the AWS Java SDK.
-    val kinesisClient = new AmazonKinesisClient(new DefaultAWSCredentialsProviderChain())
-    kinesisClient.setEndpoint(endpoint)
+    val kinesisClient = KinesisClient.builder()
+      .credentialsProvider(DefaultCredentialsProvider.create())
+      .endpointOverride(URI.create(endpoint))
+      .httpClientBuilder(ApacheHttpClient.builder())
+      .build()
 
     println(s"Putting records onto stream $stream and endpoint $endpoint at a rate of" +
         s" $recordsPerSecond records per second and $wordsPerRecord words per record")
@@ -247,12 +258,14 @@ object KinesisWordProducerASL {
         val partitionKey = s"partitionKey-$recordNum"
 
         // Create a PutRecordRequest with an Array[Byte] version of the data
-        val putRecordRequest = new PutRecordRequest().withStreamName(stream)
-            .withPartitionKey(partitionKey)
-            .withData(ByteBuffer.wrap(data.getBytes()))
+        val putRecordRequest = PutRecordRequest.builder()
+          .streamName(stream)
+          .partitionKey(partitionKey)
+          .data(SdkBytes.fromByteBuffer(ByteBuffer.wrap(data.getBytes())))
+          .build()
 
         // Put the record onto the stream and capture the PutRecordResult
-        val putRecordResult = kinesisClient.putRecord(putRecordRequest)
+        kinesisClient.putRecord(putRecordRequest)
       }
 
       // Sleep for a second
diff --git a/connector/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisBackedBlockRDD.scala b/connector/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisBackedBlockRDD.scala
index ab55d545770e..30280c92cb86 100644
--- a/connector/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisBackedBlockRDD.scala
+++ b/connector/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisBackedBlockRDD.scala
@@ -17,16 +17,20 @@
 
 package org.apache.spark.streaming.kinesis
 
+import java.net.URI
+import java.util
 import java.util.concurrent.TimeUnit
 
 import scala.collection.JavaConverters._
 import scala.reflect.ClassTag
 import scala.util.control.NonFatal
 
-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._
+import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider
+import software.amazon.awssdk.http.apache.ApacheHttpClient
+import software.amazon.awssdk.regions.Region
+import software.amazon.awssdk.services.kinesis.KinesisClient
+import software.amazon.awssdk.services.kinesis.model.{GetRecordsRequest, GetRecordsResponse, GetShardIteratorRequest, GetShardIteratorResponse, ProvisionedThroughputExceededException, ShardIteratorType}
+import software.amazon.kinesis.retrieval.{AggregatorUtil, KinesisClientRecord}
 
 import org.apache.spark._
 import org.apache.spark.internal.Logging
@@ -83,7 +87,7 @@ class KinesisBackedBlockRDD[T: ClassTag](
     @transient private val _blockIds: Array[BlockId],
     @transient val arrayOfseqNumberRanges: Array[SequenceNumberRanges],
     @transient private val isBlockIdValid: Array[Boolean] = Array.empty,
-    val messageHandler: Record => T = KinesisInputDStream.defaultMessageHandler _,
+    val messageHandler: KinesisClientRecord => T = KinesisInputDStream.defaultMessageHandler _,
     val kinesisCreds: SparkAWSCredentials = DefaultCredentials,
     val kinesisReadConfigs: KinesisReadConfigurations = KinesisReadConfigurations()
   ) extends BlockRDD[T](sc, _blockIds) {
@@ -111,9 +115,9 @@ class KinesisBackedBlockRDD[T: ClassTag](
     }
 
     def getBlockFromKinesis(): Iterator[T] = {
-      val credentials = kinesisCreds.provider.getCredentials
+      val credentialsProvider = kinesisCreds.provider
       partition.seqNumberRanges.ranges.iterator.flatMap { range =>
-        new KinesisSequenceRangeIterator(credentials, endpointUrl, regionName,
+        new KinesisSequenceRangeIterator(credentialsProvider, endpointUrl, regionName,
           range, kinesisReadConfigs).map(messageHandler)
       }
     }
@@ -133,13 +137,19 @@ class KinesisBackedBlockRDD[T: ClassTag](
  */
 private[kinesis]
 class KinesisSequenceRangeIterator(
-    credentials: AWSCredentials,
+    credentialsProvider: AwsCredentialsProvider,
     endpointUrl: String,
     regionId: String,
     range: SequenceNumberRange,
-    kinesisReadConfigs: KinesisReadConfigurations) extends NextIterator[Record] with Logging {
-
-  private val client = new AmazonKinesisClient(credentials)
+    kinesisReadConfigs: KinesisReadConfigurations)
+  extends NextIterator[KinesisClientRecord] with Logging {
+
+  private val client = KinesisClient.builder()
+    .credentialsProvider(credentialsProvider)
+    .region(Region.of(regionId))
+    .endpointOverride(URI.create(endpointUrl))
+    .httpClientBuilder(ApacheHttpClient.builder())
+    .build()
   private val streamName = range.streamName
   private val shardId = range.shardId
   // AWS limits to maximum of 10k records per get call
@@ -147,12 +157,10 @@ class KinesisSequenceRangeIterator(
 
   private var toSeqNumberReceived = false
   private var lastSeqNumber: String = null
-  private var internalIterator: Iterator[Record] = null
-
-  client.setEndpoint(endpointUrl)
+  private var internalIterator: Iterator[KinesisClientRecord] = null
 
-  override protected def getNext(): Record = {
-    var nextRecord: Record = null
+  override protected def getNext(): KinesisClientRecord = {
+    var nextRecord: KinesisClientRecord = null
     if (toSeqNumberReceived) {
       finished = true
     } else {
@@ -182,11 +190,11 @@ class KinesisSequenceRangeIterator(
 
         // Get the record, copy the data into a byte array and remember its sequence number
         nextRecord = internalIterator.next()
-        lastSeqNumber = nextRecord.getSequenceNumber()
+        lastSeqNumber = nextRecord.sequenceNumber()
 
         // If the this record's sequence number matches the stopping sequence number, then make sure
         // the iterator is marked finished next time getNext() is called
-        if (nextRecord.getSequenceNumber == range.toSeqNumber) {
+        if (nextRecord.sequenceNumber == range.toSeqNumber) {
           toSeqNumberReceived = true
         }
       }
@@ -195,7 +203,7 @@ class KinesisSequenceRangeIterator(
   }
 
   override protected def close(): Unit = {
-    client.shutdown()
+    client.close()
   }
 
   /**
@@ -204,7 +212,7 @@ class KinesisSequenceRangeIterator(
   private def getRecords(
       iteratorType: ShardIteratorType,
       seqNum: String,
-      recordCount: Int): Iterator[Record] = {
+      recordCount: Int): Iterator[KinesisClientRecord] = {
     val shardIterator = getKinesisIterator(iteratorType, seqNum)
     val result = getRecordsAndNextKinesisIterator(shardIterator, recordCount)
     result._1
@@ -216,19 +224,22 @@ class KinesisSequenceRangeIterator(
    */
   private def getRecordsAndNextKinesisIterator(
       shardIterator: String,
-      recordCount: Int): (Iterator[Record], String) = {
-    val getRecordsRequest = new GetRecordsRequest
-    getRecordsRequest.setRequestCredentials(credentials)
-    getRecordsRequest.setShardIterator(shardIterator)
-    getRecordsRequest.setLimit(Math.min(recordCount, this.maxGetRecordsLimit))
-    val getRecordsResult = retryOrTimeout[GetRecordsResult](
+      recordCount: Int): (Iterator[KinesisClientRecord], String) = {
+    val getRecordsRequest = GetRecordsRequest.builder()
+      .shardIterator(shardIterator)
+      .limit(Math.min(recordCount, this.maxGetRecordsLimit))
+      .build()
+    val getRecordsResponse = retryOrTimeout[GetRecordsResponse](
       s"getting records using shard iterator") {
         client.getRecords(getRecordsRequest)
       }
+    val aggregatorUtil = new AggregatorUtil()
+    val records = new util.ArrayList[KinesisClientRecord]()
+    getRecordsResponse.records().forEach(r => records.add(KinesisClientRecord.fromRecord(r)))
     // 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)
-    (recordIterator.iterator().asScala, getRecordsResult.getNextShardIterator)
+    val recordIterator = aggregatorUtil.deaggregate(records)
+    (recordIterator.iterator.asScala, getRecordsResponse.nextShardIterator)
   }
 
   /**
@@ -238,17 +249,18 @@ class KinesisSequenceRangeIterator(
   private def getKinesisIterator(
       iteratorType: ShardIteratorType,
       sequenceNumber: String): String = {
-    val getShardIteratorRequest = new GetShardIteratorRequest
-    getShardIteratorRequest.setRequestCredentials(credentials)
-    getShardIteratorRequest.setStreamName(streamName)
-    getShardIteratorRequest.setShardId(shardId)
-    getShardIteratorRequest.setShardIteratorType(iteratorType.toString)
-    getShardIteratorRequest.setStartingSequenceNumber(sequenceNumber)
-    val getShardIteratorResult = retryOrTimeout[GetShardIteratorResult](
+    val getShardIteratorRequest = GetShardIteratorRequest.builder()
+      .streamName(streamName)
+      .shardId(shardId)
+      .shardIteratorType(iteratorType)
+      .startingSequenceNumber(sequenceNumber)
+      .build()
+
+    val getShardIteratorResponse = retryOrTimeout[GetShardIteratorResponse](
         s"getting shard iterator from sequence number $sequenceNumber") {
           client.getShardIterator(getShardIteratorRequest)
         }
-    getShardIteratorResult.getShardIterator
+    getShardIteratorResponse.shardIterator
   }
 
   /** Helper method to retry Kinesis API request with exponential backoff and timeouts */
diff --git a/connector/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisCheckpointer.scala b/connector/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisCheckpointer.scala
index 770eb2d89d52..5d606d73ff9f 100644
--- a/connector/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisCheckpointer.scala
+++ b/connector/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisCheckpointer.scala
@@ -20,7 +20,7 @@ import java.util.concurrent._
 
 import scala.util.control.NonFatal
 
-import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorCheckpointer
+import software.amazon.kinesis.processor.RecordProcessorCheckpointer
 
 import org.apache.spark.internal.Logging
 import org.apache.spark.streaming.Duration
@@ -42,14 +42,14 @@ private[kinesis] class KinesisCheckpointer(
     clock: Clock = new SystemClock) extends Logging {
 
   // a map from shardId's to checkpointers
-  private val checkpointers = new ConcurrentHashMap[String, IRecordProcessorCheckpointer]()
+  private val checkpointers = new ConcurrentHashMap[String, RecordProcessorCheckpointer]()
 
   private val lastCheckpointedSeqNums = new ConcurrentHashMap[String, String]()
 
   private val checkpointerThread: RecurringTimer = startCheckpointerThread()
 
   /** Update the checkpointer instance to the most recent one for the given shardId. */
-  def setCheckpointer(shardId: String, checkpointer: IRecordProcessorCheckpointer): Unit = {
+  def setCheckpointer(shardId: String, checkpointer: RecordProcessorCheckpointer): Unit = {
     checkpointers.put(shardId, checkpointer)
   }
 
@@ -60,7 +60,7 @@ private[kinesis] class KinesisCheckpointer(
    * we will use that to make the final checkpoint. If `null` is provided, we will not make the
    * checkpoint, e.g. in case of [[ShutdownReason.ZOMBIE]].
    */
-  def removeCheckpointer(shardId: String, checkpointer: IRecordProcessorCheckpointer): Unit = {
+  def removeCheckpointer(shardId: String, checkpointer: RecordProcessorCheckpointer): Unit = {
     synchronized {
       checkpointers.remove(shardId)
     }
@@ -81,7 +81,7 @@ private[kinesis] class KinesisCheckpointer(
   }
 
   /** Perform the checkpoint. */
-  private def checkpoint(shardId: String, checkpointer: IRecordProcessorCheckpointer): Unit = {
+  private def checkpoint(shardId: String, checkpointer: RecordProcessorCheckpointer): Unit = {
     try {
       if (checkpointer != null) {
         receiver.getLatestSeqNumToCheckpoint(shardId).foreach { latestSeqNum =>
diff --git a/connector/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisInputDStream.scala b/connector/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisInputDStream.scala
index c68af3864938..9fa0ad9a5258 100644
--- a/connector/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisInputDStream.scala
+++ b/connector/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisInputDStream.scala
@@ -17,12 +17,11 @@
 
 package org.apache.spark.streaming.kinesis
 
-import scala.collection.JavaConverters._
 import scala.reflect.ClassTag
 
-import com.amazonaws.services.kinesis.clientlibrary.lib.worker.{InitialPositionInStream, KinesisClientLibConfiguration}
-import com.amazonaws.services.kinesis.metrics.interfaces.MetricsLevel
-import com.amazonaws.services.kinesis.model.Record
+import software.amazon.kinesis.common.InitialPositionInStream
+import software.amazon.kinesis.metrics.{MetricsLevel, MetricsUtil}
+import software.amazon.kinesis.retrieval.KinesisClientRecord
 
 import org.apache.spark.rdd.RDD
 import org.apache.spark.storage.{BlockId, StorageLevel}
@@ -42,7 +41,7 @@ private[kinesis] class KinesisInputDStream[T: ClassTag](
     val checkpointAppName: String,
     val checkpointInterval: Duration,
     val _storageLevel: StorageLevel,
-    val messageHandler: Record => T,
+    val messageHandler: KinesisClientRecord => T,
     val kinesisCreds: SparkAWSCredentials,
     val dynamoDBCreds: Option[SparkAWSCredentials],
     val cloudWatchCreds: Option[SparkAWSCredentials],
@@ -275,7 +274,7 @@ object KinesisInputDStream {
 
     /**
      * Sets the CloudWatch metrics level. Defaults to
-     * [[KinesisClientLibConfiguration.DEFAULT_METRICS_LEVEL]] if no custom value is specified.
+     * [[MetricsLevel.DETAILED]] if no custom value is specified.
      *
      * @param metricsLevel [[MetricsLevel]] to specify the CloudWatch metrics level
      * @return Reference to this [[KinesisInputDStream.Builder]]
@@ -289,7 +288,7 @@ object KinesisInputDStream {
 
     /**
      * Sets the enabled CloudWatch metrics dimensions. Defaults to
-     * [[KinesisClientLibConfiguration.DEFAULT_METRICS_ENABLED_DIMENSIONS]]
+     * [[MetricsLevel.DETAILED]]
      * if no custom value is specified.
      *
      * @param metricsEnabledDimensions Set[String] to specify which CloudWatch metrics dimensions
@@ -307,11 +306,12 @@ object KinesisInputDStream {
      * Create a new instance of [[KinesisInputDStream]] with configured parameters and the provided
      * message handler.
      *
-     * @param handler Function converting [[Record]] instances read by the KCL to DStream type [[T]]
+     * @param handler Function converting [[KinesisClientRecord]] instances read by the KCL to
+     * DStream type [[T]]
      * @return Instance of [[KinesisInputDStream]] constructed with configured parameters
      */
     def buildWithMessageHandler[T: ClassTag](
-        handler: Record => T): KinesisInputDStream[T] = {
+        handler: KinesisClientRecord => T): KinesisInputDStream[T] = {
       val ssc = getRequiredParam(streamingContext, "streamingContext")
       new KinesisInputDStream(
         ssc,
@@ -351,9 +351,9 @@ object KinesisInputDStream {
    */
   def builder: Builder = new Builder
 
-  private[kinesis] def defaultMessageHandler(record: Record): Array[Byte] = {
+  private[kinesis] def defaultMessageHandler(record: KinesisClientRecord): Array[Byte] = {
     if (record == null) return null
-    val byteBuffer = record.getData()
+    val byteBuffer = record.data()
     val byteArray = new Array[Byte](byteBuffer.remaining())
     byteBuffer.get(byteArray)
     byteArray
@@ -365,7 +365,7 @@ object KinesisInputDStream {
   private[kinesis] val DEFAULT_INITIAL_POSITION: KinesisInitialPosition = new Latest()
   private[kinesis] val DEFAULT_STORAGE_LEVEL: StorageLevel = StorageLevel.MEMORY_AND_DISK_2
   private[kinesis] val DEFAULT_METRICS_LEVEL: MetricsLevel =
-    KinesisClientLibConfiguration.DEFAULT_METRICS_LEVEL
+    MetricsLevel.DETAILED
   private[kinesis] val DEFAULT_METRICS_ENABLED_DIMENSIONS: Set[String] =
-    KinesisClientLibConfiguration.DEFAULT_METRICS_ENABLED_DIMENSIONS.asScala.toSet
+    Set(MetricsUtil.OPERATION_DIMENSION_NAME, MetricsUtil.SHARD_ID_DIMENSION_NAME)
 }
diff --git a/connector/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisReceiver.scala b/connector/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisReceiver.scala
index 6feb8f1b5598..f784fd0fb5a0 100644
--- a/connector/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisReceiver.scala
+++ b/connector/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisReceiver.scala
@@ -16,6 +16,7 @@
  */
 package org.apache.spark.streaming.kinesis
 
+import java.net.URI
 import java.util.UUID
 import java.util.concurrent.ConcurrentHashMap
 
@@ -23,10 +24,15 @@ import scala.collection.JavaConverters._
 import scala.collection.mutable
 import scala.util.control.NonFatal
 
-import com.amazonaws.services.kinesis.clientlibrary.interfaces.{IRecordProcessor, IRecordProcessorCheckpointer, IRecordProcessorFactory}
-import com.amazonaws.services.kinesis.clientlibrary.lib.worker.{KinesisClientLibConfiguration, Worker}
-import com.amazonaws.services.kinesis.metrics.interfaces.MetricsLevel
-import com.amazonaws.services.kinesis.model.Record
+import software.amazon.awssdk.regions.Region
+import software.amazon.awssdk.services.cloudwatch.CloudWatchAsyncClient
+import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient
+import software.amazon.awssdk.services.kinesis.KinesisAsyncClient
+import software.amazon.kinesis.common.{ConfigsBuilder, InitialPositionInStreamExtended, KinesisClientUtil}
+import software.amazon.kinesis.coordinator.Scheduler
+import software.amazon.kinesis.metrics.{MetricsConfig, MetricsLevel}
+import software.amazon.kinesis.processor.{RecordProcessorCheckpointer, ShardRecordProcessor, ShardRecordProcessorFactory}
+import software.amazon.kinesis.retrieval.KinesisClientRecord
 
 import org.apache.spark.internal.Logging
 import org.apache.spark.storage.{StorageLevel, StreamBlockId}
@@ -82,6 +88,7 @@ import org.apache.spark.util.Utils
  *                      AWSCredentialsProvider passed to the KCL to authorize DynamoDB API calls.
  *                      Will use kinesisCreds if value is None.
  */
+// TODO: Rename worker to scheduler
 private[kinesis] class KinesisReceiver[T](
     val streamName: String,
     endpointUrl: String,
@@ -90,7 +97,7 @@ private[kinesis] class KinesisReceiver[T](
     checkpointAppName: String,
     checkpointInterval: Duration,
     storageLevel: StorageLevel,
-    messageHandler: Record => T,
+    messageHandler: KinesisClientRecord => T,
     kinesisCreds: SparkAWSCredentials,
     dynamoDBCreds: Option[SparkAWSCredentials],
     cloudWatchCreds: Option[SparkAWSCredentials],
@@ -117,7 +124,7 @@ private[kinesis] class KinesisReceiver[T](
    * Each shard is assigned its own IRecordProcessor and the worker run multiple such
    * processors.
    */
-  @volatile private var worker: Worker = null
+  @volatile private var scheduler: Scheduler = null
   @volatile private var workerThread: Thread = null
 
   /** BlockGenerator used to generates blocks out of Kinesis data */
@@ -154,46 +161,53 @@ private[kinesis] class KinesisReceiver[T](
     kinesisCheckpointer = new KinesisCheckpointer(receiver, checkpointInterval, workerId)
     val kinesisProvider = kinesisCreds.provider
 
-    val kinesisClientLibConfiguration = {
-      val baseClientLibConfiguration = new KinesisClientLibConfiguration(
-        checkpointAppName,
-        streamName,
-        kinesisProvider,
-        dynamoDBCreds.map(_.provider).getOrElse(kinesisProvider),
-        cloudWatchCreds.map(_.provider).getOrElse(kinesisProvider),
-        workerId)
-        .withKinesisEndpoint(endpointUrl)
-        .withTaskBackoffTimeMillis(500)
-        .withRegionName(regionName)
-        .withMetricsLevel(metricsLevel)
-        .withMetricsEnabledDimensions(metricsEnabledDimensions.asJava)
-
-      // Update the Kinesis client lib config with timestamp
-      // if InitialPositionInStream.AT_TIMESTAMP is passed
-      initialPosition match {
-        case ts: AtTimestamp =>
-          baseClientLibConfiguration.withTimestampAtInitialPositionInStream(ts.getTimestamp)
-        case _ =>
-          baseClientLibConfiguration.withInitialPositionInStream(initialPosition.getPosition)
+    val kinesisClient = KinesisClientUtil.createKinesisAsyncClient(
+      KinesisAsyncClient.builder
+        .region(Region.of(regionName))
+        .credentialsProvider(kinesisProvider)
+        .endpointOverride(URI.create(endpointUrl)))
+    val dynamoClient = DynamoDbAsyncClient.builder
+      .region(Region.of(regionName))
+      .credentialsProvider(dynamoDBCreds.map(_.provider).getOrElse(kinesisProvider))
+      .build
+    val cloudWatchClient = CloudWatchAsyncClient.builder
+      .region(Region.of(regionName))
+      .credentialsProvider(cloudWatchCreds.map(_.provider).getOrElse(kinesisProvider))
+      .build
+    val recordProcessorFactory = new ShardRecordProcessorFactory {
+      override def shardRecordProcessor(): ShardRecordProcessor = {
+        new KinesisRecordProcessor(receiver, workerId)
       }
     }
 
-   /*
-    *  RecordProcessorFactory creates impls of IRecordProcessor.
-    *  IRecordProcessor adapts the KCL to our Spark KinesisReceiver via the
-    *  IRecordProcessor.processRecords() method.
-    *  We're using our custom KinesisRecordProcessor in this case.
-    */
-    val recordProcessorFactory = new IRecordProcessorFactory {
-      override def createProcessor: IRecordProcessor =
-        new KinesisRecordProcessor(receiver, workerId)
+    val configsBuilder = new ConfigsBuilder(streamName, checkpointAppName, kinesisClient,
+      dynamoClient, cloudWatchClient, workerId, recordProcessorFactory)
+    val metricsConfig = new MetricsConfig(cloudWatchClient, checkpointAppName)
+      .metricsLevel(metricsLevel)
+      .metricsEnabledDimensions(metricsEnabledDimensions.asJava)
+
+    val initialPositionInStreamExtended = initialPosition match {
+      case ts: AtTimestamp =>
+        InitialPositionInStreamExtended.newInitialPositionAtTimestamp(ts.getTimestamp)
+      case _ =>
+        InitialPositionInStreamExtended.newInitialPosition(initialPosition.getPosition)
     }
 
-    worker = new Worker(recordProcessorFactory, kinesisClientLibConfiguration)
+    scheduler = new Scheduler(
+      configsBuilder.checkpointConfig(),
+      configsBuilder.coordinatorConfig(),
+      configsBuilder.leaseManagementConfig(),
+      configsBuilder.lifecycleConfig(),
+      metricsConfig,
+      configsBuilder.processorConfig(),
+      configsBuilder.retrievalConfig()
+        .initialPositionInStreamExtended(initialPositionInStreamExtended)
+    )
+
     workerThread = new Thread() {
       override def run(): Unit = {
         try {
-          worker.run()
+          scheduler.run()
         } catch {
           case NonFatal(e) =>
             restart("Error running the KCL worker in Receiver", e)
@@ -218,9 +232,9 @@ private[kinesis] class KinesisReceiver[T](
    */
   override def onStop(): Unit = {
     if (workerThread != null) {
-      if (worker != null) {
-        worker.shutdown()
-        worker = null
+      if (scheduler != null) {
+        scheduler.shutdown()
+        scheduler = null
       }
       workerThread.join()
       workerThread = null
@@ -234,11 +248,12 @@ private[kinesis] class KinesisReceiver[T](
   }
 
   /** Add records of the given shard to the current block being generated */
-  private[kinesis] def addRecords(shardId: String, records: java.util.List[Record]): Unit = {
+  private[kinesis] def addRecords(shardId: String, records: java.util.List[KinesisClientRecord]):
+  Unit = {
     if (records.size > 0) {
       val dataIterator = records.iterator().asScala.map(messageHandler)
       val metadata = SequenceNumberRange(streamName, shardId,
-        records.get(0).getSequenceNumber(), records.get(records.size() - 1).getSequenceNumber(),
+        records.get(0).sequenceNumber(), records.get(records.size() - 1).sequenceNumber(),
         records.size())
       blockGenerator.addMultipleDataWithCallback(dataIterator, metadata)
     }
@@ -259,7 +274,7 @@ private[kinesis] class KinesisReceiver[T](
    * Set the checkpointer that will be used to checkpoint sequence numbers to DynamoDB for the
    * given shardId.
    */
-  def setCheckpointer(shardId: String, checkpointer: IRecordProcessorCheckpointer): Unit = {
+  def setCheckpointer(shardId: String, checkpointer: RecordProcessorCheckpointer): Unit = {
     assert(kinesisCheckpointer != null, "Kinesis Checkpointer not initialized!")
     kinesisCheckpointer.setCheckpointer(shardId, checkpointer)
   }
@@ -269,7 +284,7 @@ private[kinesis] class KinesisReceiver[T](
    * checkpoint one last time for the given shard. If `checkpointer` is `null`, then we will not
    * checkpoint.
    */
-  def removeCheckpointer(shardId: String, checkpointer: IRecordProcessorCheckpointer): Unit = {
+  def removeCheckpointer(shardId: String, checkpointer: RecordProcessorCheckpointer): Unit = {
     assert(kinesisCheckpointer != null, "Kinesis Checkpointer not initialized!")
     kinesisCheckpointer.removeCheckpointer(shardId, checkpointer)
   }
diff --git a/connector/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisRecordProcessor.scala b/connector/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisRecordProcessor.scala
index b35573e92e16..5644c32553b0 100644
--- a/connector/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisRecordProcessor.scala
+++ b/connector/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisRecordProcessor.scala
@@ -16,15 +16,12 @@
  */
 package org.apache.spark.streaming.kinesis
 
-import java.util.List
-
 import scala.util.Random
 import scala.util.control.NonFatal
 
-import com.amazonaws.services.kinesis.clientlibrary.exceptions.{InvalidStateException, KinesisClientLibDependencyException, ShutdownException, ThrottlingException}
-import com.amazonaws.services.kinesis.clientlibrary.interfaces.{IRecordProcessor, IRecordProcessorCheckpointer}
-import com.amazonaws.services.kinesis.clientlibrary.lib.worker.ShutdownReason
-import com.amazonaws.services.kinesis.model.Record
+import software.amazon.kinesis.exceptions.{InvalidStateException, KinesisClientLibDependencyException, ShutdownException, ThrottlingException}
+import software.amazon.kinesis.lifecycle.events.{InitializationInput, LeaseLostInput, ProcessRecordsInput, ShardEndedInput, ShutdownRequestedInput}
+import software.amazon.kinesis.processor.ShardRecordProcessor
 
 import org.apache.spark.internal.Logging
 
@@ -40,7 +37,7 @@ import org.apache.spark.internal.Logging
  * @param workerId for logging purposes
  */
 private[kinesis] class KinesisRecordProcessor[T](receiver: KinesisReceiver[T], workerId: String)
-  extends IRecordProcessor with Logging {
+  extends ShardRecordProcessor with Logging {
 
   // shardId populated during initialize()
   @volatile
@@ -51,22 +48,19 @@ private[kinesis] class KinesisRecordProcessor[T](receiver: KinesisReceiver[T], w
    *
    * @param shardId assigned by the KCL to this particular RecordProcessor.
    */
-  override def initialize(shardId: String): Unit = {
+  def initialize(shardId: String): Unit = {
     this.shardId = shardId
     logInfo(s"Initialized workerId $workerId with shardId $shardId")
   }
 
-  /**
-   * This method is called by the KCL when a batch of records is pulled from the Kinesis stream.
-   * This is the record-processing bridge between the KCL's IRecordProcessor.processRecords()
-   * and Spark Streaming's Receiver.store().
-   *
-   * @param batch list of records from the Kinesis stream shard
-   * @param checkpointer used to update Kinesis when this batch has been processed/stored
-   *   in the DStream
-   */
-  override def processRecords(batch: List[Record],
-      checkpointer: IRecordProcessorCheckpointer): Unit = {
+  override def initialize(initializationInput: InitializationInput): Unit = {
+    this.shardId = initializationInput.shardId()
+    logInfo(s"Initialized workerId $workerId with shardId $shardId")
+  }
+
+  override def processRecords(processRecordsInput: ProcessRecordsInput): Unit = {
+    val batch = processRecordsInput.records()
+    val checkpointer = processRecordsInput.checkpointer()
     if (!receiver.isStopped()) {
       try {
         // Limit the number of processed records from Kinesis stream. This is because the KCL cannot
@@ -90,7 +84,7 @@ private[kinesis] class KinesisRecordProcessor[T](receiver: KinesisReceiver[T], w
            *     more than once.
            */
           logError(s"Exception:  WorkerId $workerId encountered and exception while storing " +
-              s" or checkpointing a batch for workerId $workerId and shardId $shardId.", e)
+            s" or checkpointing a batch for workerId $workerId and shardId $shardId.", e)
 
           /* Rethrow the exception to the Kinesis Worker that is managing this RecordProcessor. */
           throw e
@@ -98,45 +92,30 @@ private[kinesis] class KinesisRecordProcessor[T](receiver: KinesisReceiver[T], w
     } else {
       /* RecordProcessor has been stopped. */
       logInfo(s"Stopped:  KinesisReceiver has stopped for workerId $workerId" +
-          s" and shardId $shardId.  No more records will be processed.")
+        s" and shardId $shardId.  No more records will be processed.")
     }
   }
 
-  /**
-   * Kinesis Client Library is shutting down this Worker for 1 of 2 reasons:
-   * 1) the stream is resharding by splitting or merging adjacent shards
-   *     (ShutdownReason.TERMINATE)
-   * 2) the failed or latent Worker has stopped sending heartbeats for whatever reason
-   *     (ShutdownReason.ZOMBIE)
-   *
-   * @param checkpointer used to perform a Kinesis checkpoint for ShutdownReason.TERMINATE
-   * @param reason for shutdown (ShutdownReason.TERMINATE or ShutdownReason.ZOMBIE)
-   */
-  override def shutdown(
-      checkpointer: IRecordProcessorCheckpointer,
-      reason: ShutdownReason): Unit = {
-    logInfo(s"Shutdown:  Shutting down workerId $workerId with reason $reason")
-    // null if not initialized before shutdown:
-    if (shardId == null) {
-      logWarning(s"No shardId for workerId $workerId?")
-    } else {
-      reason match {
-        /*
-         * TERMINATE Use Case.  Checkpoint.
-         * Checkpoint to indicate that all records from the shard have been drained and processed.
-         * It's now OK to read from the new shards that resulted from a resharding event.
-         */
-        case ShutdownReason.TERMINATE => receiver.removeCheckpointer(shardId, checkpointer)
+  override def leaseLost(leaseLostInput: LeaseLostInput): Unit = {
+  }
 
-        /*
-         * ZOMBIE Use Case or Unknown reason.  NoOp.
-         * No checkpoint because other workers may have taken over and already started processing
-         *    the same records.
-         * This may lead to records being processed more than once.
-         * Return null so that we don't checkpoint
-         */
-        case _ => receiver.removeCheckpointer(shardId, null)
-      }
+  override def shardEnded(shardEndedInput: ShardEndedInput): Unit = {
+    try {
+      log.info("Reached shard end checkpointing. shardId: {}", shardId)
+      shardEndedInput.checkpointer.checkpoint()
+    } catch {
+      case e @ (_: ShutdownException | _: InvalidStateException) =>
+        log.error(s"Exception while checkpointing at shard end. Giving up. shardId: $shardId", e)
+    }
+  }
+
+  override def shutdownRequested(shutdownRequestedInput: ShutdownRequestedInput): Unit = {
+    try {
+      log.info("Scheduler is shutting down, checkpointing.")
+      shutdownRequestedInput.checkpointer.checkpoint()
+    } catch {
+      case e @ (_: ShutdownException | _: InvalidStateException) =>
+        log.error("Exception while checkpointing at requested shutdown. Giving up.", e)
     }
   }
 }
diff --git a/connector/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisTestUtils.scala b/connector/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisTestUtils.scala
index 4dacc03c7271..b50315f830a8 100644
--- a/connector/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisTestUtils.scala
+++ b/connector/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisTestUtils.scala
@@ -17,6 +17,7 @@
 
 package org.apache.spark.streaming.kinesis
 
+import java.net.URI
 import java.nio.ByteBuffer
 import java.nio.charset.StandardCharsets
 import java.util.concurrent.TimeUnit
@@ -26,12 +27,15 @@ import scala.collection.mutable
 import scala.collection.mutable.ArrayBuffer
 import scala.util.{Failure, Random, Success, Try}
 
-import com.amazonaws.auth.{AWSCredentials, DefaultAWSCredentialsProviderChain}
-import com.amazonaws.regions.RegionUtils
-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 software.amazon.awssdk.auth.credentials.{AwsCredentials, DefaultCredentialsProvider}
+import software.amazon.awssdk.core.SdkBytes
+import software.amazon.awssdk.http.apache.ApacheHttpClient
+import software.amazon.awssdk.regions.Region
+import software.amazon.awssdk.regions.servicemetadata.KinesisServiceMetadata
+import software.amazon.awssdk.services.dynamodb.DynamoDbClient
+import software.amazon.awssdk.services.dynamodb.model.DeleteTableRequest
+import software.amazon.awssdk.services.kinesis.KinesisClient
+import software.amazon.awssdk.services.kinesis.model.{CreateStreamRequest, DeleteStreamRequest, DescribeStreamRequest, MergeShardsRequest, PutRecordRequest, ResourceNotFoundException, Shard, SplitShardRequest, StreamDescription}
 
 import org.apache.spark.internal.Logging
 
@@ -54,16 +58,22 @@ private[kinesis] class KinesisTestUtils(streamShardCount: Int = 2) extends Loggi
   @volatile
   private var _streamName: String = _
 
-  protected lazy val kinesisClient = {
-    val client = new AmazonKinesisClient(KinesisTestUtils.getAWSCredentials())
-    client.setEndpoint(endpointUrl)
+  protected lazy val kinesisClient: KinesisClient = {
+    val client = KinesisClient.builder()
+      .credentialsProvider(DefaultCredentialsProvider.create())
+      .region(Region.of(regionName))
+      .httpClientBuilder(ApacheHttpClient.builder().maxConnections(Integer.MAX_VALUE))
+      .endpointOverride(URI.create(endpointUrl))
+      .build()
     client
   }
 
   private lazy val dynamoDB = {
-    val dynamoDBClient = new AmazonDynamoDBClient(new DefaultAWSCredentialsProviderChain())
-    dynamoDBClient.setRegion(RegionUtils.getRegion(regionName))
-    new DynamoDB(dynamoDBClient)
+    val dynamoDBClient = DynamoDbClient.builder()
+      .credentialsProvider(DefaultCredentialsProvider.create())
+      .region(Region.of(regionName))
+      .build()
+    dynamoDBClient
   }
 
   protected def getProducer(aggregate: Boolean): KinesisDataGenerator = {
@@ -85,9 +95,10 @@ private[kinesis] class KinesisTestUtils(streamShardCount: Int = 2) extends Loggi
 
     // Create a stream. The number of shards determines the provisioned throughput.
     logInfo(s"Creating stream ${_streamName}")
-    val createStreamRequest = new CreateStreamRequest()
-    createStreamRequest.setStreamName(_streamName)
-    createStreamRequest.setShardCount(streamShardCount)
+    val createStreamRequest = CreateStreamRequest.builder()
+      .streamName(_streamName)
+      .shardCount(streamShardCount)
+      .build()
     kinesisClient.createStream(createStreamRequest)
 
     // The stream is now being created. Wait for it to become active.
@@ -96,26 +107,30 @@ private[kinesis] class KinesisTestUtils(streamShardCount: Int = 2) extends Loggi
     logInfo(s"Created stream ${_streamName}")
   }
 
-  def getShards(): Seq[Shard] = {
-    kinesisClient.describeStream(_streamName).getStreamDescription.getShards.asScala.toSeq
+  def getShards: Seq[Shard] = {
+    val describeStreamRequest = DescribeStreamRequest.builder()
+      .streamName(_streamName)
+      .build()
+    kinesisClient.describeStream(describeStreamRequest).streamDescription().shards.asScala.toSeq
   }
 
   def splitShard(shardId: String): Unit = {
-    val splitShardRequest = new SplitShardRequest()
-    splitShardRequest.withStreamName(_streamName)
-    splitShardRequest.withShardToSplit(shardId)
-    // Set a half of the max hash value
-    splitShardRequest.withNewStartingHashKey("170141183460469231731687303715884105728")
+    val splitShardRequest = SplitShardRequest.builder()
+      .streamName(_streamName)
+      .shardToSplit(shardId)
+      .newStartingHashKey("170141183460469231731687303715884105728")
+      .build()
     kinesisClient.splitShard(splitShardRequest)
     // Wait for the shards to become active
     waitForStreamToBeActive(_streamName)
   }
 
   def mergeShard(shardToMerge: String, adjacentShardToMerge: String): Unit = {
-    val mergeShardRequest = new MergeShardsRequest
-    mergeShardRequest.withStreamName(_streamName)
-    mergeShardRequest.withShardToMerge(shardToMerge)
-    mergeShardRequest.withAdjacentShardToMerge(adjacentShardToMerge)
+    val mergeShardRequest = MergeShardsRequest.builder()
+      .streamName(_streamName)
+      .shardToMerge(shardToMerge)
+      .adjacentShardToMerge(adjacentShardToMerge)
+      .build()
     kinesisClient.mergeShards(mergeShardRequest)
     // Wait for the shards to become active
     waitForStreamToBeActive(_streamName)
@@ -141,34 +156,41 @@ private[kinesis] class KinesisTestUtils(streamShardCount: Int = 2) extends Loggi
   }
 
   def deleteStream(): Unit = {
+    val deleteStreamRequest = DeleteStreamRequest.builder()
+      .streamName(streamName)
+      .build()
     try {
       if (streamCreated) {
-        kinesisClient.deleteStream(streamName)
+        kinesisClient.deleteStream(deleteStreamRequest)
       }
     } catch {
       case e: Exception =>
-        logWarning(s"Could not delete stream $streamName")
+        logWarning(s"Could not delete stream $streamName", e)
     }
   }
 
   def deleteDynamoDBTable(tableName: String): Unit = {
+    val deleteTableRequest = DeleteTableRequest.builder()
+      .tableName(tableName)
+      .build()
     try {
-      val table = dynamoDB.getTable(tableName)
-      table.delete()
-      table.waitForDelete()
+      dynamoDB.deleteTable(deleteTableRequest)
     } catch {
       case e: Exception =>
-        logWarning(s"Could not delete DynamoDB table $tableName")
+        logWarning(s"Could not delete DynamoDB table $tableName", e)
     }
   }
 
   private def describeStream(streamNameToDescribe: String): Option[StreamDescription] = {
     try {
-      val describeStreamRequest = new DescribeStreamRequest().withStreamName(streamNameToDescribe)
-      val desc = kinesisClient.describeStream(describeStreamRequest).getStreamDescription()
+      val describeStreamRequest = DescribeStreamRequest.builder()
+        .streamName(streamNameToDescribe)
+        .build()
+      val desc = kinesisClient.describeStream(describeStreamRequest).streamDescription()
       Some(desc)
     } catch {
       case rnfe: ResourceNotFoundException =>
+        logWarning(s"Could not describe stream $streamNameToDescribe", rnfe)
         None
     }
   }
@@ -187,9 +209,9 @@ private[kinesis] class KinesisTestUtils(streamShardCount: Int = 2) extends Loggi
     while (System.nanoTime() - startTimeNs < TimeUnit.SECONDS.toNanos(createStreamTimeoutSeconds)) {
       Thread.sleep(TimeUnit.SECONDS.toMillis(describeStreamPollTimeSeconds))
       describeStream(streamNameToWaitFor).foreach { description =>
-        val streamStatus = description.getStreamStatus()
+        val streamStatus = description.streamStatus()
         logDebug(s"\t- current state: $streamStatus\n")
-        if ("ACTIVE".equals(streamStatus)) {
+        if ("ACTIVE".equals(streamStatus.toString)) {
           return
         }
       }
@@ -206,10 +228,11 @@ private[kinesis] object KinesisTestUtils {
 
   def getRegionNameByEndpoint(endpoint: String): String = {
     val uri = new java.net.URI(endpoint)
-    RegionUtils.getRegionsForService(AmazonKinesis.ENDPOINT_PREFIX)
+    val kinesisServiceMetadata = new KinesisServiceMetadata()
+    kinesisServiceMetadata.regions()
       .asScala
-      .find(_.getAvailableEndpoints.asScala.toSeq.contains(uri.getHost))
-      .map(_.getName)
+      .find(r => kinesisServiceMetadata.endpointFor(r).toString.equals(uri.getHost))
+      .map(_.id())
       .getOrElse(
         throw new IllegalArgumentException(s"Could not resolve region for endpoint: $endpoint"))
   }
@@ -244,20 +267,20 @@ private[kinesis] object KinesisTestUtils {
   }
 
   def isAWSCredentialsPresent: Boolean = {
-    Try { new DefaultAWSCredentialsProviderChain().getCredentials() }.isSuccess
+    Try { DefaultCredentialsProvider.create().resolveCredentials() }.isSuccess
   }
 
-  def getAWSCredentials(): AWSCredentials = {
+  def getAwsCredentials: AwsCredentials = {
     assert(shouldRunTests,
       "Kinesis test not enabled, should not attempt to get AWS credentials")
-    Try { new DefaultAWSCredentialsProviderChain().getCredentials() } match {
+    Try { DefaultCredentialsProvider.create().resolveCredentials() } match {
       case Success(cred) => cred
       case Failure(e) =>
         throw new Exception(
           s"""
              |Kinesis tests enabled using environment variable $envVarNameForEnablingTests
              |but could not find AWS credentials. Please follow instructions in AWS documentation
-             |to set the credentials in your system such that the DefaultAWSCredentialsProviderChain
+             |to set the credentials in your system such that the DefaultCredentialsProvider
              |can find the credentials.
            """.stripMargin)
     }
@@ -271,19 +294,21 @@ private[kinesis] trait KinesisDataGenerator {
 }
 
 private[kinesis] class SimpleDataGenerator(
-    client: AmazonKinesisClient) extends KinesisDataGenerator {
+    client: KinesisClient) 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 =>
       val str = num.toString
       val data = ByteBuffer.wrap(str.getBytes(StandardCharsets.UTF_8))
-      val putRecordRequest = new PutRecordRequest().withStreamName(streamName)
-        .withData(data)
-        .withPartitionKey(str)
-
-      val putRecordResult = client.putRecord(putRecordRequest)
-      val shardId = putRecordResult.getShardId
-      val seqNumber = putRecordResult.getSequenceNumber()
+      val putRecordRequest = PutRecordRequest.builder()
+        .streamName(streamName)
+        .data(SdkBytes.fromByteBuffer(data))
+        .partitionKey(str)
+        .build()
+
+      val putRecordResponse = client.putRecord(putRecordRequest)
+      val shardId = putRecordResponse.shardId()
+      val seqNumber = putRecordResponse.sequenceNumber()
       val sentSeqNumbers = shardIdToSeqNumbers.getOrElseUpdate(shardId,
         new ArrayBuffer[(Int, String)]())
       sentSeqNumbers += ((num, seqNumber))
diff --git a/connector/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisUtilsPythonHelper.scala b/connector/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisUtilsPythonHelper.scala
index 8abaef6b834e..dc1098a33633 100644
--- a/connector/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisUtilsPythonHelper.scala
+++ b/connector/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisUtilsPythonHelper.scala
@@ -16,8 +16,8 @@
  */
 package org.apache.spark.streaming.kinesis
 
-import com.amazonaws.services.kinesis.clientlibrary.lib.worker.InitialPositionInStream
-import com.amazonaws.services.kinesis.metrics.interfaces.MetricsLevel
+import software.amazon.kinesis.common.InitialPositionInStream
+import software.amazon.kinesis.metrics.MetricsLevel
 
 import org.apache.spark.storage.StorageLevel
 import org.apache.spark.streaming.Duration
diff --git a/connector/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/SparkAWSCredentials.scala b/connector/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/SparkAWSCredentials.scala
index e821adca20d2..234e128b2fa9 100644
--- a/connector/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/SparkAWSCredentials.scala
+++ b/connector/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/SparkAWSCredentials.scala
@@ -17,30 +17,33 @@
 
 package org.apache.spark.streaming.kinesis
 
-import com.amazonaws.auth._
+import software.amazon.awssdk.auth.credentials.{AwsBasicCredentials, AwsCredentialsProvider, DefaultCredentialsProvider, StaticCredentialsProvider}
+import software.amazon.awssdk.services.sts.StsClient
+import software.amazon.awssdk.services.sts.auth.StsAssumeRoleCredentialsProvider
+import software.amazon.awssdk.services.sts.model.AssumeRoleRequest
 
 import org.apache.spark.internal.Logging
 
 /**
  * Serializable interface providing a method executors can call to obtain an
- * AWSCredentialsProvider instance for authenticating to AWS services.
+ * AwsCredentialsProvider instance for authenticating to AWS services.
  */
 private[kinesis] sealed trait SparkAWSCredentials extends Serializable {
   /**
    * Return an AWSCredentialProvider instance that can be used by the Kinesis Client
    * Library to authenticate to AWS services (Kinesis, CloudWatch and DynamoDB).
    */
-  def provider: AWSCredentialsProvider
+  def provider: AwsCredentialsProvider
 }
 
-/** Returns DefaultAWSCredentialsProviderChain for authentication. */
+/** Returns DefaultCredentialsProvider for authentication. */
 private[kinesis] final case object DefaultCredentials extends SparkAWSCredentials {
 
-  def provider: AWSCredentialsProvider = new DefaultAWSCredentialsProviderChain
+  def provider: AwsCredentialsProvider = DefaultCredentialsProvider.create()
 }
 
 /**
- * Returns AWSStaticCredentialsProvider constructed using basic AWS keypair. Falls back to using
+ * Returns StaticCredentialsProvider constructed using basic AWS keypair. Falls back to using
  * DefaultCredentialsProviderChain if unable to construct a AWSCredentialsProviderChain
  * instance with the provided arguments (e.g. if they are null).
  */
@@ -48,13 +51,13 @@ private[kinesis] final case class BasicCredentials(
     awsAccessKeyId: String,
     awsSecretKey: String) extends SparkAWSCredentials with Logging {
 
-  def provider: AWSCredentialsProvider = try {
-    new AWSStaticCredentialsProvider(new BasicAWSCredentials(awsAccessKeyId, awsSecretKey))
+  def provider: AwsCredentialsProvider = try {
+    StaticCredentialsProvider.create(AwsBasicCredentials.create(awsAccessKeyId, awsSecretKey))
   } catch {
     case e: IllegalArgumentException =>
-      logWarning("Unable to construct AWSStaticCredentialsProvider with provided keypair; " +
+      logWarning("Unable to construct StaticCredentialsProvider with provided keypair; " +
         "falling back to DefaultCredentialsProviderChain.", e)
-      new DefaultAWSCredentialsProviderChain
+      DefaultCredentialsProvider.create()
   }
 }
 
@@ -69,16 +72,24 @@ private[kinesis] final case class STSCredentials(
     longLivedCreds: SparkAWSCredentials = DefaultCredentials)
   extends SparkAWSCredentials  {
 
-  def provider: AWSCredentialsProvider = {
-    val builder = new STSAssumeRoleSessionCredentialsProvider.Builder(stsRoleArn, stsSessionName)
-      .withLongLivedCredentialsProvider(longLivedCreds.provider)
+  def provider: AwsCredentialsProvider = {
+    val stsClient = StsClient.builder()
+      .credentialsProvider(longLivedCreds.provider)
+      .build()
+
+    val assumeRoleRequestBuilder = AssumeRoleRequest.builder()
+      .roleArn(stsRoleArn)
+      .roleSessionName(stsSessionName)
     stsExternalId match {
       case Some(stsExternalId) =>
-        builder.withExternalId(stsExternalId)
-          .build()
+        assumeRoleRequestBuilder.externalId(stsExternalId)
       case None =>
-        builder.build()
     }
+
+    StsAssumeRoleCredentialsProvider.builder()
+      .stsClient(stsClient)
+      .refreshRequest(assumeRoleRequestBuilder.build())
+      .build()
   }
 }
 
diff --git a/connector/kinesis-asl/src/test/java/org/apache/spark/streaming/kinesis/JavaKinesisInputDStreamBuilderSuite.java b/connector/kinesis-asl/src/test/java/org/apache/spark/streaming/kinesis/JavaKinesisInputDStreamBuilderSuite.java
index 7af0abe0e8d9..b9e06d52c9c2 100644
--- a/connector/kinesis-asl/src/test/java/org/apache/spark/streaming/kinesis/JavaKinesisInputDStreamBuilderSuite.java
+++ b/connector/kinesis-asl/src/test/java/org/apache/spark/streaming/kinesis/JavaKinesisInputDStreamBuilderSuite.java
@@ -17,7 +17,6 @@
 
 package org.apache.spark.streaming.kinesis;
 
-import com.amazonaws.services.kinesis.clientlibrary.lib.worker.InitialPositionInStream;
 import org.junit.Assert;
 import org.junit.Test;
 
@@ -26,6 +25,7 @@
 import org.apache.spark.streaming.Duration;
 import org.apache.spark.streaming.LocalJavaStreamingContext;
 import org.apache.spark.streaming.Seconds;
+import software.amazon.kinesis.common.InitialPositionInStream;
 
 public class JavaKinesisInputDStreamBuilderSuite extends LocalJavaStreamingContext {
   /**
diff --git a/connector/kinesis-asl/src/test/scala/org/apache/spark/streaming/kinesis/KinesisCheckpointerSuite.scala b/connector/kinesis-asl/src/test/scala/org/apache/spark/streaming/kinesis/KinesisCheckpointerSuite.scala
index 87592b6877b3..1ff356193a69 100644
--- a/connector/kinesis-asl/src/test/scala/org/apache/spark/streaming/kinesis/KinesisCheckpointerSuite.scala
+++ b/connector/kinesis-asl/src/test/scala/org/apache/spark/streaming/kinesis/KinesisCheckpointerSuite.scala
@@ -22,13 +22,13 @@ import java.util.concurrent.TimeoutException
 import scala.concurrent.{Await, ExecutionContext, Future}
 import scala.concurrent.duration._
 
-import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorCheckpointer
 import org.mockito.ArgumentMatchers._
 import org.mockito.Mockito._
 import org.mockito.invocation.InvocationOnMock
 import org.scalatest.{BeforeAndAfterEach, PrivateMethodTester}
 import org.scalatest.concurrent.Eventually
 import org.scalatestplus.mockito.MockitoSugar
+import software.amazon.kinesis.processor.RecordProcessorCheckpointer
 
 import org.apache.spark.streaming.{Duration, TestSuiteBase}
 import org.apache.spark.util.ManualClock
@@ -48,7 +48,7 @@ class KinesisCheckpointerSuite extends TestSuiteBase
   private val someOtherSeqNum = Some(otherSeqNum)
 
   private var receiverMock: KinesisReceiver[Array[Byte]] = _
-  private var checkpointerMock: IRecordProcessorCheckpointer = _
+  private var checkpointerMock: RecordProcessorCheckpointer = _
   private var kinesisCheckpointer: KinesisCheckpointer = _
   private var clock: ManualClock = _
 
@@ -56,7 +56,7 @@ class KinesisCheckpointerSuite extends TestSuiteBase
 
   override def beforeEach(): Unit = {
     receiverMock = mock[KinesisReceiver[Array[Byte]]]
-    checkpointerMock = mock[IRecordProcessorCheckpointer]
+    checkpointerMock = mock[RecordProcessorCheckpointer]
     clock = new ManualClock()
     kinesisCheckpointer = new KinesisCheckpointer(receiverMock, checkpointInterval, workerId, clock)
   }
diff --git a/connector/kinesis-asl/src/test/scala/org/apache/spark/streaming/kinesis/KinesisInputDStreamBuilderSuite.scala b/connector/kinesis-asl/src/test/scala/org/apache/spark/streaming/kinesis/KinesisInputDStreamBuilderSuite.scala
index babdb15987a6..08c1d1a2d5a4 100644
--- a/connector/kinesis-asl/src/test/scala/org/apache/spark/streaming/kinesis/KinesisInputDStreamBuilderSuite.scala
+++ b/connector/kinesis-asl/src/test/scala/org/apache/spark/streaming/kinesis/KinesisInputDStreamBuilderSuite.scala
@@ -21,10 +21,10 @@ import java.util.Calendar
 
 import scala.collection.JavaConverters._
 
-import com.amazonaws.services.kinesis.clientlibrary.lib.worker.{InitialPositionInStream, KinesisClientLibConfiguration}
-import com.amazonaws.services.kinesis.metrics.interfaces.MetricsLevel
 import org.scalatest.BeforeAndAfterEach
 import org.scalatestplus.mockito.MockitoSugar
+import software.amazon.kinesis.common.InitialPositionInStream
+import software.amazon.kinesis.metrics.{MetricsConfig, MetricsLevel}
 
 import org.apache.spark.storage.StorageLevel
 import org.apache.spark.streaming.{Duration, Seconds, StreamingContext, TestSuiteBase}
@@ -101,7 +101,7 @@ class KinesisInputDStreamBuilderSuite extends TestSuiteBase with BeforeAndAfterE
     val customCloudWatchCreds = mock[SparkAWSCredentials]
     val customMetricsLevel = MetricsLevel.NONE
     val customMetricsEnabledDimensions =
-      KinesisClientLibConfiguration.METRICS_ALWAYS_ENABLED_DIMENSIONS.asScala.toSet
+      MetricsConfig.METRICS_ALWAYS_ENABLED_DIMENSIONS.asScala.toSet
 
     val dstream = builder
       .endpointUrl(customEndpointUrl)
diff --git a/connector/kinesis-asl/src/test/scala/org/apache/spark/streaming/kinesis/KinesisReceiverSuite.scala b/connector/kinesis-asl/src/test/scala/org/apache/spark/streaming/kinesis/KinesisReceiverSuite.scala
index d008de3b3f1c..74c813fc5d40 100644
--- a/connector/kinesis-asl/src/test/scala/org/apache/spark/streaming/kinesis/KinesisReceiverSuite.scala
+++ b/connector/kinesis-asl/src/test/scala/org/apache/spark/streaming/kinesis/KinesisReceiverSuite.scala
@@ -20,16 +20,16 @@ import java.nio.ByteBuffer
 import java.nio.charset.StandardCharsets
 import java.util.Arrays
 
-import com.amazonaws.services.kinesis.clientlibrary.exceptions._
-import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorCheckpointer
-import com.amazonaws.services.kinesis.clientlibrary.lib.worker.ShutdownReason
-import com.amazonaws.services.kinesis.model.Record
 import org.mockito.ArgumentMatchers.{anyList, anyString, eq => meq}
 import org.mockito.Mockito.{never, times, verify, when}
 import org.scalatest.BeforeAndAfter
 import org.scalatest.matchers.must.Matchers
 import org.scalatest.matchers.should.Matchers._
 import org.scalatestplus.mockito.MockitoSugar
+import software.amazon.kinesis.exceptions.{InvalidStateException, KinesisClientLibDependencyException, ShutdownException, ThrottlingException}
+import software.amazon.kinesis.lifecycle.events.{ProcessRecordsInput, ShutdownRequestedInput}
+import software.amazon.kinesis.processor.RecordProcessorCheckpointer
+import software.amazon.kinesis.retrieval.KinesisClientRecord
 
 import org.apache.spark.streaming.{Duration, TestSuiteBase}
 
@@ -48,18 +48,20 @@ class KinesisReceiverSuite extends TestSuiteBase with Matchers with BeforeAndAft
   val checkpointInterval = Duration(10)
   val someSeqNum = Some(seqNum)
 
-  val record1 = new Record()
-  record1.setData(ByteBuffer.wrap("Spark In Action".getBytes(StandardCharsets.UTF_8)))
-  val record2 = new Record()
-  record2.setData(ByteBuffer.wrap("Learning Spark".getBytes(StandardCharsets.UTF_8)))
+  val record1 = KinesisClientRecord.builder()
+    .data(ByteBuffer.wrap("Spark In Action".getBytes(StandardCharsets.UTF_8)))
+    .build()
+  val record2 = KinesisClientRecord.builder()
+    .data(ByteBuffer.wrap("Learning Spark".getBytes(StandardCharsets.UTF_8)))
+    .build()
   val batch = Arrays.asList(record1, record2)
 
   var receiverMock: KinesisReceiver[Array[Byte]] = _
-  var checkpointerMock: IRecordProcessorCheckpointer = _
+  var checkpointerMock: RecordProcessorCheckpointer = _
 
   override def beforeFunction(): Unit = {
     receiverMock = mock[KinesisReceiver[Array[Byte]]]
-    checkpointerMock = mock[IRecordProcessorCheckpointer]
+    checkpointerMock = mock[RecordProcessorCheckpointer]
   }
 
   test("process records including store and set checkpointer") {
@@ -68,7 +70,11 @@ class KinesisReceiverSuite extends TestSuiteBase with Matchers with BeforeAndAft
 
     val recordProcessor = new KinesisRecordProcessor(receiverMock, workerId)
     recordProcessor.initialize(shardId)
-    recordProcessor.processRecords(batch, checkpointerMock)
+    val processRecordsInput = ProcessRecordsInput.builder()
+      .records(batch)
+      .checkpointer(checkpointerMock)
+      .build()
+    recordProcessor.processRecords(processRecordsInput)
 
     verify(receiverMock, times(1)).isStopped()
     verify(receiverMock, times(1)).addRecords(shardId, batch)
@@ -81,7 +87,11 @@ class KinesisReceiverSuite extends TestSuiteBase with Matchers with BeforeAndAft
 
     val recordProcessor = new KinesisRecordProcessor(receiverMock, workerId)
     recordProcessor.initialize(shardId)
-    recordProcessor.processRecords(batch, checkpointerMock)
+    val processRecordsInput = ProcessRecordsInput.builder()
+      .records(batch)
+      .checkpointer(checkpointerMock)
+      .build()
+    recordProcessor.processRecords(processRecordsInput)
 
     verify(receiverMock, times(1)).isStopped()
     verify(receiverMock, times(1)).addRecords(shardId, batch.subList(0, 1))
@@ -94,7 +104,11 @@ class KinesisReceiverSuite extends TestSuiteBase with Matchers with BeforeAndAft
     when(receiverMock.getCurrentLimit).thenReturn(Int.MaxValue)
 
     val recordProcessor = new KinesisRecordProcessor(receiverMock, workerId)
-    recordProcessor.processRecords(batch, checkpointerMock)
+    val processRecordsInput = ProcessRecordsInput.builder()
+      .records(batch)
+      .checkpointer(checkpointerMock)
+      .build()
+    recordProcessor.processRecords(processRecordsInput)
 
     verify(receiverMock, times(1)).isStopped()
     verify(receiverMock, never).addRecords(anyString, anyList())
@@ -111,7 +125,11 @@ class KinesisReceiverSuite extends TestSuiteBase with Matchers with BeforeAndAft
     intercept[RuntimeException] {
       val recordProcessor = new KinesisRecordProcessor(receiverMock, workerId)
       recordProcessor.initialize(shardId)
-      recordProcessor.processRecords(batch, checkpointerMock)
+      val processRecordsInput = ProcessRecordsInput.builder()
+        .records(batch)
+        .checkpointer(checkpointerMock)
+        .build()
+      recordProcessor.processRecords(processRecordsInput)
     }
 
     verify(receiverMock, times(1)).isStopped()
@@ -119,27 +137,14 @@ class KinesisReceiverSuite extends TestSuiteBase with Matchers with BeforeAndAft
     verify(receiverMock, never).setCheckpointer(anyString, meq(checkpointerMock))
   }
 
-  test("shutdown should checkpoint if the reason is TERMINATE") {
-    when(receiverMock.getLatestSeqNumToCheckpoint(shardId)).thenReturn(someSeqNum)
-
-    val recordProcessor = new KinesisRecordProcessor(receiverMock, workerId)
-    recordProcessor.initialize(shardId)
-    recordProcessor.shutdown(checkpointerMock, ShutdownReason.TERMINATE)
-
-    verify(receiverMock, times(1)).removeCheckpointer(meq(shardId), meq(checkpointerMock))
-  }
-
-
-  test("shutdown should not checkpoint if the reason is something other than TERMINATE") {
-    when(receiverMock.getLatestSeqNumToCheckpoint(shardId)).thenReturn(someSeqNum)
-
+  test("should checkpoint when the method shutdownRequested is call") {
     val recordProcessor = new KinesisRecordProcessor(receiverMock, workerId)
-    recordProcessor.initialize(shardId)
-    recordProcessor.shutdown(checkpointerMock, ShutdownReason.ZOMBIE)
-    recordProcessor.shutdown(checkpointerMock, null)
+    val shutdownRequestedInput = ShutdownRequestedInput.builder()
+      .checkpointer(checkpointerMock)
+      .build()
+    recordProcessor.shutdownRequested(shutdownRequestedInput)
 
-    verify(receiverMock, times(2)).removeCheckpointer(meq(shardId),
-      meq[IRecordProcessorCheckpointer](null))
+    verify(checkpointerMock, times(1)).checkpoint()
   }
 
   test("retry success on first attempt") {
diff --git a/connector/kinesis-asl/src/test/scala/org/apache/spark/streaming/kinesis/KinesisStreamSuite.scala b/connector/kinesis-asl/src/test/scala/org/apache/spark/streaming/kinesis/KinesisStreamSuite.scala
index 43c4118d8f59..fc9ddf1cfe02 100644
--- a/connector/kinesis-asl/src/test/scala/org/apache/spark/streaming/kinesis/KinesisStreamSuite.scala
+++ b/connector/kinesis-asl/src/test/scala/org/apache/spark/streaming/kinesis/KinesisStreamSuite.scala
@@ -21,10 +21,10 @@ import scala.collection.mutable
 import scala.concurrent.duration._
 import scala.util.Random
 
-import com.amazonaws.services.kinesis.model.Record
 import org.scalatest.{BeforeAndAfter, BeforeAndAfterAll}
 import org.scalatest.concurrent.Eventually
 import org.scalatest.matchers.should.Matchers._
+import software.amazon.kinesis.retrieval.KinesisClientRecord
 
 import org.apache.spark.{SparkConf, SparkContext}
 import org.apache.spark.network.util.JavaUtils
@@ -195,7 +195,7 @@ abstract class KinesisStreamTests(aggregateTestData: Boolean) extends KinesisFun
   }
 
   testIfEnabled("custom message handling") {
-    def addFive(r: Record): Int = JavaUtils.bytesToString(r.getData).toInt + 5
+    def addFive(r: KinesisClientRecord): Int = JavaUtils.bytesToString(r.data()).toInt + 5
 
     val stream = KinesisInputDStream.builder.streamingContext(ssc)
       .checkpointAppName(appName)
@@ -305,17 +305,17 @@ abstract class KinesisStreamTests(aggregateTestData: Boolean) extends KinesisFun
       val testData2 = 11 to 20
       val testData3 = 21 to 30
 
-      eventually(timeout(1.minute), interval(10.seconds)) {
+      eventually(timeout(2.minute), interval(10.seconds)) {
         localTestUtils.pushData(testData1, aggregateTestData)
         collected.synchronized {
           assert(collected === testData1.toSet, "\nData received does not match data sent")
         }
       }
 
-      val shardToSplit = localTestUtils.getShards().head
-      localTestUtils.splitShard(shardToSplit.getShardId)
-      val (splitOpenShards, splitCloseShards) = localTestUtils.getShards().partition { shard =>
-        shard.getSequenceNumberRange.getEndingSequenceNumber == null
+      val shardToSplit = localTestUtils.getShards.head
+      localTestUtils.splitShard(shardToSplit.shardId())
+      val (splitOpenShards, splitCloseShards) = localTestUtils.getShards.partition { shard =>
+        shard.sequenceNumberRange().endingSequenceNumber() == null
       }
 
       // We should have one closed shard and two open shards
@@ -331,9 +331,9 @@ abstract class KinesisStreamTests(aggregateTestData: Boolean) extends KinesisFun
       }
 
       val Seq(shardToMerge, adjShard) = splitOpenShards
-      localTestUtils.mergeShard(shardToMerge.getShardId, adjShard.getShardId)
-      val (mergedOpenShards, mergedCloseShards) = localTestUtils.getShards().partition { shard =>
-        shard.getSequenceNumberRange.getEndingSequenceNumber == null
+      localTestUtils.mergeShard(shardToMerge.shardId(), adjShard.shardId())
+      val (mergedOpenShards, mergedCloseShards) = localTestUtils.getShards.partition { shard =>
+        shard.sequenceNumberRange().endingSequenceNumber() == null
       }
 
       // We should have three closed shards and one open shard
diff --git a/docs/streaming-kinesis-integration.md b/docs/streaming-kinesis-integration.md
index ed19ddcc9b08..3b5b2e184047 100644
--- a/docs/streaming-kinesis-integration.md
+++ b/docs/streaming-kinesis-integration.md
@@ -117,8 +117,7 @@ A Kinesis stream can be set up at one of the valid Kinesis endpoints with 1 or m
                 import org.apache.spark.streaming.kinesis.KinesisInputDStream
                 import org.apache.spark.streaming.{Seconds, StreamingContext}
                 import org.apache.spark.streaming.kinesis.KinesisInitialPositions
-                import com.amazonaws.services.kinesis.clientlibrary.lib.worker.KinesisClientLibConfiguration
-                import com.amazonaws.services.kinesis.metrics.interfaces.MetricsLevel
+                import software.amazon.kinesis.metrics.MetricsLevel
 
                 val kinesisStream = KinesisInputDStream.builder
                     .streamingContext(streamingContext)
@@ -130,7 +129,7 @@ A Kinesis stream can be set up at one of the valid Kinesis endpoints with 1 or m
                     .checkpointInterval([checkpoint interval])
                     .storageLevel(StorageLevel.MEMORY_AND_DISK_2)
                     .metricsLevel(MetricsLevel.DETAILED)
-                    .metricsEnabledDimensions(KinesisClientLibConfiguration.DEFAULT_METRICS_ENABLED_DIMENSIONS.asScala.toSet)
+                    .metricsEnabledDimensions(KinesisInputDStream.DEFAULT_METRICS_ENABLED_DIMENSIONS.asScala.toSet)
                     .buildWithMessageHandler([message handler])
 
 	</div>
@@ -140,8 +139,7 @@ A Kinesis stream can be set up at one of the valid Kinesis endpoints with 1 or m
                 import org.apache.spark.streaming.Seconds;
                 import org.apache.spark.streaming.StreamingContext;
                 import org.apache.spark.streaming.kinesis.KinesisInitialPositions;
-                import com.amazonaws.services.kinesis.clientlibrary.lib.worker.KinesisClientLibConfiguration;
-                import com.amazonaws.services.kinesis.metrics.interfaces.MetricsLevel;
+                import software.amazon.kinesis.metrics.MetricsLevel;
                 import scala.collection.JavaConverters;
 
                 KinesisInputDStream<byte[]> kinesisStream = KinesisInputDStream.builder()
@@ -154,7 +152,7 @@ A Kinesis stream can be set up at one of the valid Kinesis endpoints with 1 or m
                     .checkpointInterval([checkpoint interval])
                     .storageLevel(StorageLevel.MEMORY_AND_DISK_2)
                     .metricsLevel(MetricsLevel.DETAILED)
-                    .metricsEnabledDimensions(JavaConverters.asScalaSetConverter(KinesisClientLibConfiguration.DEFAULT_METRICS_ENABLED_DIMENSIONS).asScala().toSet())
+                    .metricsEnabledDimensions(JavaConverters.asScalaSetConverter(KinesisInputDStream.DEFAULT_METRICS_ENABLED_DIMENSIONS).asScala().toSet())
                     .buildWithMessageHandler([message handler]);
 
 	</div>
diff --git a/pom.xml b/pom.xml
index deccc904dd91..76c100347936 100644
--- a/pom.xml
+++ b/pom.xml
@@ -156,8 +156,10 @@
     <!-- Should be consistent with SparkBuild.scala and docs -->
     <avro.version>1.11.1</avro.version>
     <aws.kinesis.client.version>1.12.0</aws.kinesis.client.version>
+    <aws.kinesis.client.v2.version>2.4.3</aws.kinesis.client.v2.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.447</aws.java.sdk.version>
+    <aws.java.sdk.v2.version>2.17.268</aws.java.sdk.v2.version>
     <!-- the producer is used in tests -->
     <aws.kinesis.producer.version>0.12.8</aws.kinesis.producer.version>
     <gcs-connector.version>hadoop3-2.2.14</gcs-connector.version>
diff --git a/project/plugins.sbt b/project/plugins.sbt
index ce4f7afb743b..d5ebdb0265fa 100644
--- a/project/plugins.sbt
+++ b/project/plugins.sbt
@@ -46,3 +46,5 @@ addSbtPlugin("com.simplytyped" % "sbt-antlr4" % "0.8.3")
 addSbtPlugin("com.github.sbt" % "sbt-pom-reader" % "2.4.0")
 
 addSbtPlugin("com.thesamet" % "sbt-protoc" % "1.0.6")
+
+addDependencyTreePlugin
\ No newline at end of file
diff --git a/resource-managers/kubernetes/integration-tests/pom.xml b/resource-managers/kubernetes/integration-tests/pom.xml
index 02894f82eec9..ffebc8b8ddba 100644
--- a/resource-managers/kubernetes/integration-tests/pom.xml
+++ b/resource-managers/kubernetes/integration-tests/pom.xml
@@ -210,9 +210,9 @@
       </activation>
       <dependencies>
         <dependency>
-          <groupId>com.amazonaws</groupId>
-          <artifactId>aws-java-sdk-bundle</artifactId>
-          <version>1.11.375</version>
+          <groupId>software.amazon.awssdk</groupId>
+          <artifactId>s3</artifactId>
+          <version>${aws.java.sdk.v2.version}</version>
           <scope>test</scope>
         </dependency>
       </dependencies>
diff --git a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/DepsTestsSuite.scala b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/DepsTestsSuite.scala
index 8b94a65264a0..27e1b812ac7a 100644
--- a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/DepsTestsSuite.scala
+++ b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/DepsTestsSuite.scala
@@ -17,22 +17,25 @@
 package org.apache.spark.deploy.k8s.integrationtest
 
 import java.io.File
-import java.net.URL
+import java.net.{URI, URL}
 import java.nio.file.Files
 
 import scala.collection.JavaConverters._
 
-import com.amazonaws.auth.BasicAWSCredentials
-import com.amazonaws.services.s3.AmazonS3Client
 import io.fabric8.kubernetes.api.model._
 import io.fabric8.kubernetes.api.model.apps.StatefulSetBuilder
 import org.apache.hadoop.util.VersionInfo
 import org.scalatest.concurrent.{Eventually, PatienceConfiguration}
 import org.scalatest.time.{Minutes, Span}
+import software.amazon.awssdk.auth.credentials.{AwsBasicCredentials, StaticCredentialsProvider}
+import software.amazon.awssdk.core.sync.RequestBody
+import software.amazon.awssdk.regions.Region
+import software.amazon.awssdk.services.s3.S3Client
+import software.amazon.awssdk.services.s3.model.{CreateBucketRequest, PutObjectRequest}
 
 import org.apache.spark.SparkException
 import org.apache.spark.deploy.k8s.integrationtest.DepsTestsSuite.{DEPS_TIMEOUT, FILE_CONTENTS, HOST_PATH}
-import org.apache.spark.deploy.k8s.integrationtest.KubernetesSuite.{INTERVAL, MinikubeTag, SPARK_PI_MAIN_CLASS, TIMEOUT}
+import org.apache.spark.deploy.k8s.integrationtest.KubernetesSuite._
 import org.apache.spark.deploy.k8s.integrationtest.Utils.getExamplesJarName
 import org.apache.spark.deploy.k8s.integrationtest.backend.minikube.Minikube
 import org.apache.spark.internal.config.{ARCHIVES, PYSPARK_DRIVER_PYTHON, PYSPARK_PYTHON}
@@ -45,6 +48,7 @@ private[spark] trait DepsTestsSuite { k8sSuite: KubernetesSuite =>
   val BUCKET = "spark"
   val ACCESS_KEY = "minio"
   val SECRET_KEY = "miniostorage"
+  val REGION = "us-west-2"
 
   private def getMinioContainer(): Container = {
     val envVars = Map (
@@ -302,10 +306,13 @@ private[spark] trait DepsTestsSuite { k8sSuite: KubernetesSuite =>
   private def getS3Client(
       endPoint: String,
       accessKey: String = ACCESS_KEY,
-      secretKey: String = SECRET_KEY): AmazonS3Client = {
-    val credentials = new BasicAWSCredentials(accessKey, secretKey)
-    val s3client = new AmazonS3Client(credentials)
-    s3client.setEndpoint(endPoint)
+      secretKey: String = SECRET_KEY): S3Client = {
+    val credentials = AwsBasicCredentials.create(accessKey, secretKey)
+    val s3client = S3Client.builder()
+      .credentialsProvider(StaticCredentialsProvider.create(credentials))
+      .endpointOverride(URI.create(endPoint))
+      .region(Region.of(REGION))
+      .build()
     s3client
   }
 
@@ -313,9 +320,13 @@ private[spark] trait DepsTestsSuite { k8sSuite: KubernetesSuite =>
     Eventually.eventually(TIMEOUT, INTERVAL) {
       try {
         val s3client = getS3Client(endPoint, accessKey, secretKey)
-        s3client.createBucket(BUCKET)
+        val createBucketRequest = CreateBucketRequest.builder()
+          .bucket(BUCKET)
+          .build()
+        s3client.createBucket(createBucketRequest)
       } catch {
         case e: Exception =>
+          logError(s"Failed to create bucket $BUCKET", e)
           throw new SparkException(s"Failed to create bucket $BUCKET.", e)
       }
     }
@@ -328,7 +339,11 @@ private[spark] trait DepsTestsSuite { k8sSuite: KubernetesSuite =>
     Eventually.eventually(TIMEOUT, INTERVAL) {
       try {
         val s3client = getS3Client(endPoint)
-        s3client.putObject(BUCKET, objectKey, objectContent)
+        val putObjectRequest = PutObjectRequest.builder()
+          .bucket(BUCKET)
+          .key(objectKey)
+          .build()
+        s3client.putObject(putObjectRequest, RequestBody.fromString(objectContent))
       } catch {
         case e: Exception =>
           throw new SparkException(s"Failed to create object $BUCKET/$objectKey.", e)