From c198307b50e2b37b63bfd8dca485f484a48b1003 Mon Sep 17 00:00:00 2001 From: Eric Hare Date: Tue, 1 Sep 2026 16:33:00 -0700 Subject: [PATCH 01/11] test: reproduce missing billing uploader start --- .../billing/BillingS3LogHandlerTest.java | 41 +++++++++++++++++++ 1 file changed, 41 insertions(+) diff --git a/src/test/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3LogHandlerTest.java b/src/test/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3LogHandlerTest.java index 165a3bd21a..a701edc827 100644 --- a/src/test/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3LogHandlerTest.java +++ b/src/test/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3LogHandlerTest.java @@ -1,5 +1,17 @@ package io.stargate.sgv2.jsonapi.service.billing; +import static org.assertj.core.api.Assertions.assertThat; + +import io.micrometer.core.instrument.simple.SimpleMeterRegistry; +import io.smallrye.mutiny.Uni; +import io.stargate.sgv2.jsonapi.metrics.BatchedLogBufferMetrics; +import java.time.Duration; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.logging.Level; +import java.util.logging.LogRecord; +import org.junit.jupiter.api.Test; + /** * Unit tests for {@link BillingS3LogHandler}: the flush triggers (seal on publish, age tick, drain * on close), the upload-concurrency gate, failure containment, and the at-most-once accounting @@ -7,6 +19,35 @@ * is covered by {@code BillingS3ExportIntegrationTest}. */ class BillingS3LogHandlerTest { + + @Test + void startRunsUploaderLoop() throws InterruptedException { + var uploaded = new CountDownLatch(1); + AsyncBatchedLogUploader uploader = + batch -> { + uploaded.countDown(); + return Uni.createFrom() + .item(new AsyncBatchedLogUploader.UploadResult(true, null, batch)); + }; + var buffer = + new BatchedLogBuffer( + 1, + 1_000_000, + Duration.ofMinutes(1), + 10, + new BatchedLogBufferMetrics(new SimpleMeterRegistry(), "billing-test")); + var handler = new BillingS3LogHandler(buffer, uploader); + + handler.start(); + try { + handler.publish(new LogRecord(Level.INFO, "{\"event\":\"dataapi\"}")); + + assertThat(uploaded.await(3, TimeUnit.SECONDS)).isTrue(); + } finally { + handler.close(); + } + } + // // private static final Logger LOG = LoggerFactory.getLogger(BillingS3LogHandlerTest.class); // From 22ba30df1e01fcfbf1d0860de8cccce47723456e Mon Sep 17 00:00:00 2001 From: Eric Hare Date: Tue, 1 Sep 2026 16:33:54 -0700 Subject: [PATCH 02/11] fix: start billing S3 uploader loop --- .../service/billing/BillingS3HandlerInstaller.java | 2 +- .../jsonapi/service/billing/BillingS3LogHandler.java | 9 ++++++++- 2 files changed, 9 insertions(+), 2 deletions(-) diff --git a/src/main/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3HandlerInstaller.java b/src/main/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3HandlerInstaller.java index b975262e04..34f3a049c5 100644 --- a/src/main/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3HandlerInstaller.java +++ b/src/main/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3HandlerInstaller.java @@ -69,7 +69,7 @@ void onStart(@Observes StartupEvent event) { LOGGER.info( "Billing has attached BillingS3LogHandler to the logger named: {}", BILLING_LOGGER_NAME); - // TODO: XXXX call start on the thread. + this.handler.start(); } void onStop(@Observes ShutdownEvent event) { diff --git a/src/main/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3LogHandler.java b/src/main/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3LogHandler.java index 76633fe105..507e2ee193 100644 --- a/src/main/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3LogHandler.java +++ b/src/main/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3LogHandler.java @@ -56,7 +56,7 @@ public final class BillingS3LogHandler extends Handler { * Started at 1 and then decremented in {@link #startUploading()} when it exists so we know we * have finished uploading. */ - private final CountDownLatch uploadingFinished = new CountDownLatch(0); + private final CountDownLatch uploadingFinished = new CountDownLatch(1); private final AsyncBatchedLogUploader uploader; private final BatchedLogBuffer batchedLogBuffer; @@ -138,6 +138,13 @@ public void close() { // Flush pipeline // ============================================================ + void start() { + Thread.ofPlatform() + .name("billing-s3-uploader") + .daemon(true) + .start(this::startUploading); + } + /** Called on a worker thread to start uploading log records. */ void startUploading() { From 0cb348472b61f45f737e451242174e68ecc52a08 Mon Sep 17 00:00:00 2001 From: Eric Hare Date: Tue, 1 Sep 2026 16:44:28 -0700 Subject: [PATCH 03/11] test: require Vert.x billing uploader scheduling --- .../BillingS3HandlerInstallerTest.java | 21 +++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/src/test/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3HandlerInstallerTest.java b/src/test/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3HandlerInstallerTest.java index 741f783864..639192defc 100644 --- a/src/test/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3HandlerInstallerTest.java +++ b/src/test/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3HandlerInstallerTest.java @@ -1,11 +1,32 @@ package io.stargate.sgv2.jsonapi.service.billing; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; + +import io.vertx.core.Vertx; +import java.util.concurrent.Callable; +import org.junit.jupiter.api.Test; + /** * Unit tests for {@link BillingS3HandlerInstaller}: install/uninstall symmetry on the {@code * billing.events} JUL logger, the disabled path, and fail-loud startup on bad config. Delivery * through an installed handler is covered by {@code BillingS3ExportIntegrationTest}. */ class BillingS3HandlerInstallerTest { + + @Test + void submitsUploaderToVertxWorkerPool() { + var vertx = mock(Vertx.class); + var handler = new BillingS3LogHandler(null, null); + var installer = new BillingS3HandlerInstaller(null, null, vertx); + + installer.startUploading(handler); + + verify(vertx).executeBlocking(any(Callable.class), eq(false)); + } + // // private static BillingS3ExportConfig config(boolean enabled, String bucket, String region) { // BillingS3ExportConfig config = mock(BillingS3ExportConfig.class); From abce41012d83d4317bc4d43cdabb441eb1575505 Mon Sep 17 00:00:00 2001 From: Eric Hare Date: Tue, 1 Sep 2026 16:45:47 -0700 Subject: [PATCH 04/11] fix: run billing uploader on Vert.x worker pool --- .../billing/BillingS3HandlerInstaller.java | 17 +++++++++++++++-- .../service/billing/BillingS3LogHandler.java | 7 ------- .../billing/BillingS3LogHandlerTest.java | 9 +++++---- 3 files changed, 20 insertions(+), 13 deletions(-) diff --git a/src/main/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3HandlerInstaller.java b/src/main/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3HandlerInstaller.java index 34f3a049c5..28735fdfe0 100644 --- a/src/main/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3HandlerInstaller.java +++ b/src/main/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3HandlerInstaller.java @@ -6,6 +6,7 @@ import io.stargate.sgv2.jsonapi.config.BillingS3ExportConfig; import io.stargate.sgv2.jsonapi.metrics.BatchedLogBufferMetrics; import io.stargate.sgv2.jsonapi.metrics.BatchedLogUploaderMetrics; +import io.vertx.core.Vertx; import jakarta.enterprise.context.ApplicationScoped; import jakarta.enterprise.event.Observes; import jakarta.inject.Inject; @@ -28,13 +29,16 @@ public class BillingS3HandlerInstaller { private final BillingS3ExportConfig config; private final MeterRegistry meterRegistry; + private final Vertx vertx; private volatile BillingS3LogHandler handler; @Inject - public BillingS3HandlerInstaller(BillingS3ExportConfig config, MeterRegistry meterRegistry) { + public BillingS3HandlerInstaller( + BillingS3ExportConfig config, MeterRegistry meterRegistry, Vertx vertx) { this.config = config; this.meterRegistry = meterRegistry; + this.vertx = vertx; } void onStart(@Observes StartupEvent event) { @@ -69,7 +73,16 @@ void onStart(@Observes StartupEvent event) { LOGGER.info( "Billing has attached BillingS3LogHandler to the logger named: {}", BILLING_LOGGER_NAME); - this.handler.start(); + startUploading(this.handler); + } + + void startUploading(BillingS3LogHandler handler) { + vertx.executeBlocking( + () -> { + handler.startUploading(); + return null; + }, + false); } void onStop(@Observes ShutdownEvent event) { diff --git a/src/main/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3LogHandler.java b/src/main/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3LogHandler.java index 507e2ee193..4e5a239356 100644 --- a/src/main/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3LogHandler.java +++ b/src/main/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3LogHandler.java @@ -138,13 +138,6 @@ public void close() { // Flush pipeline // ============================================================ - void start() { - Thread.ofPlatform() - .name("billing-s3-uploader") - .daemon(true) - .start(this::startUploading); - } - /** Called on a worker thread to start uploading log records. */ void startUploading() { diff --git a/src/test/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3LogHandlerTest.java b/src/test/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3LogHandlerTest.java index a701edc827..fc9c68091f 100644 --- a/src/test/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3LogHandlerTest.java +++ b/src/test/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3LogHandlerTest.java @@ -6,6 +6,7 @@ import io.smallrye.mutiny.Uni; import io.stargate.sgv2.jsonapi.metrics.BatchedLogBufferMetrics; import java.time.Duration; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.logging.Level; @@ -21,13 +22,12 @@ class BillingS3LogHandlerTest { @Test - void startRunsUploaderLoop() throws InterruptedException { + void uploaderLoopProcessesPublishedRecord() throws Exception { var uploaded = new CountDownLatch(1); AsyncBatchedLogUploader uploader = batch -> { uploaded.countDown(); - return Uni.createFrom() - .item(new AsyncBatchedLogUploader.UploadResult(true, null, batch)); + return Uni.createFrom().item(new AsyncBatchedLogUploader.UploadResult(true, null, batch)); }; var buffer = new BatchedLogBuffer( @@ -38,13 +38,14 @@ void startRunsUploaderLoop() throws InterruptedException { new BatchedLogBufferMetrics(new SimpleMeterRegistry(), "billing-test")); var handler = new BillingS3LogHandler(buffer, uploader); - handler.start(); + var uploading = CompletableFuture.runAsync(handler::startUploading); try { handler.publish(new LogRecord(Level.INFO, "{\"event\":\"dataapi\"}")); assertThat(uploaded.await(3, TimeUnit.SECONDS)).isTrue(); } finally { handler.close(); + uploading.get(3, TimeUnit.SECONDS); } } From 4652191602f1855403c864aeed5d2f6c6173d77c Mon Sep 17 00:00:00 2001 From: Eric Hare Date: Tue, 1 Sep 2026 17:40:17 -0700 Subject: [PATCH 05/11] test: require Quarkus worker pool scheduling --- .../BillingS3HandlerInstallerTest.java | 42 ++++++++++++++----- 1 file changed, 31 insertions(+), 11 deletions(-) diff --git a/src/test/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3HandlerInstallerTest.java b/src/test/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3HandlerInstallerTest.java index 639192defc..08cdeb6417 100644 --- a/src/test/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3HandlerInstallerTest.java +++ b/src/test/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3HandlerInstallerTest.java @@ -1,12 +1,15 @@ package io.stargate.sgv2.jsonapi.service.billing; -import static org.mockito.ArgumentMatchers.any; -import static org.mockito.ArgumentMatchers.eq; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.verify; +import static org.assertj.core.api.Assertions.assertThat; -import io.vertx.core.Vertx; -import java.util.concurrent.Callable; +import io.micrometer.core.instrument.simple.SimpleMeterRegistry; +import io.smallrye.mutiny.Uni; +import io.stargate.sgv2.jsonapi.metrics.BatchedLogBufferMetrics; +import java.time.Duration; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.logging.Level; +import java.util.logging.LogRecord; import org.junit.jupiter.api.Test; /** @@ -17,14 +20,31 @@ class BillingS3HandlerInstallerTest { @Test - void submitsUploaderToVertxWorkerPool() { - var vertx = mock(Vertx.class); - var handler = new BillingS3LogHandler(null, null); - var installer = new BillingS3HandlerInstaller(null, null, vertx); + void submitsUploaderToQuarkusWorkerPool() throws Exception { + var uploaded = new CountDownLatch(1); + AsyncBatchedLogUploader uploader = + batch -> { + uploaded.countDown(); + return Uni.createFrom().item(new AsyncBatchedLogUploader.UploadResult(true, null, batch)); + }; + var buffer = + new BatchedLogBuffer( + 1, + 1_000_000, + Duration.ofMinutes(1), + 10, + new BatchedLogBufferMetrics(new SimpleMeterRegistry(), "billing-test")); + var handler = new BillingS3LogHandler(buffer, uploader); + var installer = new BillingS3HandlerInstaller(null, null); installer.startUploading(handler); + try { + handler.publish(new LogRecord(Level.INFO, "{\"event\":\"dataapi\"}")); - verify(vertx).executeBlocking(any(Callable.class), eq(false)); + assertThat(uploaded.await(3, TimeUnit.SECONDS)).isTrue(); + } finally { + handler.close(); + } } // From bcd9a0bee4b9d3f54a2a566ec0c0821faaf8fdbb Mon Sep 17 00:00:00 2001 From: Eric Hare Date: Tue, 1 Sep 2026 17:41:20 -0700 Subject: [PATCH 06/11] fix: use Quarkus billing worker pool --- .../service/billing/BillingS3HandlerInstaller.java | 14 +++----------- 1 file changed, 3 insertions(+), 11 deletions(-) diff --git a/src/main/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3HandlerInstaller.java b/src/main/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3HandlerInstaller.java index 28735fdfe0..21e313edd2 100644 --- a/src/main/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3HandlerInstaller.java +++ b/src/main/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3HandlerInstaller.java @@ -3,10 +3,10 @@ import io.micrometer.core.instrument.MeterRegistry; import io.quarkus.runtime.ShutdownEvent; import io.quarkus.runtime.StartupEvent; +import io.smallrye.mutiny.infrastructure.Infrastructure; import io.stargate.sgv2.jsonapi.config.BillingS3ExportConfig; import io.stargate.sgv2.jsonapi.metrics.BatchedLogBufferMetrics; import io.stargate.sgv2.jsonapi.metrics.BatchedLogUploaderMetrics; -import io.vertx.core.Vertx; import jakarta.enterprise.context.ApplicationScoped; import jakarta.enterprise.event.Observes; import jakarta.inject.Inject; @@ -29,16 +29,13 @@ public class BillingS3HandlerInstaller { private final BillingS3ExportConfig config; private final MeterRegistry meterRegistry; - private final Vertx vertx; private volatile BillingS3LogHandler handler; @Inject - public BillingS3HandlerInstaller( - BillingS3ExportConfig config, MeterRegistry meterRegistry, Vertx vertx) { + public BillingS3HandlerInstaller(BillingS3ExportConfig config, MeterRegistry meterRegistry) { this.config = config; this.meterRegistry = meterRegistry; - this.vertx = vertx; } void onStart(@Observes StartupEvent event) { @@ -77,12 +74,7 @@ void onStart(@Observes StartupEvent event) { } void startUploading(BillingS3LogHandler handler) { - vertx.executeBlocking( - () -> { - handler.startUploading(); - return null; - }, - false); + Infrastructure.getDefaultWorkerPool().execute(handler::startUploading); } void onStop(@Observes ShutdownEvent event) { From 22481d566739bb3ce21b3718d7b03999ee3de8dc Mon Sep 17 00:00:00 2001 From: Eric Hare Date: Tue, 1 Sep 2026 17:42:36 -0700 Subject: [PATCH 07/11] refactor: remove duplicate billing uploader test --- .../billing/BillingS3LogHandlerTest.java | 42 ------------------- 1 file changed, 42 deletions(-) diff --git a/src/test/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3LogHandlerTest.java b/src/test/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3LogHandlerTest.java index fc9c68091f..165a3bd21a 100644 --- a/src/test/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3LogHandlerTest.java +++ b/src/test/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3LogHandlerTest.java @@ -1,18 +1,5 @@ package io.stargate.sgv2.jsonapi.service.billing; -import static org.assertj.core.api.Assertions.assertThat; - -import io.micrometer.core.instrument.simple.SimpleMeterRegistry; -import io.smallrye.mutiny.Uni; -import io.stargate.sgv2.jsonapi.metrics.BatchedLogBufferMetrics; -import java.time.Duration; -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.TimeUnit; -import java.util.logging.Level; -import java.util.logging.LogRecord; -import org.junit.jupiter.api.Test; - /** * Unit tests for {@link BillingS3LogHandler}: the flush triggers (seal on publish, age tick, drain * on close), the upload-concurrency gate, failure containment, and the at-most-once accounting @@ -20,35 +7,6 @@ * is covered by {@code BillingS3ExportIntegrationTest}. */ class BillingS3LogHandlerTest { - - @Test - void uploaderLoopProcessesPublishedRecord() throws Exception { - var uploaded = new CountDownLatch(1); - AsyncBatchedLogUploader uploader = - batch -> { - uploaded.countDown(); - return Uni.createFrom().item(new AsyncBatchedLogUploader.UploadResult(true, null, batch)); - }; - var buffer = - new BatchedLogBuffer( - 1, - 1_000_000, - Duration.ofMinutes(1), - 10, - new BatchedLogBufferMetrics(new SimpleMeterRegistry(), "billing-test")); - var handler = new BillingS3LogHandler(buffer, uploader); - - var uploading = CompletableFuture.runAsync(handler::startUploading); - try { - handler.publish(new LogRecord(Level.INFO, "{\"event\":\"dataapi\"}")); - - assertThat(uploaded.await(3, TimeUnit.SECONDS)).isTrue(); - } finally { - handler.close(); - uploading.get(3, TimeUnit.SECONDS); - } - } - // // private static final Logger LOG = LoggerFactory.getLogger(BillingS3LogHandlerTest.class); // From 1f32f31aad0152884239ce70661843861bce782c Mon Sep 17 00:00:00 2001 From: Eric Hare Date: Tue, 1 Sep 2026 19:11:22 -0700 Subject: [PATCH 08/11] test: exercise billing uploader startup lifecycle --- .../BillingS3HandlerInstallerTest.java | 55 ++++++++----------- 1 file changed, 24 insertions(+), 31 deletions(-) diff --git a/src/test/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3HandlerInstallerTest.java b/src/test/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3HandlerInstallerTest.java index 08cdeb6417..6359190bed 100644 --- a/src/test/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3HandlerInstallerTest.java +++ b/src/test/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3HandlerInstallerTest.java @@ -1,15 +1,15 @@ package io.stargate.sgv2.jsonapi.service.billing; -import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; import io.micrometer.core.instrument.simple.SimpleMeterRegistry; -import io.smallrye.mutiny.Uni; -import io.stargate.sgv2.jsonapi.metrics.BatchedLogBufferMetrics; +import io.quarkus.runtime.ShutdownEvent; +import io.quarkus.runtime.StartupEvent; +import io.stargate.sgv2.jsonapi.config.BillingS3ExportConfig; import java.time.Duration; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.TimeUnit; -import java.util.logging.Level; -import java.util.logging.LogRecord; +import java.util.Optional; import org.junit.jupiter.api.Test; /** @@ -20,31 +20,24 @@ class BillingS3HandlerInstallerTest { @Test - void submitsUploaderToQuarkusWorkerPool() throws Exception { - var uploaded = new CountDownLatch(1); - AsyncBatchedLogUploader uploader = - batch -> { - uploaded.countDown(); - return Uni.createFrom().item(new AsyncBatchedLogUploader.UploadResult(true, null, batch)); - }; - var buffer = - new BatchedLogBuffer( - 1, - 1_000_000, - Duration.ofMinutes(1), - 10, - new BatchedLogBufferMetrics(new SimpleMeterRegistry(), "billing-test")); - var handler = new BillingS3LogHandler(buffer, uploader); - var installer = new BillingS3HandlerInstaller(null, null); + void startupRunsUploaderOnQuarkusWorkerPool() { + var config = mock(BillingS3ExportConfig.class); + when(config.enabled()).thenReturn(true); + when(config.region()).thenReturn("us-east-2"); + when(config.bucket()).thenReturn("test-bucket"); + when(config.endpointOverride()).thenReturn(Optional.empty()); + when(config.maxEventsPerBatch()).thenReturn(1); + when(config.maxBytesPerBatch()).thenReturn(1_000_000L); + when(config.maxAge()).thenReturn(Duration.ofMinutes(1)); + when(config.queueCapacity()).thenReturn(10); + var installer = new BillingS3HandlerInstaller(config, new SimpleMeterRegistry()); - installer.startUploading(handler); - try { - handler.publish(new LogRecord(Level.INFO, "{\"event\":\"dataapi\"}")); - - assertThat(uploaded.await(3, TimeUnit.SECONDS)).isTrue(); - } finally { - handler.close(); - } + assertTimeoutPreemptively( + Duration.ofSeconds(10), + () -> { + installer.onStart(new StartupEvent()); + installer.onStop(new ShutdownEvent()); + }); } // From 0695775ad2a392b662b6355fdb37f8e5fb7594ce Mon Sep 17 00:00:00 2001 From: Eric Hare Date: Tue, 1 Sep 2026 19:11:59 -0700 Subject: [PATCH 09/11] refactor: inline billing uploader scheduling --- .../jsonapi/service/billing/BillingS3HandlerInstaller.java | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/src/main/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3HandlerInstaller.java b/src/main/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3HandlerInstaller.java index 21e313edd2..a5a024d4a2 100644 --- a/src/main/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3HandlerInstaller.java +++ b/src/main/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3HandlerInstaller.java @@ -70,11 +70,7 @@ void onStart(@Observes StartupEvent event) { LOGGER.info( "Billing has attached BillingS3LogHandler to the logger named: {}", BILLING_LOGGER_NAME); - startUploading(this.handler); - } - - void startUploading(BillingS3LogHandler handler) { - Infrastructure.getDefaultWorkerPool().execute(handler::startUploading); + Infrastructure.getDefaultWorkerPool().execute(this.handler::startUploading); } void onStop(@Observes ShutdownEvent event) { From 9878f23feaa41fa16168b9704f9d8d8ed32b38a3 Mon Sep 17 00:00:00 2001 From: Eric Hare Date: Thu, 3 Sep 2026 13:59:45 -0700 Subject: [PATCH 10/11] test: verify billing handler installation --- .../BillingS3HandlerInstallerTest.java | 41 +++++++++++++++---- 1 file changed, 32 insertions(+), 9 deletions(-) diff --git a/src/test/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3HandlerInstallerTest.java b/src/test/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3HandlerInstallerTest.java index 6359190bed..1cf23cec9b 100644 --- a/src/test/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3HandlerInstallerTest.java +++ b/src/test/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3HandlerInstallerTest.java @@ -1,7 +1,9 @@ package io.stargate.sgv2.jsonapi.service.billing; -import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively; +import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.timeout; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import io.micrometer.core.instrument.simple.SimpleMeterRegistry; @@ -10,6 +12,8 @@ import io.stargate.sgv2.jsonapi.config.BillingS3ExportConfig; import java.time.Duration; import java.util.Optional; +import java.util.concurrent.atomic.AtomicReference; +import java.util.logging.Logger; import org.junit.jupiter.api.Test; /** @@ -20,7 +24,7 @@ class BillingS3HandlerInstallerTest { @Test - void startupRunsUploaderOnQuarkusWorkerPool() { + void startupCreatesAttachesAndStartsHandler() { var config = mock(BillingS3ExportConfig.class); when(config.enabled()).thenReturn(true); when(config.region()).thenReturn("us-east-2"); @@ -30,14 +34,33 @@ void startupRunsUploaderOnQuarkusWorkerPool() { when(config.maxBytesPerBatch()).thenReturn(1_000_000L); when(config.maxAge()).thenReturn(Duration.ofMinutes(1)); when(config.queueCapacity()).thenReturn(10); - var installer = new BillingS3HandlerInstaller(config, new SimpleMeterRegistry()); + var handler = mock(BillingS3LogHandler.class); + var createdBuffer = new AtomicReference(); + var createdUploader = new AtomicReference(); + var installer = + new BillingS3HandlerInstaller( + config, + new SimpleMeterRegistry(), + (buffer, uploader) -> { + createdBuffer.set(buffer); + createdUploader.set(uploader); + return handler; + }); + var billingLogger = Logger.getLogger("billing.events"); - assertTimeoutPreemptively( - Duration.ofSeconds(10), - () -> { - installer.onStart(new StartupEvent()); - installer.onStop(new ShutdownEvent()); - }); + installer.onStart(new StartupEvent()); + try { + assertThat(createdBuffer.get()).isNotNull(); + assertThat(createdBuffer.get().remainingCapacity()).isEqualTo(10); + assertThat(createdUploader.get()).isInstanceOf(S3BatchedLogUploader.class); + assertThat(billingLogger.getHandlers()).contains(handler); + verify(handler, timeout(10_000)).startUploading(); + } finally { + installer.onStop(new ShutdownEvent()); + } + + assertThat(billingLogger.getHandlers()).doesNotContain(handler); + verify(handler).close(); } // From fd6464a96b55a8f33fd562a11971ea9fe8174225 Mon Sep 17 00:00:00 2001 From: Eric Hare Date: Thu, 3 Sep 2026 14:02:21 -0700 Subject: [PATCH 11/11] refactor: inject billing handler factory --- .../billing/BillingS3HandlerInstaller.java | 15 ++++++++++++- .../BillingS3HandlerInstallerTest.java | 22 ++++++------------- 2 files changed, 21 insertions(+), 16 deletions(-) diff --git a/src/main/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3HandlerInstaller.java b/src/main/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3HandlerInstaller.java index a5a024d4a2..48ed92a819 100644 --- a/src/main/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3HandlerInstaller.java +++ b/src/main/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3HandlerInstaller.java @@ -1,5 +1,6 @@ package io.stargate.sgv2.jsonapi.service.billing; +import com.google.common.annotations.VisibleForTesting; import io.micrometer.core.instrument.MeterRegistry; import io.quarkus.runtime.ShutdownEvent; import io.quarkus.runtime.StartupEvent; @@ -10,6 +11,7 @@ import jakarta.enterprise.context.ApplicationScoped; import jakarta.enterprise.event.Observes; import jakarta.inject.Inject; +import java.util.function.BiFunction; import java.util.logging.Logger; import org.slf4j.LoggerFactory; @@ -29,13 +31,24 @@ public class BillingS3HandlerInstaller { private final BillingS3ExportConfig config; private final MeterRegistry meterRegistry; + private final BiFunction + handlerFactory; private volatile BillingS3LogHandler handler; @Inject public BillingS3HandlerInstaller(BillingS3ExportConfig config, MeterRegistry meterRegistry) { + this(config, meterRegistry, BillingS3LogHandler::new); + } + + @VisibleForTesting + BillingS3HandlerInstaller( + BillingS3ExportConfig config, + MeterRegistry meterRegistry, + BiFunction handlerFactory) { this.config = config; this.meterRegistry = meterRegistry; + this.handlerFactory = handlerFactory; } void onStart(@Observes StartupEvent event) { @@ -63,7 +76,7 @@ void onStart(@Observes StartupEvent event) { config.queueCapacity(), new BatchedLogBufferMetrics(meterRegistry, METRICS_PREFIX)); LOGGER.info("Billing is using log buffer: {}", buffer); - this.handler = new BillingS3LogHandler(buffer, uploader); + this.handler = handlerFactory.apply(buffer, uploader); // TODO: LOGGER NAME SHOULD BE IN CONFIG Logger.getLogger(BILLING_LOGGER_NAME).addHandler(this.handler); diff --git a/src/test/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3HandlerInstallerTest.java b/src/test/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3HandlerInstallerTest.java index 1cf23cec9b..56a1597799 100644 --- a/src/test/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3HandlerInstallerTest.java +++ b/src/test/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3HandlerInstallerTest.java @@ -12,8 +12,7 @@ import io.stargate.sgv2.jsonapi.config.BillingS3ExportConfig; import java.time.Duration; import java.util.Optional; -import java.util.concurrent.atomic.AtomicReference; -import java.util.logging.Logger; +import java.util.concurrent.atomic.AtomicBoolean; import org.junit.jupiter.api.Test; /** @@ -24,7 +23,7 @@ class BillingS3HandlerInstallerTest { @Test - void startupCreatesAttachesAndStartsHandler() { + void startupCreatesAndStartsHandler() { var config = mock(BillingS3ExportConfig.class); when(config.enabled()).thenReturn(true); when(config.region()).thenReturn("us-east-2"); @@ -35,32 +34,25 @@ void startupCreatesAttachesAndStartsHandler() { when(config.maxAge()).thenReturn(Duration.ofMinutes(1)); when(config.queueCapacity()).thenReturn(10); var handler = mock(BillingS3LogHandler.class); - var createdBuffer = new AtomicReference(); - var createdUploader = new AtomicReference(); + var handlerCreated = new AtomicBoolean(); var installer = new BillingS3HandlerInstaller( config, new SimpleMeterRegistry(), (buffer, uploader) -> { - createdBuffer.set(buffer); - createdUploader.set(uploader); + assertThat(buffer).isNotNull(); + assertThat(uploader).isNotNull(); + handlerCreated.set(true); return handler; }); - var billingLogger = Logger.getLogger("billing.events"); installer.onStart(new StartupEvent()); try { - assertThat(createdBuffer.get()).isNotNull(); - assertThat(createdBuffer.get().remainingCapacity()).isEqualTo(10); - assertThat(createdUploader.get()).isInstanceOf(S3BatchedLogUploader.class); - assertThat(billingLogger.getHandlers()).contains(handler); + assertThat(handlerCreated).isTrue(); verify(handler, timeout(10_000)).startUploading(); } finally { installer.onStop(new ShutdownEvent()); } - - assertThat(billingLogger.getHandlers()).doesNotContain(handler); - verify(handler).close(); } //