fix: Kinesis: NextToken and StreamName cannot be provided together

[tabVersion]

…ption: NextToken and StreamName cannot be provided together."

I hereby agree to the terms of the RisingWave Labs, Inc. Contributor License Agreement.

What's changed and what's your intention?

• why did this happen?
  • We did not meet a Kinesis has over 1k shards before. And we wrongly use the ListShards pagination API.

as title, the constraint is described in API doc

https://docs.rs/aws-sdk-kinesis/latest/aws_sdk_kinesis/operation/list_shards/struct.ListShardsInput.html

When the number of shards in the data stream is greater than the default value for the MaxResults parameter, or if you explicitly specify a value for MaxResults that is less than the number of shards in the data stream, the response includes a pagination token named NextToken. You can specify this NextToken value in a subsequent call to ListShards to list the next set of shards.

Don't specify StreamName or StreamCreationTimestamp if you specify NextToken because the latter unambiguously identifies the stream.

You can optionally specify a value for the MaxResults parameter when you specify NextToken. If you specify a MaxResults value that is less than the number of shards that the operation returns if you don't specify MaxResults, the response will contain a new NextToken value. You can use the new NextToken value in a subsequent call to the ListShards operation.

Tokens expire after 300 seconds. When you obtain a value for NextToken in the response to a call to ListShards, you have 300 seconds to use that value. If you specify an expired token in a call to ListShards, you get ExpiredNextTokenException.

Checklist

• I have written necessary rustdoc comments
• I have added necessary unit tests and integration tests
• I have added test labels as necessary. See details.
• I have added fuzzing tests or opened an issue to track them. (Optional, recommended for new SQL features Sqlsmith: Sql feature generation #7934).
• My PR contains breaking changes. (If it deprecates some features, please create a tracking issue to remove them in the future).
• All checks passed in ./risedev check (or alias, ./risedev c)
• My PR changes performance-critical code. (Please run macro/micro-benchmarks and show the results.)

• My PR contains critical fixes that are necessary to be merged into the latest release. (Please check out the details)

Documentation

• My PR needs documentation updates. (Please use the Release note section below to summarize the impact on users)

Release note

If this PR includes changes that directly affect users or other significant modifications relevant to the community, kindly draft a release note to provide a concise summary of these changes. Please prioritize highlighting the impact these changes will have on users.

[BugenZhao]

Please refer to the guide to replace it with .context.

https://www.notion.so/risingwave-labs/A-Guide-to-Error-Handling-that-Just-Works-Part-I-e108f8d0845b4bddaa0e5843791a8cac?pvs=4#a5c15caaa931431aaaa3bbaeec801a67

[tabVersion]

Kinesis SdkError has one generic param, seems cannot fit into ConnectorError Impl

pub type SdkError<E, R = ::aws_smithy_runtime_api::client::orchestrator::HttpResponse> = ::aws_smithy_runtime_api::client::result::SdkError<E, R>;

def_anyhow_newtype! {
    pub ConnectorError,

    // Common errors
    std::io::Error => transparent,

    // Fine-grained connector errors
    AccessError => transparent,
    WireFormatError => transparent,
    ConcurrentRequestError => transparent,
    InvalidOptionError => transparent,
    SinkError => transparent,
    PbFieldNotFound => transparent,

    // TODO(error-handling): Remove implicit contexts below and specify ad-hoc context for each conversion.

    // Parsing errors
    url::ParseError => "failed to parse url",
    serde_json::Error => "failed to parse json",
    csv::Error => "failed to parse csv",

    uuid::Error => transparent, // believed to be self-explanatory

    // Connector errors
    opendal::Error => transparent, // believed to be self-explanatory
    parquet::errors::ParquetError => transparent,
    ArrayError => "Array error",
    sqlx::Error => transparent, // believed to be self-explanatory
    mysql_async::Error => "MySQL error",
    tokio_postgres::Error => "Postgres error",
    apache_avro::Error => "Avro error",
    rdkafka::error::KafkaError => "Kafka error",
    pulsar::Error => "Pulsar error",
    aws_sdk_kinesis::error => "Kinesis error",

    async_nats::jetstream::consumer::StreamError => "Nats error",
    async_nats::jetstream::consumer::pull::MessagesError => "Nats error",
    async_nats::jetstream::context::CreateStreamError => "Nats error",
    async_nats::jetstream::stream::ConsumerError => "Nats error",
    icelake::Error => "Iceberg error",
    iceberg::Error => "IcebergV2 error",
    redis::RedisError => "Redis error",
    arrow_schema::ArrowError => "Arrow error",
    arrow_schema_iceberg::ArrowError => "Arrow error",
    google_cloud_pubsub::client::google_cloud_auth::error::Error => "Google Cloud error",
    rumqttc::tokio_rustls::rustls::Error => "TLS error",
    rumqttc::v5::ClientError => "MQTT error",
    rumqttc::v5::OptionError => "MQTT error",
    mongodb::error::Error => "Mongodb error",

    openssl::error::ErrorStack => "OpenSSL error",
}

[BugenZhao]

Can you just simply try return Err(anyhow!(e).context("failed to list kinesis shards").into())

[tabVersion]

Oh I think it just works.

[xxchan]

Can you add the background in the description? Specifically, when does this cause problem?

BTW, this might demonstrate the need for #17116

[tabVersion]

Can you add the background in the description? Specifically, when does this cause problem?

BTW, this might demonstrate the need for #17116

Updated. And maybe, we have so much work to do, I am saying good to have but hard to tell the priority.