FrankChen021 commented on code in PR #20127:
URL: https://github.com/apache/druid/pull/20127#discussion_r3886676184
##########
embedded-tests/src/test/java/org/apache/druid/testing/embedded/kinesis/KinesisResource.java:
##########
@@ -178,14 +175,33 @@ public void publishRecordsToTopic(String topic,
List<byte[]> records, Map<String
public void publishRecordsToTopicPartition(String topic, String
partitionKey, List<byte[]> records)
{
- for (byte[] record : records) {
- kinesisClient.putRecord(
- PutRecordRequest.builder()
- .streamName(topic)
- .partitionKey(partitionKey)
- .data(SdkBytes.fromByteArray(record))
- .build()
+ publishRecordsInBatches(topic, records, record -> partitionKey);
+ }
+
+ private void publishRecordsInBatches(
+ String topic,
+ List<byte[]> records,
+ Function<byte[], String> partitionKeyFunction
+ )
+ {
+ for (int start = 0; start < records.size(); start +=
PUT_RECORDS_BATCH_SIZE) {
Review Comment:
P2 PutRecords batching can exceed Kinesis' aggregate request limit
The batching logic can send up to 500 individually valid records without
enforcing Kinesis' 5 MiB aggregate request limit. A batch can therefore exceed
the service limit and fail, whereas the previous single-record loop accepted
the same records. Bound batches by both record count and total serialized
bytes, or retain single-record puts for oversized batches.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
