FrankChen021 commented on code in PR #20127:
URL: https://github.com/apache/druid/pull/20127#discussion_r3853026355
##########
embedded-tests/src/test/java/org/apache/druid/testing/embedded/msq/MSQWorkerFaultToleranceTest.java:
##########
@@ -94,7 +94,8 @@ public void
test_cancelledWorker_isRetried_ifFaultToleranceIsEnabled() throws Ex
final EmbeddedIndexer faultyIndexer = new EmbeddedIndexer()
.addProperty("druid.plaintextPort", "7091")
.addProperty("druid.unsafe.cluster.testing", "true")
-
.addProperty("druid.unsafe.cluster.testing.overlordClient.taskStatusDelay",
"PT1H")
+ // Keep the injected delay short so the retry path is exercised
without a one-hour wait.
+
.addProperty("druid.unsafe.cluster.testing.overlordClient.taskStatusDelay",
"PT1S")
Review Comment:
[P2] One-second delay no longer reliably blocks the faulty worker
The faulty worker can resume after this delay while the functional indexer
is starting, allowing it to finish before cancellation. The FAILED assertion
then becomes timing-dependent. Keep the delay longer than setup or synchronize
on a state proving the worker remains blocked.
##########
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] Batching ignores Kinesis aggregate request-size limit
PutRecords is limited to both 500 records and a 5 MiB aggregate payload. A
500-record batch can exceed 5 MiB and be rejected, whereas the previous
per-record loop accepted individually valid records. Bound batches by total
size as well as count.
--
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]
