Add junit based tests

Author: amarziali

utils/queue-utils/build.gradle.kts

@@ -1,5 +1,4 @@
 import groovy.lang.Closure
-import org.gradle.kotlin.dsl.extra
 
 plugins {
   `java-library`
@@ -16,8 +15,17 @@ java {
   }
 }
 
-fun AbstractCompile.configureCompiler(javaVersionInteger: Int, compatibilityVersion: JavaVersion? = null, unsetReleaseFlagReason: String? = null) {
-  (project.extra["configureCompiler"] as Closure<*>).call(this, javaVersionInteger, compatibilityVersion, unsetReleaseFlagReason)
+fun AbstractCompile.configureCompiler(
+  javaVersionInteger: Int,
+  compatibilityVersion: JavaVersion? = null,
+  unsetReleaseFlagReason: String? = null
+) {
+  (project.extra["configureCompiler"] as Closure<*>).call(
+    this,
+    javaVersionInteger,
+    compatibilityVersion,
+    unsetReleaseFlagReason
+  )
 }
 
 listOf(JavaCompile::class.java, GroovyCompile::class.java).forEach { compileTaskType ->
@@ -30,7 +38,8 @@ dependencies {
   api(project(":internal-api"))
   api(libs.jctools)
 
-  testImplementation(libs.junit)
+  testImplementation(libs.bundles.junit5)
+  testImplementation(libs.junit.jupiter.params)
   testImplementation(libs.slf4j)
 }
 

utils/queue-utils/src/main/java/datadog/common/queue/MpscBlockingConsumerArrayQueueVarHandle.java

@@ -591,19 +591,26 @@ public int fill(Supplier<E> s, int limit) {
       }
       // pIndex is even (lower bit is 0) -> actual index is (pIndex >> 1), consumer is awake
 
-      // we want 'limit' slots, but will settle for whatever is visible to 'producerLimit'
-      batchLimit =
-          Math.min(producerLimit, pIndex + shiftedBatchSize); //  -> producerLimit >= batchLimit
+      // Calculate available space (similar to reference implementation)
+      long available = producerLimit - pIndex;
 
       // Use producer limit to save a read of the more rapidly mutated consumer index.
       // Assumption: queue is usually empty or near empty
-      if (pIndex >= producerLimit) {
+      if (available < shiftedBatchSize) {
         if (!recalculateProducerLimit(mask, pIndex, producerLimit)) {
           return 0;
         }
-        batchLimit = Math.min(lpProducerLimit(), pIndex + shiftedBatchSize);
+        producerLimit = lpProducerLimit();
+        available = producerLimit - pIndex;
+        if (available <= 0) {
+          return 0; // FULL
+        }
       }
 
+      // we want 'limit' slots, but will settle for whatever is available
+      long claimedShifted = Math.min(available, shiftedBatchSize);
+      batchLimit = pIndex + claimedShifted;
+
       // Claim the index
       if (casProducerIndex(pIndex, batchLimit)) {
         break;

utils/queue-utils/src/test/java/datadog/common/queue/AbstractQueueTest.groovy

@@ -0,0 +1,117 @@
+package datadog.common.queue;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+abstract class AbstractQueueTest<T extends BaseQueue<Integer>> {
+  abstract T createQueue(int capacity);
+
+  protected T queue;
+
+  @BeforeEach
+  void setUp() {
+    queue = createQueue(8);
+  }
+
+  @Test
+  void offerAndPollShouldPreserveFIFOOrder() {
+    queue.offer(1);
+    queue.offer(2);
+    queue.offer(3);
+
+    assertEquals(1, queue.poll());
+    assertEquals(2, queue.poll());
+    assertEquals(3, queue.poll());
+    assertNull(queue.poll());
+  }
+
+  @Test
+  void offerShouldReturnFalseWhenQueueIsFull() {
+    queue.clear();
+    for (int i = 1; i <= 8; i++) {
+      queue.offer(i);
+    }
+
+    assertFalse(queue.offer(999));
+    assertEquals(8, queue.size());
+  }
+
+  @Test
+  void peekShouldReturnHeadElementWithoutRemovingIt() {
+    queue.clear();
+    queue.offer(10);
+    queue.offer(20);
+
+    assertEquals(10, queue.peek());
+    assertEquals(10, queue.peek());
+    assertEquals(2, queue.size());
+  }
+
+  @Test
+  void pollShouldReturnNullWhenEmpty() {
+    queue.clear();
+
+    assertNull(queue.poll());
+  }
+
+  @Test
+  void sizeShouldReflectCurrentNumberOfItems() {
+    queue.clear();
+    queue.offer(1);
+    queue.offer(2);
+
+    assertEquals(2, queue.size());
+
+    queue.poll();
+    queue.poll();
+
+    assertEquals(0, queue.size());
+  }
+
+  @Test
+  void drainShouldConsumeAllAvailableElements() {
+    queue.clear();
+    for (int i = 1; i <= 5; i++) {
+      queue.offer(i);
+    }
+    List<Integer> drained = new ArrayList<>();
+
+    int count = queue.drain(drained::add);
+
+    assertEquals(5, count);
+    assertEquals(List.of(1, 2, 3, 4, 5), drained);
+    assertTrue(queue.isEmpty());
+  }
+
+  @Test
+  void drainWithLimitShouldOnlyConsumeThatManyElements() {
+    queue.clear();
+    for (int i = 1; i <= 6; i++) {
+      queue.offer(i);
+    }
+    List<Integer> drained = new ArrayList<>();
+
+    int count = queue.drain(drained::add, 3);
+
+    assertEquals(3, count);
+    assertEquals(List.of(1, 2, 3), drained);
+    assertEquals(3, queue.size());
+  }
+
+  @Test
+  void remainingCapacityShouldReflectCurrentOccupancy() {
+    T q = createQueue(4);
+    q.offer(1);
+    q.offer(2);
+
+    assertEquals(2, q.capacity() - q.size());
+
+    q.poll();
+
+    assertEquals(3, q.capacity() - q.size());
+  }
+}