[SPARK-45720] Upgrade AWS SDK to v2 for Spark Kinesis connector module #44211

junyuc25 · 2023-12-06T13:22:20Z

What changes were proposed in this pull request?

As Spark is moving to 4.0, one of the major improvement is to upgrade AWS SDK to v2.

Currently other than directly using AWS SDKv1 codes, the Spark Kinesis connector is also
using on these libraries that depends on SDKv1:

Kinesis Client Library (KCL) allows users to easily consume and process data from Amazon Kinesis
Kinesis Producer Library (KPL) allows users to create reliable and efficient message producers for Amazon Kinesis

The main purpose of this PR is to upgrading AWS SDK to v2 for the Spark Kinesis
conector. While the changes includes upgrading AWS SDK and KCL to v2, we will not
upgrade KPL because it has not yet been migrated to SDKv2.

Parent Jira: parent Jira: https://issues.apache.org/jira/browse/SPARK-44124.
Previous PR to setup Kinesis tests in Github Actions: [SPARK-32246][BUILD][INFRA] Enable streaming-kinesis-asl tests in Github Action CI #43736
Previous stale PR: [WIP] AWS sdk upgrade #42581

Why are the changes needed?

As the GA of AWS SDK v2, the SDKv1 has entered maintenance mode where its future
release are only limited to address critical bug and security issues. More details about the SDK maintenance policy can be found here. To keep Spark’s dependent softwares up to date, we should consider upgrading the SDK to v2.
These changes could keep Spark Kinesis connector up to date, and enable users to
receive continuous support from the above libraries.

Does this PR introduce any user-facing change?

Yes. With this change, the Spark Kinesis connector will no longer work with SDKv1.
Any applications that are running with previous version of Spark Kinesis connector
would require update before migrating to Spark 4.0.

AWS SDKv2 and KCLv2 contain several major changes
that are not backward compatible with their previous versions. And some public classes
in the module (i.e. KinesisInputDStream) are using one of these breaking changes. Thus
these user-facing classes require updates as well.

How was this patch tested?

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

No

dongjoon-hyun · 2023-12-06T20:56:39Z

connector/kinesis-asl-assembly/pom.xml

+      <groupId>com.google.guava</groupId>
+      <artifactId>guava</artifactId>
+      <version>${aws.kinesis.client.guava.version}</version>
+      <scope>compile</scope>


Ur, I'm not sure this is okay or not, @junyuc25 .

Could you elaborate a bit more on why this is not OK? I think this pattern is also seen in other modules like https://github.com/apache/spark/blob/master/connector/connect/server/pom.xml#L159-L164

cc @dongjoon-hyun

Could you share more about your concern here @dongjoon-hyun ? Thanks!

Sorry for being late.

Is this only for Kinesis or for all AWS SKD v2?

Instead of the following, can we use the latest33.0.0-jre like [SPARK-46768][BUILD] Upgrade Guava used by the connect module to 33.0-jre #44795 ?

<aws.kinesis.client.guava.version>32.1.1-jre</aws.kinesis.client.guava.version>

This change is only for the Kinesis connector modules.

Yes I changed it to use the same Guava version as the one used in other module.

dongjoon-hyun · 2023-12-06T20:57:53Z

Anyway, thank you so much for working on this area, @junyuc25 .

pan3793 · 2023-12-08T12:54:03Z

connector/kinesis-asl/pom.xml

@@ -54,14 +54,38 @@
      <scope>test</scope>
    </dependency>
    <dependency>
-      <groupId>com.amazonaws</groupId>
+      <groupId>software.amazon.kinesis</groupId>
      <artifactId>amazon-kinesis-client</artifactId>


Could you please clarify the relationship between
software.amazon.kinesis:amazon-kinesis-client and software.amazon.awssdk:kinesis?

Seems they don't share the version, and the latter is hygienic (no third-party dependencies other than software.amazon.awssdk:*)

Handling Jackson/Guava/Protobuf dependencies conflict is always painful, is it possible to provide a similar hygienic artifact for the former to make the downstream project easier to consume?

Hi @pan3793, KCL (software.amazon.kinesis:amazon-kinesis-client) is built on top of Kinesis Data Stream API (software.amazon.awssdk:kinesis), and KCL provides additional functionalities such as load balancing, error recovery etc. According to this doc, generally it is recommended to use KCL over Kinesis Data Stream API:https://docs.aws.amazon.com/streams/latest/dev/developing-consumers-with-sdk.html .

Perhaps we could try to raise an issue to the KCL repo and ask if they can release a "clean" version of the library with all the third-party dependencies shaded. But currently I'm not aware there is such a hygienic version of KCL. I guess we probably have to live with this issue if we use KCL.

Perhaps we could try to raise an issue to the KCL repo and ask if they can release a "clean" version of the library with all the third-party dependencies shaded.

Would you like to drive this? It will benefit all downstream projects.

Hi @pan3793, I submitted an ticket to the KCL repo: awslabs/amazon-kinesis-client#1245. Let's see what response we would get. On the other hand, it seems to me that this would be a follow-up task, rather than a blocker for this PR? Please correct me if I am wrong.

