FrankChen021 commented on code in PR #20708:
URL: https://github.com/apache/druid/pull/20708#discussion_r4367132472
##########
extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaConsumerMonitor.java:
##########
@@ -197,7 +198,9 @@ public boolean doMonitor(final ServiceEmitter emitter)
// Capture `poll-idle-ratio-avg` metric for autoscaler purposes.
if (POLL_IDLE_RATIO_METRIC_NAME.equals(metricName.name())) {
if (entry.getValue().metricValue() != null) {
- pollIdleRatioAvg.set(((Number)
entry.getValue().metricValue()).doubleValue());
+ final double newValue = ((Number)
entry.getValue().metricValue()).doubleValue();
+ pollIdleRatioAvg.set(newValue);
+ runningAvgPollIdleRatio.addSample(newValue);
Review Comment:
[P2] NaN permanently poisons the running idle-ratio average
**Finding:** Kafka's Avg metric returns NaN before its first sample and
after an empty sampling window. Adding that value makes sum permanently NaN, so
later valid samples never recover and the autoscaler rejects the metric.
**Suggestion:** Ignore non-finite metric values before adding them to
RunningAverage.
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScaler.java:
##########
@@ -576,7 +608,7 @@ CostMetrics collectMetrics()
final int partitionCount = supervisor.getPartitionCount();
final Map<String, Map<String, Object>> taskStats = supervisor.getStats();
- final double movingAvgRate = extractMovingAverage(taskStats);
+ final double movingAvgRate = extractRunningAvgProcessingRate(taskStats);
Review Comment:
[P2] New extraction disables scaling with older task reports
**Finding:** During a mixed-version rollout, older tasks expose
pollIdleRatio and movingAverages but not the new running-average fields. The
new extractors then return -1, causing metric validation to reject scaling or
silently exclude older tasks from averages.
**Suggestion:** Fall back to the legacy task-report fields while the new
metrics are absent.
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java:
##########
@@ -655,6 +665,7 @@ public void run()
final SequenceMetadata<PartitionIdType, SequenceOffsetType>
latestSequence = getLastSequenceMetadataOrNull();
if (stopRequested.get() || latestSequence == null ||
latestSequence.isCheckpointed()) {
status = Status.PUBLISHING;
+ markReadingStopped();
Review Comment:
[P2] Reading time remains active through normal publishing
**Finding:** When all assignments are consumed, the loop exits through
stillReading == false and transitions to publishing without calling
markReadingStopped(). The processing-rate denominator therefore includes
persistence and handoff time, potentially understating throughput and
distorting autoscaler decisions.
**Suggestion:** Close the reading interval whenever the read loop exits
normally, before entering publishing.
--
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]
