refactor: bump aws sdk, use rustls in s3 client

Cargo.toml

@@ -68,37 +68,30 @@ repository = "https://github.com/risingwavelabs/risingwave"
 
 [workspace.dependencies]
 await-tree = "0.1.1"
-aws-config = { version = "0.55", default-features = false, features = [
-  "rt-tokio",
-  "native-tls",
+aws-config = { version = "0.57", default-features = false, features = [
+  "rt-tokio", "rustls",
 ] }
-aws-credential-types = { version = "0.55", default-features = false, features = [
+aws-credential-types = { version = "0.57", default-features = false, features = [
   "hardcoded-credentials",
 ] }
-aws-sdk-kinesis = { version = "0.28", default-features = false, features = [
-  "rt-tokio",
-  "native-tls",
+aws-sdk-kinesis = { version = "0.35", default-features = false, features = [
+  "rt-tokio", "rustls",
 ] }
-aws-sdk-s3 = { version = "0.28", default-features = false, features = [
-  "rt-tokio",
-  "native-tls",
+aws-sdk-s3 = { version = "0.35", default-features = false, features = [
+  "rt-tokio", "rustls",
 ] }
-aws-sdk-ec2 = { version = "0.28", default-features = false, features = [
-  "rt-tokio",
-  "native-tls",
+aws-sdk-ec2 = { version = "0.35", default-features = false, features = [
+  "rt-tokio", "rustls",
 ] }
-aws-sdk-sqs = { version = "0.28", default-features = false, features = [
-  "rt-tokio",
-  "native-tls",
+aws-sdk-sqs = { version = "0.35", default-features = false, features = [
+  "rt-tokio", "rustls",
 ] }
-aws-smithy-client = { version = "0.55", default-features = false, features = [
-  "rt-tokio",
-  "native-tls",
-] }
-aws-smithy-http = "0.55"
-aws-smithy-types = "0.55"
-aws-endpoint = "0.55"
-aws-types = "0.55"
+aws-smithy-http = "0.57"
+aws-smithy-runtime-api = "0.57"
+aws-smithy-runtime = "0.57"
+aws-smithy-types = "0.57"
+aws-endpoint = "0.57"
+aws-types = "0.57"
 etcd-client = { package = "madsim-etcd-client", version = "0.4" }
 futures-async-stream = "0.2.9"
 hytra = "0.1"

src/bench/Cargo.toml

@@ -12,6 +12,7 @@ async-trait = "0.1"
 aws-config = { workspace = true }
 aws-sdk-s3 = { workspace = true }
 aws-smithy-http = { workspace = true }
+aws-smithy-types = { workspace = true }
 bcc = { version = "0.0.33", optional = true }
 bytes = "1"
 bytesize = { version = "1", features = ["serde"] }

src/bench/s3_bench/main.rs

@@ -21,7 +21,7 @@ use std::time::{Duration, Instant};
 use aws_config::retry::RetryConfig;
 use aws_sdk_s3::types::{CompletedMultipartUpload, CompletedPart};
 use aws_sdk_s3::Client;
-use aws_smithy_http::body::SdkBody;
+use aws_smithy_types::body::SdkBody;
 use bytesize::ByteSize;
 use clap::Parser;
 use futures::stream::{self, StreamExt};

src/connector/Cargo.toml

@@ -32,6 +32,8 @@ aws-credential-types = { workspace = true }
 aws-sdk-kinesis = { workspace = true }
 aws-sdk-s3 = { workspace = true }
 aws-smithy-http = { workspace = true }
+aws-smithy-runtime-api = { workspace = true }
+aws-smithy-types = { workspace = true }
 aws-types = { workspace = true }
 base64 = "0.21"
 byteorder = "1"

src/connector/src/aws_auth.rs

@@ -114,7 +114,8 @@ impl AwsAuthProps {
             if let Some(id) = &self.external_id {
                 role = role.external_id(id);
             }
-            Ok(SharedCredentialsProvider::new(role.build(credential)))
+            let provider = role.build_from_provider(credential).await;
+            Ok(SharedCredentialsProvider::new(provider))
         } else {
             Ok(credential)
         }

src/connector/src/source/filesystem/s3/source/reader.rs

@@ -19,10 +19,12 @@ use anyhow::{anyhow, Result};
 use async_trait::async_trait;
 use aws_sdk_s3::client as s3_client;
 use aws_sdk_s3::operation::get_object::GetObjectError;
-use aws_smithy_http::byte_stream::ByteStream;
-use aws_smithy_http::result::SdkError;
-use futures::TryStreamExt;
+use aws_smithy_http::futures_stream_adapter::FuturesStreamCompatByteStream;
+use aws_smithy_runtime_api::client::result::SdkError;
+use aws_smithy_types::body::SdkBody;
+use aws_smithy_types::byte_stream::ByteStream;
 use futures_async_stream::try_stream;
+use hyper::Response;
 use io::StreamReader;
 use risingwave_common::error::RwError;
 use tokio::io::BufReader;
@@ -91,9 +93,10 @@ impl S3FileReader {
             }
         };
 
-        let stream_reader = StreamReader::new(
-            byte_stream.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e)),
-        );
+        // FYI: https://github.com/awslabs/smithy-rs/pull/2983
+        let byte_stream = FuturesStreamCompatByteStream::new(byte_stream);
+
+        let stream_reader = StreamReader::new(byte_stream);
 
         let reader = pin!(BufReader::new(stream_reader));
 
@@ -142,7 +145,7 @@ impl S3FileReader {
         bucket_name: &str,
         object_name: &str,
         start: usize,
-    ) -> std::result::Result<ByteStream, SdkError<GetObjectError>> {
+    ) -> std::result::Result<ByteStream, SdkError<GetObjectError, Response<SdkBody>>> {
         let range = if start == 0 {
             None
         } else {

src/connector/src/source/kinesis/enumerator/client.rs

@@ -75,7 +75,7 @@ impl SplitEnumerator for KinesisSplitEnumerator {
         Ok(shard_collect
             .into_iter()
             .map(|x| KinesisSplit {
-                shard_id: x.shard_id().unwrap_or_default().to_string().into(),
+                shard_id: x.shard_id().to_string().into(),
                 // handle start with position in reader part
                 start_position: KinesisOffset::None,
                 end_position: KinesisOffset::None,

src/connector/src/source/kinesis/source/message.rs

@@ -40,9 +40,9 @@ impl KinesisMessage {
     pub fn new(shard_id: SplitId, message: Record) -> Self {
         KinesisMessage {
             shard_id,
-            sequence_number: message.sequence_number.unwrap(),
-            partition_key: message.partition_key.unwrap(),
-            payload: message.data.unwrap().into_inner(),
+            sequence_number: message.sequence_number,
+            partition_key: message.partition_key,
+            payload: message.data.into_inner(),
         }
     }
 }

src/connector/src/source/kinesis/source/reader.rs

@@ -137,7 +137,7 @@ impl CommonSplitReader for KinesisSplitReader {
             match self.get_records().await {
                 Ok(resp) => {
                     self.shard_iter = resp.next_shard_iterator().map(String::from);
-                    let chunk = (resp.records().unwrap().iter())
+                    let chunk = (resp.records().iter())
                         .map(|r| {
                             SourceMessage::from(KinesisMessage::new(
                                 self.shard_id.clone(),

src/meta/src/rpc/cloud_provider.rs

@@ -62,14 +62,12 @@ impl AwsEc2Client {
                 )
             })?;
 
-        if let Some(ret) = output.unsuccessful() {
-            if !ret.is_empty() {
-                return Err(MetaError::from(anyhow!(
-                    "Failed to delete VPC endpoint {}, error: {:?}",
-                    vpc_endpoint_id,
-                    ret
-                )));
-            }
+        if !output.unsuccessful().is_empty() {
+            return Err(MetaError::from(anyhow!(
+                "Failed to delete VPC endpoint {}, error: {:?}",
+                vpc_endpoint_id,
+                output.unsuccessful()
+            )));
         }
         Ok(())
     }
@@ -325,13 +323,11 @@ impl AwsEc2Client {
         let endpoint = output.vpc_endpoint().unwrap();
         let mut dns_names = Vec::new();
 
-        if let Some(dns_entries) = endpoint.dns_entries() {
-            dns_entries.iter().for_each(|e| {
-                if let Some(dns_name) = e.dns_name() {
-                    dns_names.push(dns_name.to_string());
-                }
-            });
-        }
+        endpoint.dns_entries().iter().for_each(|e| {
+            if let Some(dns_name) = e.dns_name() {
+                dns_names.push(dns_name.to_string());
+            }
+        });
 
         Ok((
             endpoint.vpc_endpoint_id().unwrap_or_default().to_string(),

src/object_store/Cargo.toml

@@ -12,21 +12,24 @@ repository = { workspace = true }
 async-trait = "0.1"
 await-tree = { workspace = true }
 aws-config = { workspace = true }
-aws-sdk-s3 = { version = "0.2", package = "madsim-aws-sdk-s3" }
-aws-smithy-client = { workspace = true }
+aws-sdk-s3 = { version = "0.3", package = "madsim-aws-sdk-s3" }
 aws-smithy-http = { workspace = true }
+aws-smithy-runtime = { workspace = true }
+aws-smithy-runtime-api = { workspace = true }
 aws-smithy-types = { workspace = true }
 bytes = { version = "1", features = ["serde"] }
 crc32fast = "1.3.2"
 either = "1"
 fail = "0.5"
 futures = { version = "0.3", default-features = false, features = ["alloc"] }
 hyper = "0.14"
+hyper-rustls = { version = "0.24.2", features = ["webpki-roots"] }
 hyper-tls = "0.5.0"
 itertools = "0.11"
 opendal = "0.41"
 prometheus = { version = "0.13", features = ["process"] }
 risingwave_common = { workspace = true }
+rustls = "0.21.8"
 spin = "0.9"
 thiserror = "1"
 tokio = { version = "0.2", package = "madsim-tokio", features = ["fs"] }

src/object_store/src/object/error.rs

@@ -20,6 +20,8 @@ use aws_sdk_s3::error::DisplayErrorContext;
 use aws_sdk_s3::operation::get_object::GetObjectError;
 use aws_sdk_s3::operation::head_object::HeadObjectError;
 use aws_sdk_s3::primitives::ByteStreamError;
+use aws_smithy_types::body::SdkBody;
+use hyper::Response;
 use risingwave_common::error::BoxedError;
 use thiserror::Error;
 use tokio::sync::oneshot::error::RecvError;
@@ -79,13 +81,19 @@ impl ObjectError {
     pub fn is_object_not_found_error(&self) -> bool {
         match &self.inner {
             ObjectErrorInner::S3(e) => {
-                if let Some(aws_smithy_http::result::SdkError::ServiceError(err)) =
-                    e.downcast_ref::<aws_smithy_http::result::SdkError<GetObjectError>>()
+                if let Some(aws_smithy_runtime_api::client::result::SdkError::ServiceError(err)) = e
+                    .downcast_ref::<aws_smithy_runtime_api::client::result::SdkError<
+                        GetObjectError,
+                        Response<SdkBody>,
+                    >>()
                 {
                     return matches!(err.err(), GetObjectError::NoSuchKey(_));
                 }
-                if let Some(aws_smithy_http::result::SdkError::ServiceError(err)) =
-                    e.downcast_ref::<aws_smithy_http::result::SdkError<HeadObjectError>>()
+                if let Some(aws_smithy_runtime_api::client::result::SdkError::ServiceError(err)) = e
+                    .downcast_ref::<aws_smithy_runtime_api::client::result::SdkError<
+                        HeadObjectError,
+                        Response<SdkBody>,
+                    >>()
                 {
                     return matches!(err.err(), HeadObjectError::NotFound(_));
                 }