aws_kinesis_stream sink batching is not working

[romkimchi1002]

A note for the community

• Please vote on this issue by adding a 👍 reaction to the original issue to help the community and maintainers prioritize this request
• If you are interested in working on this issue or have submitted a pull request, please leave a comment

Problem

records are not batched together before being sent to kinesis. each record is sent individually to kinesis

Configuration

acknowledgements:
  enabled: true

sinks:
  kinesis:
    type: 'aws_kinesis_streams'
    inputs:
      - 'events_source'
    encoding:
      codec: 'json'
    auth:
      access_key_id: '#######'
      secret_access_key: '#######'
    stream_name: '#######'
    region: '######'
    # using default batching configuration

sources:
  events_source:
    type: 'http_server'
    address: '127.0.0.1:5000'
    auth: 
      username: 'user'
      password: 'pass'
    encoding: 'json'
    method: 'POST'
    path: '/data'
    response_code: 202

Version

vector 0.38.0 (aarch64-apple-darwin)

Debug Output

No response

Example Data

i tried debugging it myself and downloded vecror as a source.
i added to this file
vector/src/sinks/aws_kinesis/streams/record.rs
this print line:

async fn send(
        &self,
        records: Vec<Self::T>,
        stream_name: String,
    ) -> Result<KinesisResponse, SdkError<Self::E, HttpResponse>> {
        let rec_count = records.len();
        let total_size = records
            .iter()
            .fold(0, |acc, record| acc + record.data().as_ref().len());

        println!("Sending {} records to stream {}. size: {}", rec_count, stream_name, total_size); // Debugging line

this line always return 1 no matter what configuration i'm using.

after modifying the code it seems that the partition_key is used for the batching, which is a bug since partition key should be a field used by the kinesis for spreading data across shards.

Additional Context

No response

References

No response

[jszwedko]

I think there may be some, understandable, confusion here. The batch settings for the aws_kinesis sink correspond to how the events are batched into AWS API PutRecords calls. That is, if you the batch size is set to 500, it'll send 500 records in a single request. However, each event is always a single Kinesis record. It seems like you might be expecting it to batch within a Kinesis record.

Can you share more about your use-case for putting multiple events into a single Kinesis record?

[jszwedko]

Related:

• Batch multiple events as a single record in aws_kinesis_firehose sink #1407

[romkimchi1002]

I think there may be some, understandable, confusion here. The batch settings for the aws_kinesis sink correspond to how the events are batched into AWS API PutRecords calls. That is, if you the batch size is set to 500, it'll send 500 records in a single request. However, each event is always a single Kinesis record. It seems like you might be expecting it to batch within a Kinesis record.

Can you share more about your use-case for putting multiple events into a single Kinesis record?

hi! im not trying to batch them into a single kinesis event. The line i added in the vector source indicates that the putRecords only put 1 vector event instead of 500.

async fn send(
        &self,
        records: Vec<Self::T>,
        stream_name: String,
    ) -> Result<KinesisResponse, SdkError<Self::E, HttpResponse>> {
        let rec_count = records.len();
        let total_size = records
            .iter()
            .fold(0, |acc, record| acc + record.data().as_ref().len());

        println!("Sending {} records to stream {}. size: {}", rec_count, stream_name, total_size); // Debugging line

i think that the issue might be in this function: batched_partitioned in the file: vector/src/sinks/aws_kinesis/sink.rs. but i didn't find the definition of it anywhere. reminder that when i modified vector code to use a constant partition key, the batching works, i assume that batched_partitioned function uses the partition key for batching when it shouldn't.

when using:

 fn gen_partition_key(number_of_shards) -> String {
    "fixed_partition_key".to_string()
}

the debugging print gives:

[jszwedko]

Oh I see, thanks for the additional detail @romkimchi1002 ! It does look like there is a bug here then; likely with the partition key as you noted.

[romkimchi1002]

thanks!

[steven-aerts]

I hereby confirm that this problem also exists for the aws_kinesis_firehose sink.
We were able to work around it, by selecting a (locally) constant field as partition_key_field:

[sinks.firehose_al]
  type = "aws_kinesis_firehose"
  inputs = ["input"]
  stream_name = "firehose-stream-name"
  encoding.codec = "json"
  partition_key_field = "hostname" # Workaround for https://github.com/vectordotdev/vector/issues/20575