clear_and_get() throws intermittent error in a multi-threaded environment

[jyoti-dhiman]

We are using kinesis_aggregator in a multi-threaded production environment to push records to kinesis and intermittently see this error:
Message AggregatedRecord is missing required fields: records[3].partition_key_index

• First I thought the error was related to the partition key and I used UUID as the partition key(and was assured it is null), but still, error was occurring
• I was able to replicate the error and after some debugging, I reached the following conclusion
  It is not a data-related issue, I tested with JMeter with the same record and intermittently this error occurs on the record.
  I think the error is occurring because maybe the kinesis aggregator is not thread-safe, we have 4 processes(with 30 threads each) on each machine, I suspect that if threads call, send_record(kinesis_agg.clear_and_get()) at the same time, and looking at the implementation of clear_and_get, which resets the aggregated record(hence the partition_key_index will be blank), another thread which has also called clear_and_get fails.

How can I fix this error?
How can I use kinesis aggregator in a multi-threaded environment?

[IanMeyers]

Hello,

Yes, the classes are annotated as @NotThreadSafe for this reason. I'm very keen to accept a pull request with multi-threading support, but to-date haven't been able to spend the time to make it so.

Thx,

Ian