This is an automated email from the ASF dual-hosted git repository.
FrankChen021 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new 094858a5151 fix: Track persist and merge time in nanoseconds to fix
flaky StreamAppenderatorTest (#20626)
094858a5151 is described below
commit 094858a5151fbe959cdafa97628e71060e275aca
Author: Nimish Rao <[email protected]>
AuthorDate: Sun Oct 25 13:18:57 2026 +0530
fix: Track persist and merge time in nanoseconds to fix flaky
StreamAppenderatorTest (#20626)
* Fix flaky StreamAppenderatorTest assertions on Windows #20602
* Track persist and merge time in nanoseconds
* Remove unrelated formatting changes from StreamAppenderatorTest
* Fix batch persist and merge time units and remove unrelated formatting
changes
* Derive monitor deltas from cumulative millis, inline merge time
conversion, update BatchAppenderatorTest
* Inline merge time conversion at log statements
---
.../apache/druid/java/util/common/Stopwatch.java | 9 +++++++
.../segment/realtime/SegmentGenerationMetrics.java | 31 +++++++++++++++-------
.../realtime/appenderator/BatchAppenderator.java | 15 ++++++-----
.../realtime/appenderator/StreamAppenderator.java | 12 ++++-----
.../appenderator/BatchAppenderatorTest.java | 12 +++++++--
.../appenderator/StreamAppenderatorTest.java | 12 +++++++--
6 files changed, 64 insertions(+), 27 deletions(-)
diff --git
a/processing/src/main/java/org/apache/druid/java/util/common/Stopwatch.java
b/processing/src/main/java/org/apache/druid/java/util/common/Stopwatch.java
index ab48fd7f9b0..d81b7e98631 100644
--- a/processing/src/main/java/org/apache/druid/java/util/common/Stopwatch.java
+++ b/processing/src/main/java/org/apache/druid/java/util/common/Stopwatch.java
@@ -88,6 +88,15 @@ public class Stopwatch
return delegate.elapsed(TimeUnit.MILLISECONDS);
}
+ /**
+ * Returns the nanoseconds elapsed on the stopwatch.
+ */
+
+ public long nanosElapsed()
+ {
+ return delegate.elapsed(TimeUnit.NANOSECONDS);
+ }
+
/**
* Checks if the given duration has already elapsed on the stopwatch.
*/
diff --git
a/server/src/main/java/org/apache/druid/segment/realtime/SegmentGenerationMetrics.java
b/server/src/main/java/org/apache/druid/segment/realtime/SegmentGenerationMetrics.java
index 97d347b0e3a..4c3cd407cd1 100644
---
a/server/src/main/java/org/apache/druid/segment/realtime/SegmentGenerationMetrics.java
+++
b/server/src/main/java/org/apache/druid/segment/realtime/SegmentGenerationMetrics.java
@@ -21,6 +21,7 @@ package org.apache.druid.segment.realtime;
import com.google.common.annotations.VisibleForTesting;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
@@ -36,7 +37,7 @@ public class SegmentGenerationMetrics
private final AtomicLong dedupCount = new AtomicLong(0);
private final AtomicLong rowOutputCount = new AtomicLong(0);
private final AtomicLong numPersists = new AtomicLong(0);
- private final AtomicLong persistTimeMillis = new AtomicLong(0);
+ private final AtomicLong persistTimeNanos = new AtomicLong(0);
private final AtomicLong persistBackPressureMillis = new AtomicLong(0);
private final AtomicLong failedPersists = new AtomicLong(0);
private final AtomicLong failedHandoffs = new AtomicLong(0);
@@ -44,7 +45,7 @@ public class SegmentGenerationMetrics
private final AtomicLong mergedRows = new AtomicLong(0);
// Measures the number of rows that have been pushed to deep storage.
private final AtomicLong pushedRows = new AtomicLong(0);
- private final AtomicLong mergeTimeMillis = new AtomicLong(0);
+ private final AtomicLong mergeTimeNanos = new AtomicLong(0);
private final AtomicLong mergeCpuTime = new AtomicLong(0);
private final AtomicLong persistCpuTime = new AtomicLong(0);
private final AtomicLong handOffCount = new AtomicLong(0);
@@ -127,9 +128,9 @@ public class SegmentGenerationMetrics
numPersists.incrementAndGet();
}
- public void incrementPersistTimeMillis(long millis)
+ public void incrementPersistTimeNanos(long nanos)
{
- persistTimeMillis.addAndGet(millis);
+ persistTimeNanos.addAndGet(nanos);
}
public void incrementPersistBackPressureMillis(long millis)
@@ -186,9 +187,9 @@ public class SegmentGenerationMetrics
}
}
- public void setMergeTime(long elapsedMergeTimeMillis)
+ public void setMergeTime(long elapsedMergeTimeNanos)
{
- mergeTimeMillis.set(elapsedMergeTimeMillis);
+ mergeTimeNanos.set(elapsedMergeTimeNanos);
}
public void setMergeCpuTime(long elapsedCpuTimeNanos)
@@ -229,7 +230,12 @@ public class SegmentGenerationMetrics
public long persistTimeMillis()
{
- return persistTimeMillis.get();
+ return TimeUnit.NANOSECONDS.toMillis(persistTimeNanos.get());
+ }
+
+ public long persistTimeNanos()
+ {
+ return persistTimeNanos.get();
}
public long persistBackPressureMillis()
@@ -259,7 +265,12 @@ public class SegmentGenerationMetrics
public long mergeTimeMillis()
{
- return mergeTimeMillis.get();
+ return TimeUnit.NANOSECONDS.toMillis(mergeTimeNanos.get());
+ }
+
+ public long mergeTimeNanos()
+ {
+ return mergeTimeNanos.get();
}
public long mergeCpuTime()
@@ -306,11 +317,11 @@ public class SegmentGenerationMetrics
retVal.dedupCount.set(dedupCount.get());
retVal.rowOutputCount.set(rowOutputCount.get());
retVal.numPersists.set(numPersists.get());
- retVal.persistTimeMillis.set(persistTimeMillis.get());
+ retVal.persistTimeNanos.set(persistTimeNanos.get());
retVal.persistBackPressureMillis.set(persistBackPressureMillis.get());
retVal.failedPersists.set(failedPersists.get());
retVal.failedHandoffs.set(failedHandoffs.get());
- retVal.mergeTimeMillis.set(mergeTimeMillis.get());
+ retVal.mergeTimeNanos.set(mergeTimeNanos.get());
retVal.mergeCpuTime.set(mergeCpuTime.get());
retVal.persistCpuTime.set(persistCpuTime.get());
retVal.handOffCount.set(handOffCount.get());
diff --git
a/server/src/main/java/org/apache/druid/segment/realtime/appenderator/BatchAppenderator.java
b/server/src/main/java/org/apache/druid/segment/realtime/appenderator/BatchAppenderator.java
index ec1ac78f209..f15ca0067c2 100644
---
a/server/src/main/java/org/apache/druid/segment/realtime/appenderator/BatchAppenderator.java
+++
b/server/src/main/java/org/apache/druid/segment/realtime/appenderator/BatchAppenderator.java
@@ -622,8 +622,9 @@ public class BatchAppenderator implements Appenderator
}
finally {
metrics.setPersistCpuTime(JvmUtils.safeGetThreadCpuTime() -
startPersistCpuNanos);
- final long persistMillis = persistStopwatch.millisElapsed();
- metrics.incrementPersistTimeMillis(persistMillis);
+ final long persistNanos = persistStopwatch.nanosElapsed();
+ final long persistMillis =
TimeUnit.NANOSECONDS.toMillis(persistNanos);
+ metrics.incrementPersistTimeNanos(persistNanos);
metrics.incrementNumPersists();
persistStopwatch.stop();
// make sure no push can start while persisting:
@@ -803,7 +804,7 @@ public class BatchAppenderator implements Appenderator
final File mergedFile;
final DataSegment mergedSegment;
final Stopwatch mergeStopwatch = Stopwatch.createStarted();
- final long mergeTimeMillis;
+ final long mergeTimeNanos;
final long startMergeCpuNanos = JvmUtils.safeGetThreadCpuTime();
List<QueryableIndex> indexes = new ArrayList<>();
long rowsinMergedSegment = 0L;
@@ -836,10 +837,10 @@ public class BatchAppenderator implements Appenderator
metrics.incrementMergedRows(rowsinMergedSegment);
metrics.setMergeCpuTime(JvmUtils.safeGetThreadCpuTime() -
startMergeCpuNanos);
- mergeTimeMillis = mergeStopwatch.millisElapsed();
- metrics.setMergeTime(mergeTimeMillis);
+ mergeTimeNanos = mergeStopwatch.nanosElapsed();
+ metrics.setMergeTime(mergeTimeNanos);
- log.debug("Segment[%s] built in %,dms.", identifier, mergeTimeMillis);
+ log.debug("Segment[%s] built in %,dms.", identifier,
TimeUnit.NANOSECONDS.toMillis(mergeTimeNanos));
QueryableIndex index = indexIO.loadIndex(mergedFile);
closer.register(index);
// Clustered segments have no top-level columns
(getAvailableDimensions() is empty); their logical
@@ -896,7 +897,7 @@ public class BatchAppenderator implements Appenderator
segment.getSize(),
segment.getTotalRows(),
indexes.size(),
- mergeTimeMillis,
+ TimeUnit.NANOSECONDS.toMillis(mergeTimeNanos),
pushTimeMillis,
objectMapper.writeValueAsString(segment.getLoadSpec())
);
diff --git
a/server/src/main/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderator.java
b/server/src/main/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderator.java
index d8290aaa9b0..7a7108c5db8 100644
---
a/server/src/main/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderator.java
+++
b/server/src/main/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderator.java
@@ -753,7 +753,7 @@ public class StreamAppenderator implements Appenderator
}
finally {
metrics.setPersistCpuTime(JvmUtils.safeGetThreadCpuTime() -
startPersistCpuNanos);
-
metrics.incrementPersistTimeMillis(persistStopwatch.millisElapsed());
+
metrics.incrementPersistTimeNanos(persistStopwatch.nanosElapsed());
metrics.incrementNumPersists();
persistStopwatch.stop();
}
@@ -943,7 +943,7 @@ public class StreamAppenderator implements Appenderator
final File mergedFile;
final DataSegment mergedSegment;
final Stopwatch mergeStopwatch = Stopwatch.createStarted();
- final long mergeTimeMillis;
+ final long mergeTimeNanos;
final long startMergeCpuNanos = JvmUtils.safeGetThreadCpuTime();
List<QueryableIndex> indexes = new ArrayList<>();
Closer closer = Closer.create();
@@ -970,10 +970,10 @@ public class StreamAppenderator implements Appenderator
);
metrics.setMergeCpuTime(JvmUtils.safeGetThreadCpuTime() -
startMergeCpuNanos);
- mergeTimeMillis = mergeStopwatch.millisElapsed();
- metrics.setMergeTime(mergeTimeMillis);
+ mergeTimeNanos = mergeStopwatch.nanosElapsed();
+ metrics.setMergeTime(mergeTimeNanos);
- log.debug("Segment[%s] built in %,dms.", identifier, mergeTimeMillis);
+ log.debug("Segment[%s] built in %,dms.", identifier,
TimeUnit.NANOSECONDS.toMillis(mergeTimeNanos));
QueryableIndex index = indexIO.loadIndex(mergedFile);
closer.register(index);
// Clustered segments have no top-level columns
(getAvailableDimensions() is empty); their logical
@@ -1014,7 +1014,7 @@ public class StreamAppenderator implements Appenderator
segment.getSize(),
segment.getTotalRows(),
indexes.size(),
- mergeTimeMillis,
+ TimeUnit.NANOSECONDS.toMillis(mergeTimeNanos),
pushTimeMillis,
objectMapper.writeValueAsString(segment.getLoadSpec())
);
diff --git
a/server/src/test/java/org/apache/druid/segment/realtime/appenderator/BatchAppenderatorTest.java
b/server/src/test/java/org/apache/druid/segment/realtime/appenderator/BatchAppenderatorTest.java
index d66a5dfc18c..0937e97f0e3 100644
---
a/server/src/test/java/org/apache/druid/segment/realtime/appenderator/BatchAppenderatorTest.java
+++
b/server/src/test/java/org/apache/druid/segment/realtime/appenderator/BatchAppenderatorTest.java
@@ -159,10 +159,18 @@ public class BatchAppenderatorTest extends
InitializedNullHandlingTest
SegmentGenerationMetrics segmentGenerationMetrics = tester.getMetrics();
Assertions.assertEquals(2, segmentGenerationMetrics.numPersists());
Assertions.assertEquals(4, segmentGenerationMetrics.rowOutput());
- Assertions.assertTrue(segmentGenerationMetrics.persistTimeMillis() > 0);
+ Assertions.assertTrue(segmentGenerationMetrics.persistTimeNanos() > 0);
+ Assertions.assertEquals(
+ segmentGenerationMetrics.persistTimeNanos() / 1_000_000,
+ segmentGenerationMetrics.persistTimeMillis()
+ );
Assertions.assertTrue(segmentGenerationMetrics.persistCpuTime() > 0);
- Assertions.assertTrue(segmentGenerationMetrics.mergeTimeMillis() > 0);
+ Assertions.assertTrue(segmentGenerationMetrics.mergeTimeNanos() > 0);
+ Assertions.assertEquals(
+ segmentGenerationMetrics.mergeTimeNanos() / 1_000_000,
+ segmentGenerationMetrics.mergeTimeMillis()
+ );
Assertions.assertTrue(segmentGenerationMetrics.mergeCpuTime() > 0);
appenderator.close();
diff --git
a/server/src/test/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderatorTest.java
b/server/src/test/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderatorTest.java
index 6868515d0c0..43accdf5ac7 100644
---
a/server/src/test/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderatorTest.java
+++
b/server/src/test/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderatorTest.java
@@ -189,10 +189,18 @@ public class StreamAppenderatorTest extends
InitializedNullHandlingTest
SegmentGenerationMetrics segmentGenerationMetrics = tester.getMetrics();
Assertions.assertEquals(2, segmentGenerationMetrics.numPersists());
Assertions.assertEquals(3, segmentGenerationMetrics.rowOutput());
- Assertions.assertTrue(segmentGenerationMetrics.persistTimeMillis() > 0);
+ Assertions.assertTrue(segmentGenerationMetrics.persistTimeNanos() > 0);
+ Assertions.assertEquals(
+ segmentGenerationMetrics.persistTimeNanos() / 1_000_000,
+ segmentGenerationMetrics.persistTimeMillis()
+ );
Assertions.assertTrue(segmentGenerationMetrics.persistCpuTime() > 0);
- Assertions.assertTrue(segmentGenerationMetrics.mergeTimeMillis() > 0);
+ Assertions.assertTrue(segmentGenerationMetrics.mergeTimeNanos() > 0);
+ Assertions.assertEquals(
+ segmentGenerationMetrics.mergeTimeNanos() / 1_000_000,
+ segmentGenerationMetrics.mergeTimeMillis()
+ );
Assertions.assertTrue(segmentGenerationMetrics.mergeCpuTime() > 0);
// clear
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