LuciferYang · 2023-12-21T07:06:52Z

connector/kinesis-asl-assembly/pom.xml

-         spark default of 2.5.0 because KCL appears to have introduced 
-         a dependency on protobuf 2.6.1 somewhere between KCL 1.4.0 and 1.6.1.
-       -->
+      <scope>compile</scope>


I'm glad to see that Kinesis can use the same protobuf-java as other modules, but is this feasible?

Hi @LuciferYang, I saw similar patterns in other modules as well: https://github.com/apache/spark/blob/v3.5.0/connector/connect/server/pom.xml#L173-L175. I believe this should work as long as we shade it.

For the record, I removed this shaded protobuf-java completely in Apache Spark 4.0.0 independently.

[SPARK-47038][BUILD] Remove shaded protobuf-java 2.6.1 dependency from kinesis-asl-assembly #45096

Thanks @dongjoon-hyun for letting me know. Updated the PR to use the same protobuf-java version as specified in <protobuf.version>. I had to change the scope to "compile" to fix ClassNotFoundExceptions.

LantaoJin · 2024-01-19T07:34:25Z

...or/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;
+


trivial: omit the empty line #L19

LantaoJin · 2024-01-19T07:42:05Z

...tor/kinesis-asl/src/main/scala/org/apache/spark/examples/streaming/KinesisExampleUtils.scala

      .asScala
-      .find(_.getAvailableEndpoints.asScala.toSeq.contains(uri.getHost))
-      .map(_.getName)
+      .find(r => kinesisServiceMetadata.endpointFor(r).toString.equals(uri.getHost))


How about r => kinesisServiceMetadata.endpointFor(r).equals(uri)

It looks like the URI objects would be different I we make this change. Comparing these two different URI causes test failure.

The URI created from kinesisServiceMetadata.endpointFor(Regioin.US_WEST_2): kinesis.us-west-2.amazon.com

The URI created from given endpoint string contains HTTP protocol: https://kinesis.us-west-2.amazon.com/

Stacktrace:

Cause: java.lang.IllegalArgumentException: Could not resolve region for endpoint: https://kinesis.us-west-2.amazonaws.com at org.apache.spark.streaming.kinesis.KinesisTestUtils$.$anonfun$getRegionNameByEndpoint$3(KinesisTestUtils.scala:237) at scala.Option.getOrElse(Option.scala:201) at org.apache.spark.streaming.kinesis.KinesisTestUtils$.getRegionNameByEndpoint(KinesisTestUtils.scala:237) at org.apache.spark.streaming.kinesis.KinesisStreamTests.<init>(KinesisStreamSuite.scala:51) at org.apache.spark.streaming.kinesis.WithoutAggregationKinesisStreamSuite.<init>(KinesisStreamSuite.scala:434)

So to avoid this failure, I would prefer to keep this line. kinesisServiceMetadata.endpointFor(r).toString.equals(uri.getHost)

LantaoJin · 2024-01-19T07:57:35Z

...ctor/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisCheckpointer.scala

@@ -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


Could you correct this comment section? At least no more IRecordProcessor.

Thanks for catching this. Updated.

LantaoJin · 2024-01-19T08:03:53Z

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


  // shardId populated during initialize()
  @volatile
  private var shardId: String = _

