This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 6827b9194109 CAMEL-26660: camel-aws2-kinesis - batch producer sends
the failed records again and splits requests at 10 MiB (#29269)
6827b9194109 is described below
commit 6827b91941097cdfd948d6cf346e0fc15575f86a
Author: Federico Mariani <[email protected]>
AuthorDate: Fri Nov 6 08:21:39 2026 +0100
CAMEL-26660: camel-aws2-kinesis - batch producer sends the failed records
again and splits requests at 10 MiB (#29269)
The Kinesis batch producer sends the records that failed again, with
backoff and jitter, splits requests at 10 MiB, and sets the written/failed
count headers also when putRecords throws.
AI-generated with Claude Code on behalf of Federico Mariani.
Co-Authored-By: Claude <[email protected]>
---
.../camel/catalog/docs/aws2-kinesis-component.adoc | 35 +++-
.../src/main/docs/aws2-kinesis-component.adoc | 35 +++-
.../component/aws2/kinesis/Kinesis2Producer.java | 141 +++++++++++++--
.../aws2/kinesis/Kinesis2ProducerTest.java | 196 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 20 +++
5 files changed, 408 insertions(+), 19 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/aws2-kinesis-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/aws2-kinesis-component.adoc
index 00c03c331a4b..0c2d556f2023 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/aws2-kinesis-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/aws2-kinesis-component.adoc
@@ -124,12 +124,45 @@ the other shards; a subclass should compare the key with
`getShardId()` the same
This component implements the Batch Producer.
This allows you to send multiple messages in a single request to Amazon
Kinesis.
-Messages with batch size more than 500 is allowed. Producer will split them
into multiple requests.
+A batch can have any number of records: the producer splits it into
`PutRecords` requests of at most 500 records
+and 10 MiB, the limits of a request (the size of a record is its data plus its
partition key).
+A record above 10 MiB, the largest record that Kinesis accepts, fails the
exchange with an `IllegalArgumentException`
+before any request is sent. A stream can have a lower limit for the size of
its records (1 to 10 MiB), which Kinesis
+checks.
The batch type needs to implement the `Iterable` interface. For example, it
can be a `List`, `Set` or any other collection type.
The message type can be one or more of types `byte[]`, `ByteBuffer`, UTF-8
`String`, or `InputStream`. Other types are not supported.
A `java.nio.file.Path` body is not a batch: like a `java.io.File`, it is sent
as a single record with the content of the file.
+A `PutRecords` request can be partly written: Kinesis writes some of its
records and returns an error code for the
+others (for example `ProvisionedThroughputExceededException`). The producer
then sends only the records that failed
+again, up to 3 times, waiting 100, 200 and 400 milliseconds plus a random part
of up to a quarter of the delay (so
+that producers throttled at the same time do not all send again at the same
time), and then goes on with the next
+requests of the batch.
+When every record fails, for example when the whole stream is throttled, each
request waits 700 to 875 milliseconds in
+total before its records are counted as not written, so a batch of 10,000
records (20 requests of 500 records) can
+block the caller for 14 to 17.5 seconds, plus the time of the 80 calls and of
the retries of the AWS SDK client.
+
+A record that is sent again is written after the next records of its request:
if record A fails and record B, with
+the same partition key, is written, the stream has B before A. `PutRecords`
does not guarantee the order of the
+records either. When the order of the records of a partition key matters, send
them one by one, with a body that is
+not an `Iterable`.
+
+After the last request, the `CamelAwsKinesisRecordCount` header has the number
of records of the batch, and the
+`CamelAwsKinesisFailedRecordCount` header the number of records that are still
not written. When that number is not
+0, the exchange fails with an exception that gives both numbers and the first
error code.
+When a `PutRecords` call throws, for example an `InvalidArgumentException`, a
KMS error, or an error of the AWS SDK
+client after its own retries, the exchange fails at once with that exception,
and the next requests of the batch are
+not sent. Both headers are set in that case too: the records of that call and
of the requests that were not sent are
+counted as not written, even if the call may have written some of them (for
example after a timeout).
+
+A failure does not undo the records that are already written. A redelivery of
the exchange, by the error handler of
+the route or by the source of the message (for example a transacted JMS or an
SQS consumer), sends the whole batch
+again, so the records that were written before the failure are written twice.
`PutRecords` has no idempotency token:
+a request whose outcome is unknown, for example after a timeout, can also
write a record twice when it is sent again,
+for example by the retries of the AWS SDK client. A consumer of the stream
that must not process a record twice needs
+an id in the data of the record to detect a duplicate.
+
A body that is not an `Iterable` is sent as a single record, converted to a
`byte[]`.
A body that does not convert to a `byte[]` fails with an
`InvalidPayloadException`.
diff --git
a/components/camel-aws/camel-aws2-kinesis/src/main/docs/aws2-kinesis-component.adoc
b/components/camel-aws/camel-aws2-kinesis/src/main/docs/aws2-kinesis-component.adoc
index 00c03c331a4b..0c2d556f2023 100644
---
a/components/camel-aws/camel-aws2-kinesis/src/main/docs/aws2-kinesis-component.adoc
+++
b/components/camel-aws/camel-aws2-kinesis/src/main/docs/aws2-kinesis-component.adoc
@@ -124,12 +124,45 @@ the other shards; a subclass should compare the key with
`getShardId()` the same
This component implements the Batch Producer.
This allows you to send multiple messages in a single request to Amazon
Kinesis.
-Messages with batch size more than 500 is allowed. Producer will split them
into multiple requests.
+A batch can have any number of records: the producer splits it into
`PutRecords` requests of at most 500 records
+and 10 MiB, the limits of a request (the size of a record is its data plus its
partition key).
+A record above 10 MiB, the largest record that Kinesis accepts, fails the
exchange with an `IllegalArgumentException`
+before any request is sent. A stream can have a lower limit for the size of
its records (1 to 10 MiB), which Kinesis
+checks.
The batch type needs to implement the `Iterable` interface. For example, it
can be a `List`, `Set` or any other collection type.
The message type can be one or more of types `byte[]`, `ByteBuffer`, UTF-8
`String`, or `InputStream`. Other types are not supported.
A `java.nio.file.Path` body is not a batch: like a `java.io.File`, it is sent
as a single record with the content of the file.
+A `PutRecords` request can be partly written: Kinesis writes some of its
records and returns an error code for the
+others (for example `ProvisionedThroughputExceededException`). The producer
then sends only the records that failed
+again, up to 3 times, waiting 100, 200 and 400 milliseconds plus a random part
of up to a quarter of the delay (so
+that producers throttled at the same time do not all send again at the same
time), and then goes on with the next
+requests of the batch.
+When every record fails, for example when the whole stream is throttled, each
request waits 700 to 875 milliseconds in
+total before its records are counted as not written, so a batch of 10,000
records (20 requests of 500 records) can
+block the caller for 14 to 17.5 seconds, plus the time of the 80 calls and of
the retries of the AWS SDK client.
+
+A record that is sent again is written after the next records of its request:
if record A fails and record B, with
+the same partition key, is written, the stream has B before A. `PutRecords`
does not guarantee the order of the
+records either. When the order of the records of a partition key matters, send
them one by one, with a body that is
+not an `Iterable`.
+
+After the last request, the `CamelAwsKinesisRecordCount` header has the number
of records of the batch, and the
+`CamelAwsKinesisFailedRecordCount` header the number of records that are still
not written. When that number is not
+0, the exchange fails with an exception that gives both numbers and the first
error code.
+When a `PutRecords` call throws, for example an `InvalidArgumentException`, a
KMS error, or an error of the AWS SDK
+client after its own retries, the exchange fails at once with that exception,
and the next requests of the batch are
+not sent. Both headers are set in that case too: the records of that call and
of the requests that were not sent are
+counted as not written, even if the call may have written some of them (for
example after a timeout).
+
+A failure does not undo the records that are already written. A redelivery of
the exchange, by the error handler of
+the route or by the source of the message (for example a transacted JMS or an
SQS consumer), sends the whole batch
+again, so the records that were written before the failure are written twice.
`PutRecords` has no idempotency token:
+a request whose outcome is unknown, for example after a timeout, can also
write a record twice when it is sent again,
+for example by the retries of the AWS SDK client. A consumer of the stream
that must not process a record twice needs
+an id in the data of the record to detect a duplicate.
+
A body that is not an `Iterable` is sent as a single record, converted to a
`byte[]`.
A body that does not convert to a `byte[]` fails with an
`InvalidPayloadException`.
diff --git
a/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Producer.java
b/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Producer.java
index 03ae59779917..d41484fbb285 100644
---
a/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Producer.java
+++
b/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Producer.java
@@ -18,26 +18,48 @@ package org.apache.camel.component.aws2.kinesis;
import java.io.InputStream;
import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.List;
+import java.util.concurrent.ThreadLocalRandom;
import org.apache.camel.Exchange;
import org.apache.camel.InvalidPayloadException;
import org.apache.camel.Message;
+import org.apache.camel.RuntimeCamelException;
import org.apache.camel.support.DefaultProducer;
import org.apache.camel.util.ObjectHelper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import software.amazon.awssdk.core.SdkBytes;
import software.amazon.awssdk.services.kinesis.model.PutRecordRequest;
import software.amazon.awssdk.services.kinesis.model.PutRecordResponse;
import software.amazon.awssdk.services.kinesis.model.PutRecordsRequest;
import software.amazon.awssdk.services.kinesis.model.PutRecordsRequestEntry;
import software.amazon.awssdk.services.kinesis.model.PutRecordsResponse;
+import software.amazon.awssdk.services.kinesis.model.PutRecordsResultEntry;
public class Kinesis2Producer extends DefaultProducer {
- // Maximum number of records that can be sent in a single PutRecords
request
+ private static final Logger LOG =
LoggerFactory.getLogger(Kinesis2Producer.class);
+
+ // Limits of a single PutRecords request: 500 records and 10 MiB,
partition keys included
private static final int MAX_BATCH_SIZE = 500;
+ private static final long MAX_BATCH_BYTES = 10L * 1024 * 1024;
+ // Largest record (data and partition key) that PutRecords accepts. A
stream can have a lower limit (1 to 10 MiB).
+ private static final long MAX_RECORD_BYTES = 10L * 1024 * 1024;
+ // How many times the records that failed in a PutRecords request are sent
again, and the delay before the first
+ // time. The delay doubles at each retry, and a random part of up to a
quarter of it is added, so that producers
+ // throttled at the same time do not all send again at the same time.
+ private static final int MAX_RETRIES = 3;
+ private static final long RETRY_DELAY_MILLIS = 100;
+ // The failure of a record that has no result in the PutRecords response.
Kinesis returns a result for every record,
+ // but a record without one is not known to be written, so it is sent
again like a failed record.
+ private static final PutRecordsResultEntry MISSING_RESULT =
PutRecordsResultEntry.builder()
+ .errorCode("MissingResult")
+ .errorMessage("The PutRecords response has no result for this
record")
+ .build();
private KinesisConnection connection;
@@ -85,33 +107,99 @@ public class Kinesis2Producer extends DefaultProducer {
} else if (partitionKeys.isEmpty()) {
throw new IllegalArgumentException("CamelAwsKinesisPartitionKeys
header must not be empty");
}
+ // every record is converted and checked before the first request is
sent
List<List<PutRecordsRequestEntry>> requestBatchList
= createRequestBatchList(exchange, partitionKey,
partitionKeys);
int totalRecordCount = 0;
- int totalFailedCount = 0;
for (List<PutRecordsRequestEntry> requestBatch : requestBatchList) {
+ totalRecordCount += requestBatch.size();
+ }
+ BatchProgress progress = new BatchProgress();
+ try {
+ // the next requests are sent even when records of an earlier one
could not be written
+ for (List<PutRecordsRequestEntry> requestBatch : requestBatchList)
{
+ sendRequestBatch(requestBatch, progress);
+ }
+ } finally {
+ // also set when a PutRecords call throws: the records that
Kinesis did not report as written, those of the
+ // requests that were not sent included, are counted as failed
+ Message message = exchange.getMessage();
+ message.setHeader(Kinesis2Constants.RECORD_COUNT,
totalRecordCount);
+ message.setHeader(Kinesis2Constants.FAILED_RECORD_COUNT,
totalRecordCount - progress.writtenCount);
+ }
+ if (progress.firstFailure != null) {
+ throw new RuntimeCamelException(
+ "Failed to send " + (totalRecordCount -
progress.writtenCount) + " of " + totalRecordCount
+ + " records to the Kinesis stream "
+ +
getEndpoint().getConfiguration().getStreamName() + " after " + MAX_RETRIES
+ + " retries, " +
progress.writtenCount
+ + " records were written. First
error: " + progress.firstFailure.errorCode()
+ + ": " +
progress.firstFailure.errorMessage());
+ }
+ }
+
+ /**
+ * Sends the records of one PutRecords request, then sends again the
records that failed, at most
+ * {@link #MAX_RETRIES} times. Adds the records that are written, and the
first failure of a record that is still
+ * not written, to the progress of the batch.
+ */
+ private void sendRequestBatch(List<PutRecordsRequestEntry> requestBatch,
BatchProgress progress) {
+ String streamName = getEndpoint().getConfiguration().getStreamName();
+ List<PutRecordsRequestEntry> pending = requestBatch;
+ for (int retry = 0;; retry++) {
PutRecordsRequest putRecordsRequest = PutRecordsRequest.builder()
-
.streamName(getEndpoint().getConfiguration().getStreamName())
- .records(requestBatch)
+ .streamName(streamName)
+ .records(pending)
.build();
PutRecordsResponse putRecordsResponse =
connection.getClient(getEndpoint()).putRecords(putRecordsRequest);
- totalRecordCount += putRecordsResponse.records().size();
- totalFailedCount += putRecordsResponse.failedRecordCount();
- if (putRecordsResponse.failedRecordCount() > 0) {
- throw new RuntimeException(
- "Failed to send records " +
putRecordsResponse.failedRecordCount() + " of "
- +
putRecordsResponse.records().size());
+ // the result of a record is at the same position as the record in
the request,
+ // and the result of a record that was not written has an error
code
+ List<PutRecordsResultEntry> results = putRecordsResponse.records();
+ List<PutRecordsRequestEntry> failedEntries = new ArrayList<>();
+ List<PutRecordsResultEntry> failures = new ArrayList<>();
+ for (int i = 0; i < pending.size(); i++) {
+ PutRecordsResultEntry result = i < results.size() ?
results.get(i) : MISSING_RESULT;
+ if (result.errorCode() != null) {
+ failedEntries.add(pending.get(i));
+ failures.add(result);
+ }
+ }
+ progress.writtenCount += pending.size() - failures.size();
+ if (failures.isEmpty() || retry == MAX_RETRIES) {
+ if (progress.firstFailure == null && !failures.isEmpty()) {
+ progress.firstFailure = failures.get(0);
+ }
+ return;
+ }
+ long delay = retryDelayMillis(retry);
+ LOG.debug("{} of {} records were not written to the Kinesis stream
{} ({}), sending them again in {} ms",
+ failures.size(), pending.size(), streamName,
failures.get(0).errorCode(), delay);
+ try {
+ Thread.sleep(delay);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeCamelException(
+ "Interrupted before sending again " + failures.size()
+ " records to the Kinesis stream " + streamName,
+ e);
}
+ pending = failedEntries;
}
- Message message = exchange.getMessage();
- message.setHeader(Kinesis2Constants.RECORD_COUNT, totalRecordCount);
- message.setHeader(Kinesis2Constants.FAILED_RECORD_COUNT,
totalFailedCount);
+ }
+
+ /**
+ * The delay before the given retry (0 for the first one): {@link
#RETRY_DELAY_MILLIS}, doubled at each retry, plus
+ * a random part of up to a quarter of it.
+ */
+ static long retryDelayMillis(int retry) {
+ long delay = RETRY_DELAY_MILLIS << retry;
+ return delay + ThreadLocalRandom.current().nextLong(delay / 4 + 1);
}
private List<List<PutRecordsRequestEntry>> createRequestBatchList(
Exchange exchange, Object partitionKey, List<String>
partitionKeys) {
List<List<PutRecordsRequestEntry>> requestBatchList = new
ArrayList<>();
List<PutRecordsRequestEntry> requestBatch = new
ArrayList<>(MAX_BATCH_SIZE);
+ long requestBatchBytes = 0;
int index = 0;
for (Object record : exchange.getIn().getBody(Iterable.class)) {
SdkBytes sdkBytes;
@@ -136,15 +224,26 @@ public class Kinesis2Producer extends DefaultProducer {
key = partitionKey.toString();
}
+ long recordBytes = sdkBytes.asByteArrayUnsafe().length
+ + (key != null ? (long)
key.getBytes(StandardCharsets.UTF_8).length : 0);
+ if (recordBytes > MAX_RECORD_BYTES) {
+ throw new IllegalArgumentException(
+ "The record at index " + index + " of the batch is " +
recordBytes
+ + " bytes with its
partition key, more than "
+ + MAX_RECORD_BYTES + "
bytes, the largest record that Kinesis accepts");
+ }
+ if (!requestBatch.isEmpty()
+ && (requestBatch.size() == MAX_BATCH_SIZE ||
requestBatchBytes + recordBytes > MAX_BATCH_BYTES)) {
+ requestBatchList.add(requestBatch);
+ requestBatch = new ArrayList<>(MAX_BATCH_SIZE);
+ requestBatchBytes = 0;
+ }
PutRecordsRequestEntry putRecordsRequestEntry =
PutRecordsRequestEntry.builder()
.data(sdkBytes)
.partitionKey(key)
.build();
requestBatch.add(putRecordsRequestEntry);
- if (requestBatch.size() == MAX_BATCH_SIZE) {
- requestBatchList.add(requestBatch);
- requestBatch = new ArrayList<>(MAX_BATCH_SIZE);
- }
+ requestBatchBytes += recordBytes;
index++;
}
if (!requestBatch.isEmpty()) {
@@ -183,4 +282,12 @@ public class Kinesis2Producer extends DefaultProducer {
throw new IllegalArgumentException("Partition key must be
specified");
}
}
+
+ /**
+ * How many records of a batch Kinesis wrote, and the first failure of a
record that is still not written.
+ */
+ private static final class BatchProgress {
+ private int writtenCount;
+ private PutRecordsResultEntry firstFailure;
+ }
}
diff --git
a/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/Kinesis2ProducerTest.java
b/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/Kinesis2ProducerTest.java
index 466df6e43a4d..84d1235c2099 100644
---
a/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/Kinesis2ProducerTest.java
+++
b/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/Kinesis2ProducerTest.java
@@ -19,13 +19,19 @@ package org.apache.camel.component.aws2.kinesis;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
+import java.util.ArrayList;
import java.util.Arrays;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Set;
+import java.util.function.Predicate;
+import java.util.stream.IntStream;
import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
import org.apache.camel.InvalidPayloadException;
+import org.apache.camel.RuntimeCamelException;
import org.apache.camel.impl.DefaultCamelContext;
import org.apache.camel.support.DefaultExchange;
import org.junit.jupiter.api.BeforeEach;
@@ -36,6 +42,7 @@ import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import software.amazon.awssdk.services.kinesis.KinesisClient;
+import software.amazon.awssdk.services.kinesis.model.KmsThrottlingException;
import software.amazon.awssdk.services.kinesis.model.PutRecordRequest;
import software.amazon.awssdk.services.kinesis.model.PutRecordResponse;
import software.amazon.awssdk.services.kinesis.model.PutRecordsRequest;
@@ -44,6 +51,7 @@ import
software.amazon.awssdk.services.kinesis.model.PutRecordsResponse;
import software.amazon.awssdk.services.kinesis.model.PutRecordsResultEntry;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
@@ -59,6 +67,8 @@ public class Kinesis2ProducerTest {
private final CamelContext context = new DefaultCamelContext();
private final Kinesis2Component component = new Kinesis2Component(context);
+ private final List<PutRecordsRequest> requests = new ArrayList<>();
+ private final List<PutRecordsRequestEntry> written = new ArrayList<>();
private Kinesis2Producer producer;
@BeforeEach
@@ -174,6 +184,149 @@ public class Kinesis2ProducerTest {
assertThrows(IllegalArgumentException.class, () ->
producer.process(exchange));
}
+ @Test
+ public void batchSendsAgainOnlyTheFailedRecordAndThenTheNextRequests()
throws Exception {
+ // the first record of the second request fails once
+ Set<String> failedOnce = new HashSet<>();
+ stubPutRecords(entry -> entry.data().asUtf8String().equals("r500") &&
failedOnce.add("r500"));
+
+ Exchange exchange = new DefaultExchange(context);
+ exchange.getIn().setHeader(Kinesis2Constants.PARTITION_KEY, "key");
+ exchange.getIn().setBody(records(1200));
+
+ producer.process(exchange);
+
+ assertEquals(List.of(500, 500, 1, 200), requestSizes());
+ assertEquals(List.of("r500"), data(requests.get(2).records()));
+ // every record is written once
+ assertEquals(1200, written.size());
+ assertEquals(new HashSet<>(records(1200)), new
HashSet<>(data(written)));
+ assertEquals(1200,
exchange.getMessage().getHeader(Kinesis2Constants.RECORD_COUNT, Integer.class));
+ assertEquals(0,
exchange.getMessage().getHeader(Kinesis2Constants.FAILED_RECORD_COUNT,
Integer.class));
+ }
+
+ @Test
+ public void
batchWithARecordThatKeepsFailingThrowsAfterTheOtherRequestsAreSent() {
+ stubPutRecords(entry -> entry.data().asUtf8String().equals("r100"));
+
+ Exchange exchange = new DefaultExchange(context);
+ exchange.getIn().setHeader(Kinesis2Constants.PARTITION_KEY, "key");
+ exchange.getIn().setBody(records(600));
+
+ RuntimeCamelException e = assertThrows(RuntimeCamelException.class, ()
-> producer.process(exchange));
+
+ // the failed record is sent again 3 times, then the second request is
sent
+ assertEquals(List.of(500, 1, 1, 1, 100), requestSizes());
+ assertEquals(599, written.size());
+ assertEquals(600,
exchange.getMessage().getHeader(Kinesis2Constants.RECORD_COUNT, Integer.class));
+ assertEquals(1,
exchange.getMessage().getHeader(Kinesis2Constants.FAILED_RECORD_COUNT,
Integer.class));
+ assertTrue(e.getMessage().contains("Failed to send 1 of 600 records"),
e.getMessage());
+ assertTrue(e.getMessage().contains("599 records were written"),
e.getMessage());
+
assertTrue(e.getMessage().contains("ProvisionedThroughputExceededException"),
e.getMessage());
+ }
+
+ @Test
+ public void batchSetsTheCountsWhenAPutRecordsCallThrows() {
+ // a record of the second request fails, and the call that sends it
again throws
+ KmsThrottlingException failure =
KmsThrottlingException.builder().message("Rate exceeded").build();
+
when(kinesisClient.putRecords(any(PutRecordsRequest.class))).thenAnswer(invocation
-> {
+ PutRecordsRequest request = invocation.getArgument(0);
+ if (requests.size() == 2) {
+ requests.add(request);
+ throw failure;
+ }
+ return respond(request, entry ->
entry.data().asUtf8String().equals("r600"));
+ });
+
+ Exchange exchange = new DefaultExchange(context);
+ exchange.getIn().setHeader(Kinesis2Constants.PARTITION_KEY, "key");
+ exchange.getIn().setBody(records(1200));
+
+ KmsThrottlingException e = assertThrows(KmsThrottlingException.class,
() -> producer.process(exchange));
+
+ // the exception of the call fails the exchange as it is, and the last
request is not sent
+ assertSame(failure, e);
+ assertEquals(List.of(500, 500, 1), requestSizes());
+ assertEquals(999, written.size());
+ // the record of the call that threw and the 200 records of the
request that was not sent are not written
+ assertEquals(1200,
exchange.getMessage().getHeader(Kinesis2Constants.RECORD_COUNT, Integer.class));
+ assertEquals(201,
exchange.getMessage().getHeader(Kinesis2Constants.FAILED_RECORD_COUNT,
Integer.class));
+ }
+
+ @Test
+ public void batchSendsAgainARecordWithoutResultInTheResponse() throws
Exception {
+ // the first response has no result for the last record of the request
+
when(kinesisClient.putRecords(any(PutRecordsRequest.class))).thenAnswer(invocation
-> {
+ PutRecordsRequest request = invocation.getArgument(0);
+ if (requests.isEmpty()) {
+ PutRecordsResponse response = respond(request, entry ->
entry.data().asUtf8String().equals("r2"));
+ return
response.toBuilder().failedRecordCount(0).records(response.records().subList(0,
2)).build();
+ }
+ return respond(request, entry -> false);
+ });
+
+ Exchange exchange = new DefaultExchange(context);
+ exchange.getIn().setHeader(Kinesis2Constants.PARTITION_KEY, "key");
+ exchange.getIn().setBody(records(3));
+
+ producer.process(exchange);
+
+ assertEquals(List.of(3, 1), requestSizes());
+ assertEquals(List.of("r2"), data(requests.get(1).records()));
+ assertEquals(records(3), data(written));
+ assertEquals(3,
exchange.getMessage().getHeader(Kinesis2Constants.RECORD_COUNT, Integer.class));
+ assertEquals(0,
exchange.getMessage().getHeader(Kinesis2Constants.FAILED_RECORD_COUNT,
Integer.class));
+ }
+
+ @Test
+ public void retryDelayDoublesAndHasARandomPart() {
+ for (int retry = 0; retry < 3; retry++) {
+ long base = 100L << retry;
+ Set<Long> delays = new HashSet<>();
+ for (int i = 0; i < 200; i++) {
+ long delay = Kinesis2Producer.retryDelayMillis(retry);
+ assertTrue(delay >= base && delay <= base + base / 4, "delay "
+ delay + " ms for retry " + retry);
+ delays.add(delay);
+ }
+ // producers throttled at the same time do not all wait the same
time before they send again
+ assertTrue(delays.size() > 1, "the delay of retry " + retry + "
has no random part: " + delays);
+ }
+ }
+
+ @Test
+ public void batchIsSplitAtTheRequestSizeLimit() throws Exception {
+ stubPutRecords(entry -> false);
+
+ // with its partition key of 2 bytes a record is 1 MiB, so 10 records
fill a request of 10 MiB
+ List<byte[]> body = new ArrayList<>();
+ for (int i = 0; i < 21; i++) {
+ body.add(new byte[1024 * 1024 - 2]);
+ }
+ Exchange exchange = new DefaultExchange(context);
+ exchange.getIn().setHeader(Kinesis2Constants.PARTITION_KEY, "pk");
+ exchange.getIn().setBody(body);
+
+ producer.process(exchange);
+
+ assertEquals(List.of(10, 10, 1), requestSizes());
+ assertEquals(21, written.size());
+ assertEquals(21,
exchange.getMessage().getHeader(Kinesis2Constants.RECORD_COUNT, Integer.class));
+ }
+
+ @Test
+ public void
batchWithARecordAboveTheRecordLimitFailsBeforeAnyRequestIsSent() {
+ // 10 MiB of data and a partition key of 2 bytes
+ List<Object> body = List.of("r0", new byte[10 * 1024 * 1024]);
+ Exchange exchange = new DefaultExchange(context);
+ exchange.getIn().setHeader(Kinesis2Constants.PARTITION_KEY, "pk");
+ exchange.getIn().setBody(body);
+
+ IllegalArgumentException e =
assertThrows(IllegalArgumentException.class, () -> producer.process(exchange));
+
+ assertTrue(e.getMessage().contains("The record at index 1 of the batch
is 10485762 bytes"), e.getMessage());
+ verify(kinesisClient,
never()).putRecords(any(PutRecordsRequest.class));
+ }
+
@Test
public void pathBodyIsSentAsSingleRecordWithTheFileContent(@TempDir Path
dir) throws Exception {
when(kinesisClient.putRecord(any(PutRecordRequest.class)))
@@ -205,4 +358,47 @@ public class Kinesis2ProducerTest {
assertTrue(e.getMessage().contains("java.util."), e.getMessage());
verify(kinesisClient, never()).putRecord(any(PutRecordRequest.class));
}
+
+ /**
+ * Kinesis writes the records of a PutRecords request, except the records
the predicate selects: their result has an
+ * error code, as in a partly failed request.
+ */
+ private void stubPutRecords(Predicate<PutRecordsRequestEntry> fails) {
+ when(kinesisClient.putRecords(any(PutRecordsRequest.class)))
+ .thenAnswer(invocation -> respond(invocation.getArgument(0),
fails));
+ }
+
+ private PutRecordsResponse respond(PutRecordsRequest request,
Predicate<PutRecordsRequestEntry> fails) {
+ requests.add(request);
+ List<PutRecordsResultEntry> results = new ArrayList<>();
+ int failed = 0;
+ for (PutRecordsRequestEntry entry : request.records()) {
+ if (fails.test(entry)) {
+ failed++;
+ results.add(PutRecordsResultEntry.builder()
+ .errorCode("ProvisionedThroughputExceededException")
+ .errorMessage("Rate exceeded for shard
shardId-000000000000")
+ .build());
+ } else {
+ written.add(entry);
+ results.add(PutRecordsResultEntry.builder()
+ .shardId("shardId-000000000000")
+ .sequenceNumber(String.valueOf(written.size()))
+ .build());
+ }
+ }
+ return
PutRecordsResponse.builder().failedRecordCount(failed).records(results).build();
+ }
+
+ private List<Integer> requestSizes() {
+ return requests.stream().map(request ->
request.records().size()).toList();
+ }
+
+ private static List<String> records(int count) {
+ return IntStream.range(0, count).mapToObj(i -> "r" + i).toList();
+ }
+
+ private static List<String> data(List<PutRecordsRequestEntry> entries) {
+ return entries.stream().map(entry ->
entry.data().asUtf8String()).toList();
+ }
}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index 4de48ed4ae14..aea6f09f6e66 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -874,6 +874,26 @@ closed that client for every other endpoint.
Camel now closes only the clients it created. A client set with
`amazonKinesisClient` or `amazonKinesisAsyncClient`, on
the component or on an endpoint, is no longer closed when Camel stops: close
it yourself when you no longer need it.
+=== camel-aws2-kinesis - the batch producer sends the failed records again and
splits at 10 MiB
+
+When the body is an `Iterable`, the producer now splits the batch into
`PutRecords` requests of at most 500 records
+and 10 MiB (the data and partition keys of the records). Before, only the
number of records was limited, and a request
+above 10 MiB was rejected by Kinesis. A record above 10 MiB now fails the
exchange with an `IllegalArgumentException`
+before any request is sent.
+
+When some records of a request are not written, the producer now sends only
those records again, up to 3 times with a
+delay of 100 to 400 milliseconds (plus a random part of up to a quarter of
it), and then sends the next requests of the
+batch. Before, the exchange failed at once and
+the next requests were not sent. If records are still not written, the
exchange fails after the last request, with a
+`RuntimeCamelException` (a `RuntimeException`, as before) whose message gives
the number of records written and not
+written. The `CamelAwsKinesisRecordCount` and
`CamelAwsKinesisFailedRecordCount` headers are now also set when the
+exchange fails, including when a `PutRecords` call throws: the exchange then
fails with the exception of the call, as
+before, and the records of that call and of the requests that were not sent
are counted as not written.
+Because of the retries, a batch whose records all fail now blocks the caller
longer before it fails (700 to 875
+milliseconds per request, 14 to 17.5 seconds for 10,000 records), and a record
that is sent again is written after the
+next records of its request. A redelivery of the exchange still sends the
whole batch again: see the Batch Producer
+section of xref:components::aws2-kinesis-component.adoc[AWS Kinesis].
+
=== camel-stitch - a response with an empty body, and the default sequence of
a message
A response of Stitch with an empty body now sets the `CamelStitchCode`,
`CamelStitchStatus` and `CamelStitchHeaders`
