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 daf25972184 test: retain allocations in metric collector test (#20457)
daf25972184 is described below
commit daf25972184c9be25d159be2a94b005b9f49c928
Author: Edmond Abraham <[email protected]>
AuthorDate: Thu Oct 1 09:22:32 2026 -0400
test: retain allocations in metric collector test (#20457)
---
.../metrics/AllocationMetricCollectorTest.java | 53 ++++++++++++----------
1 file changed, 28 insertions(+), 25 deletions(-)
diff --git
a/processing/src/test/java/org/apache/druid/java/util/metrics/AllocationMetricCollectorTest.java
b/processing/src/test/java/org/apache/druid/java/util/metrics/AllocationMetricCollectorTest.java
index ba748b82255..68134a45a68 100644
---
a/processing/src/test/java/org/apache/druid/java/util/metrics/AllocationMetricCollectorTest.java
+++
b/processing/src/test/java/org/apache/druid/java/util/metrics/AllocationMetricCollectorTest.java
@@ -26,16 +26,18 @@ import org.junit.jupiter.api.Test;
import java.util.ArrayList;
import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
public class AllocationMetricCollectorTest
{
private static final Logger log = new
Logger(AllocationMetricCollectorTest.class);
+ private static final int WORKER_COUNT = 2;
private final List<Thread> threads = new ArrayList<>();
- private final int objectHeader64BitSize = 16;
+ private final List<byte[]> retainedAllocations = new
CopyOnWriteArrayList<>();
/**
- * Test a calculated delta is larger than objects size generated by this
method.
+ * Test a calculated delta includes allocations from multiple threads.
* @throws InterruptedException
*/
@SuppressWarnings("OptionalIsPresent")
@@ -51,57 +53,58 @@ public class AllocationMetricCollectorTest
Assertions.assertTrue(delta > 0);
log.info("First delta: %s", delta);
- int generatedSize2 = generateObjectsConcurrently(1000);
+ final int generatedSize2 = generateBytesConcurrently(16_000);
long delta2 = collector.calculateDelta();
- Assertions.assertTrue(delta2 > generatedSize2);
+ Assertions.assertTrue(delta2 >= generatedSize2);
+ Assertions.assertEquals(generatedSize2, retainedAllocationBytes());
log.info("Second delta: %s", delta2);
- int generatedSize3 = generateObjectsConcurrently(100000);
+ final int generatedSize3 = generateBytesConcurrently(1_600_000);
long delta3 = collector.calculateDelta();
- Assertions.assertTrue(delta3 > generatedSize3);
+ Assertions.assertTrue(delta3 >= generatedSize3);
+ Assertions.assertEquals(generatedSize2 + generatedSize3,
retainedAllocationBytes());
log.info("Third delta: %s", delta3);
}
- private int generateObjectsConcurrently(int countPerThread) throws
InterruptedException
+ private int generateBytesConcurrently(int bytesPerThread) throws
InterruptedException
{
- int threads = Math.max(1, Runtime.getRuntime().availableProcessors() - 1);
- log.info("Threads: %s", threads);
- int objectsCount = countPerThread * threads;
- log.info("Total objects: %s", objectsCount);
- int totalSize = objectsCount * objectHeader64BitSize;
- log.info("Total size: %s", totalSize);
- CountDownLatch countDownLatch = new CountDownLatch(threads);
- for (int i = 0; i < threads; i++) {
- Thread thread = new Thread(() -> {
- @SuppressWarnings("MismatchedQueryAndUpdateOfCollection")
- List<Object> list = new ArrayList<>(countPerThread);
- for (int j = 0; j < countPerThread; j++) {
- list.add(new Object());
- }
+ final int totalSize = bytesPerThread * WORKER_COUNT;
+ final CountDownLatch countDownLatch = new CountDownLatch(WORKER_COUNT);
+ for (int i = 0; i < WORKER_COUNT; i++) {
+ final Thread thread = new Thread(() -> {
+ // Keep the arrays reachable until after the allocation delta is
measured.
+ retainedAllocations.add(new byte[bytesPerThread]);
countDownLatch.countDown();
- //noinspection EmptyCatchBlock,UnusedCatchParameter
try {
Thread.sleep(Long.MAX_VALUE);
}
- catch (InterruptedException e) {
+ catch (InterruptedException ignored) {
+ Thread.currentThread().interrupt();
}
});
thread.setDaemon(true);
thread.start();
this.threads.add(thread);
-
}
countDownLatch.await();
return totalSize;
}
+ private long retainedAllocationBytes()
+ {
+ return retainedAllocations.stream().mapToLong(array -> array.length).sum();
+ }
+
@AfterEach
- public void stopThreads()
+ public void stopThreads() throws InterruptedException
{
// threads are in sleep so that their ids are still present in JVM
allocation "registry"
// so stop them manually
for (Thread thread : threads) {
thread.interrupt();
}
+ for (Thread thread : threads) {
+ thread.join();
+ }
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