-  /**
-   * The Kinesis Client Library calls this method during IRecordProcessor initialization.


Why the comments before methods (such as initialize, processRecords ) are removed in this class? And for the new adding public methods, please add method comments as well.

Added comments.

LantaoJin · 2024-01-19T08:04:51Z

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

-    val splitShardRequest = new SplitShardRequest()
-    splitShardRequest.withStreamName(_streamName)
-    splitShardRequest.withShardToSplit(shardId)
-    // Set a half of the max hash value


This comment should not be omitted.

Added it back

LantaoJin · 2024-01-19T08:07:47Z

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

      .asScala
-      .find(_.getAvailableEndpoints.asScala.toSeq.contains(uri.getHost))
-      .map(_.getName)
+      .find(r => kinesisServiceMetadata.endpointFor(r).toString.equals(uri.getHost))


Same as above response.

LantaoJin · 2024-01-19T08:08:59Z

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

  }

-  def getAWSCredentials(): AWSCredentials = {
+  def getAwsCredentials: AwsCredentials = {


This modification will change the API. Please revert it and keep () in method.

LantaoJin · 2024-01-19T08:13:10Z

...ctor/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/SparkAWSCredentials.scala

 }

 /**
- * 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


AWSCredentialsProviderChain was retired, check all terms in all comments in this file.

LantaoJin · 2024-01-22T11:31:02Z

@junyuc25 why you close this PR? And you should remove the [WIP] in the title when your PR is ready for review, or committers cannot know when could start to review.

junyuc25 · 2024-01-23T10:11:40Z

@junyuc25 why you close this PR? And you should remove the [WIP] in the title when your PR is ready for review, or committers cannot know when could start to review.

Looks like I deleted the branch by accident. Updated the title and reopened the PR.

Upgrade codes related to IAM, DDB and Kinesis clients Shade SDKv2 dependencies Use Spark's default protobuf version in the kinesis-asl-assembly module Import mbknor-jackson-jsonschema_2.13 for glue-schema-registry Jira: https://issues.apache.org/jira/browse/SPARK-45720

Fix jaxb version

github-actions bot added BUILD DOCS labels Dec 6, 2023

junyuc25 force-pushed the upgrade-kinesis-connector branch from b6e4c48 to 9970a6a Compare December 6, 2023 13:43

junyuc25 changed the title ~~[SPARK-45720] Upgrade AWS SDK to v2 for Spark Kinesis connector module~~ [WIP][SPARK-45720] Upgrade AWS SDK to v2 for Spark Kinesis connector module Dec 6, 2023

dongjoon-hyun reviewed Dec 6, 2023

View reviewed changes

pan3793 reviewed Dec 8, 2023

View reviewed changes

LuciferYang reviewed Dec 21, 2023

View reviewed changes

junyuc25 marked this pull request as ready for review January 10, 2024 08:19

LantaoJin reviewed Jan 19, 2024

View reviewed changes

junyuc25 closed this Jan 22, 2024

junyuc25 force-pushed the upgrade-kinesis-connector branch from c74e184 to b07bdea Compare January 22, 2024 08:26

junyuc25 changed the title ~~[WIP][SPARK-45720] Upgrade AWS SDK to v2 for Spark Kinesis connector module~~ [SPARK-45720] Upgrade AWS SDK to v2 for Spark Kinesis connector module Jan 23, 2024

junyuc25 reopened this Jan 23, 2024

junyuc25 force-pushed the upgrade-kinesis-connector branch 2 times, most recently from d0f3157 to 615c370 Compare February 26, 2024 02:41

junyuc25 force-pushed the upgrade-kinesis-connector branch from 615c370 to eb93c37 Compare May 9, 2024 12:36

github-actions bot added the DSTREAM label May 9, 2024

Junyu Chen added 9 commits May 10, 2024 17:50

Exclude zstd-jni brought in by glue-schema-registry

8940369

Update method comments

bc98524

Fix test failure

f5306ed

Change guava version to match the one used in connect module

acfbc87

Fix ClassNotFoundException when running kinesis tests with JDK17

ec5f749

Fix jaxb version

Use protobuf-java 3.25.1 and shade it in the kinesis-asl-assembly module

2ae2a28

Fix build errors after rebase

0bb8140

Update log format

28c25ed

junyuc25 force-pushed the upgrade-kinesis-connector branch from 2767d5d to 28c25ed Compare May 10, 2024 13:35

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-45720] Upgrade AWS SDK to v2 for Spark Kinesis connector module #44211

[SPARK-45720] Upgrade AWS SDK to v2 for Spark Kinesis connector module #44211

junyuc25 commented Dec 6, 2023

dongjoon-hyun Dec 6, 2023

junyuc25 Dec 7, 2023

junyuc25 Dec 13, 2023

junyuc25 Jan 5, 2024

dongjoon-hyun Feb 16, 2024

junyuc25 Feb 26, 2024

dongjoon-hyun commented Dec 6, 2023

pan3793 Dec 8, 2023 •

edited

junyuc25 Dec 11, 2023

pan3793 Jan 5, 2024

junyuc25 Jan 9, 2024

LuciferYang Dec 21, 2023 •

edited

junyuc25 Dec 29, 2023

dongjoon-hyun Feb 16, 2024

junyuc25 Feb 26, 2024

LantaoJin Jan 19, 2024

LantaoJin Jan 19, 2024

junyuc25 Jan 23, 2024

LantaoJin Jan 19, 2024

junyuc25 Jan 23, 2024

LantaoJin Jan 19, 2024

junyuc25 Jan 23, 2024

LantaoJin Jan 19, 2024

junyuc25 Jan 23, 2024

LantaoJin Jan 19, 2024

junyuc25 Jan 23, 2024

LantaoJin Jan 19, 2024

junyuc25 Jan 23, 2024

LantaoJin Jan 19, 2024

junyuc25 Jan 23, 2024

LantaoJin commented Jan 22, 2024 •

edited

junyuc25 commented Jan 23, 2024

		@@ -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

[SPARK-45720] Upgrade AWS SDK to v2 for Spark Kinesis connector module #44211

Are you sure you want to change the base?

[SPARK-45720] Upgrade AWS SDK to v2 for Spark Kinesis connector module #44211

Conversation

junyuc25 commented Dec 6, 2023

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dongjoon-hyun commented Dec 6, 2023

pan3793 Dec 8, 2023 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

LuciferYang Dec 21, 2023 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

LantaoJin commented Jan 22, 2024 • edited

junyuc25 commented Jan 23, 2024

pan3793 Dec 8, 2023 •

edited

LuciferYang Dec 21, 2023 •

edited

LantaoJin commented Jan 22, 2024 •

edited