Federico Mariani created CAMEL-26660:
----------------------------------------
Summary: 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.23.0, 4.22.2
Reporter: Federico Mariani
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)
