zachjsh commented on code in PR #15360:
URL: https://github.com/apache/druid/pull/15360#discussion_r1401104992
##########
extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisRecordSupplier.java:
##########
@@ -1059,11 +1045,22 @@ private void
filterBufferAndResetBackgroundFetch(Set<StreamPartition<String>> pa
}
// filter records in buffer and only retain ones whose partition was not
seeked
- BlockingQueue<OrderedPartitionableRecord<String, String, ByteEntity>> newQ
= new LinkedBlockingQueue<>(recordBufferSize);
+ MemoryBoundLinkedBlockingQueue<OrderedPartitionableRecord<String, String,
ByteEntity>> newQ =
+ new MemoryBoundLinkedBlockingQueue<>(recordBufferSizeBytes);
records.stream()
- .filter(x -> !partitions.contains(x.getStreamPartition()))
- .forEachOrdered(newQ::offer);
+ .filter(x -> !partitions.contains(x.getData().getStreamPartition()))
+ .forEachOrdered(x -> {
+ if (!newQ.offer(x)) {
Review Comment:
it is a new failure more. I believe if the data was not added here, it could
have resulted in data loss. Any other suggestion here? I was a little concerned
about this too, but I think potential data loss is worse.
##########
extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisIndexTaskTuningConfig.java:
##########
@@ -81,7 +80,7 @@ public KinesisIndexTaskTuningConfig(
Long handoffConditionTimeout,
Boolean resetOffsetAutomatically,
Boolean skipSequenceNumberAvailabilityCheck,
- Integer recordBufferSize,
+ @Nullable Integer recordBufferSizeBytes,
Review Comment:
good thought, will add.
--
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]
