Start the billing S3 uploader loop

src/main/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3HandlerInstaller.java

@@ -1,14 +1,17 @@
 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;
+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 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;
 
@@ -28,13 +31,24 @@ public class BillingS3HandlerInstaller {
 
   private final BillingS3ExportConfig config;
   private final MeterRegistry meterRegistry;
+  private final BiFunction<BatchedLogBuffer, AsyncBatchedLogUploader, BillingS3LogHandler>
+      handlerFactory;
 
   private volatile BillingS3LogHandler handler;
 
   @Inject
   public BillingS3HandlerInstaller(BillingS3ExportConfig config, MeterRegistry meterRegistry) {
+    this(config, meterRegistry, BillingS3LogHandler::new);
+  }
+
+  @VisibleForTesting
+  BillingS3HandlerInstaller(
+      BillingS3ExportConfig config,
+      MeterRegistry meterRegistry,
+      BiFunction<BatchedLogBuffer, AsyncBatchedLogUploader, BillingS3LogHandler> handlerFactory) {
     this.config = config;
     this.meterRegistry = meterRegistry;
+    this.handlerFactory = handlerFactory;
   }
 
   void onStart(@Observes StartupEvent event) {
@@ -62,14 +76,14 @@ 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);
     LOGGER.info(
         "Billing has attached BillingS3LogHandler to the logger named: {}", BILLING_LOGGER_NAME);
 
-    // TODO: XXXX call start on the thread.
+    Infrastructure.getDefaultWorkerPool().execute(this.handler::startUploading);
   }
 
   void onStop(@Observes ShutdownEvent event) {

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;

src/test/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3HandlerInstallerTest.java

@@ -1,11 +1,60 @@
 package io.stargate.sgv2.jsonapi.service.billing;
 
+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;
+import io.quarkus.runtime.ShutdownEvent;
+import io.quarkus.runtime.StartupEvent;
+import io.stargate.sgv2.jsonapi.config.BillingS3ExportConfig;
+import java.time.Duration;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicBoolean;
+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 startupCreatesAndStartsHandler() {
+    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 handler = mock(BillingS3LogHandler.class);
+    var handlerCreated = new AtomicBoolean();
+    var installer =
+        new BillingS3HandlerInstaller(
+            config,
+            new SimpleMeterRegistry(),
+            (buffer, uploader) -> {
+              assertThat(buffer).isNotNull();
+              assertThat(uploader).isNotNull();
+              handlerCreated.set(true);
+              return handler;
+            });
+
+    installer.onStart(new StartupEvent());
+    try {
+      assertThat(handlerCreated).isTrue();
+      verify(handler, timeout(10_000)).startUploading();
+    } finally {
+      installer.onStop(new ShutdownEvent());
+    }
+  }
+
   //
   //  private static BillingS3ExportConfig config(boolean enabled, String bucket, String region) {
   //    BillingS3ExportConfig config = mock(BillingS3ExportConfig.class);