[
https://issues.apache.org/jira/browse/CAMEL-26660?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Federico Mariani reassigned CAMEL-26660:
----------------------------------------
Assignee: Federico Mariani
> camel-aws2-kinesis - batch producer stops at a partly failed chunk, re-sends
> from record 0 on retry, no 10 MiB split
> --------------------------------------------------------------------------------------------------------------------
>
> Key: CAMEL-26660
> URL: https://issues.apache.org/jira/browse/CAMEL-26660
> Project: Camel
> Issue Type: Bug
> Components: camel-aws2-kinesis
> Affects Versions: 4.18.5, 4.22.2, 4.23.0
> Reporter: Federico Mariani
> Assignee: Federico Mariani
> Priority: Major
>
> h3. Problem
> When the body is an {{Iterable}}, {{Kinesis2Producer}} sends it as PutRecords
> requests of at most 500 records
> ({{components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Producer.java}}
> lines 40 and 144 on main, commit 7978db3d42dd). The component page says
> "Messages with batch size more than 500 is allowed. Producer will split them
> into multiple requests."
> * *A partly failed chunk stops the send.* PutRecords is not atomic: the
> response can have {{FailedRecordCount > 0}} while the other records of the
> request are written. The producer then throws {{RuntimeException("Failed to
> send records N of M")}} (lines 100-103).
> ** The chunks after that one are never sent.
> ** The records of the earlier chunks, and the good records of that chunk,
> stay written.
> ** The exception does not say which records were written.
> {{CamelAwsKinesisRecordCount}} and {{CamelAwsKinesisFailedRecordCount}} are
> only set after the loop (lines 106-108), so
> {{CamelAwsKinesisFailedRecordCount}} can only ever be 0.
> * *A retry re-sends from record 0.* The producer never re-sends by itself,
> but it does not remember what was written. Any retry of the exchange sends
> the whole Iterable again, so every record already written is written twice. A
> retry is either a Camel redelivery (off by default) or a redelivery by the
> source of the route (for example a transacted JMS or an SQS consumer).
> * *No split by size.* Chunks are cut at 500 records only. The SDK 2.57.7
> service model limits PutRecords to 500 records and 10 MiB per request (and 10
> MiB per record). A bigger chunk is rejected as a whole.
> h3. How to reproduce
> A Mockito {{KinesisClient}} is bound as {{amazonKinesisClient}} and used by
> {{from("direct:in").to("aws2-kinesis:s?amazonKinesisClient=#client")}}. The
> body is a {{List}} of 1200 strings with a {{CamelAwsKinesisPartitionKey}}
> header. The fake answers the 2nd PutRecords call with its first entry failed
> ({{FailedRecordCount=1}}) and writes every other entry. Measured on main:
> ||Error handler||PutRecords calls (records each)||Records written||Records
> written twice||Outcome||
> |default (no redelivery)|500, 500|999|0|exception, records 1000-1199 never
> sent|
> |maximumRedeliveries(1)|500, 500, 500, 500, 200|2199|999|exchange completes|
> Size: a List of 400 records of 100 KiB goes out as a single PutRecords call
> of 40,961,490 bytes.
> h3. Proposed change
> * Cut the chunks at 500 records *and* at the request size limit (data plus
> partition key). Reject a single record above the record limit before anything
> is sent.
> * On a partial failure, retry only the failed entries of the chunk, with a
> bounded number of attempts and a backoff, then go on with the next chunks.
> Each response entry matches the request entry at the same position and
> carries an {{ErrorCode}} when it failed (service model).
> * If entries still fail, there are two options, and which one is the default
> needs a decision:
> ** fail the exchange with an exception that tells which records were and were
> not written;
> ** or complete the exchange with {{CamelAwsKinesisFailedRecordCount}} and the
> failed entries in the message, as the SQS producer does for SendMessageBatch.
> * Set {{CamelAwsKinesisRecordCount}} and {{CamelAwsKinesisFailedRecordCount}}
> on failure too.
> * Risk: PutRecords has no idempotency token (its input is only Records,
> StreamName, StreamARN, StreamId and DryRun). A retry after a timeout, whose
> outcome is unknown, can still write a record twice. This should be documented.
> h3. Impact / upgrade
> * An internal retry of the failed entries, with the exception kept as the
> outcome, changes no default. Completing the exchange instead of throwing on a
> partial failure would change a default and needs an entry in the 4.24 upgrade
> guide.
> * The same code is on camel-4.18.x (4.18.5), camel-4.22.x (4.22.2) and
> 4.23.0. This was checked in the source; it was reproduced on main only.
> * Single-record sends (non-Iterable body, PutRecord) are not affected.
> h3. Related
> * CAMEL-20794 (commit b52830b560fa) added the batch producer and this error
> handling.
> * CAMEL-24197 (commit 48a28da60ef1) added per-record partition keys.
> _Claude Code on behalf of Federico Mariani_
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
