perf(core): [Logs and Metrics Enable Flags 10] Avoid unused Metrics worker thread

sentry-android-core/src/test/java/io/sentry/android/core/AndroidMetricsBatchProcessorTest.kt

@@ -1,11 +1,14 @@
 package io.sentry.android.core
 
 import androidx.test.ext.junit.runners.AndroidJUnit4
+import com.google.common.truth.Truth.assertThat
 import io.sentry.ISentryClient
 import io.sentry.SentryMetricsEvent
 import io.sentry.SentryOptions
 import io.sentry.protocol.SentryId
 import io.sentry.test.ImmediateExecutorService
+import io.sentry.test.getProperty
+import java.util.concurrent.atomic.AtomicBoolean
 import kotlin.test.AfterTest
 import kotlin.test.BeforeTest
 import kotlin.test.Test
@@ -14,6 +17,7 @@ import kotlin.test.assertTrue
 import org.junit.runner.RunWith
 import org.mockito.kotlin.any
 import org.mockito.kotlin.mock
+import org.mockito.kotlin.never
 import org.mockito.kotlin.verify
 import org.mockito.kotlin.whenever
 
@@ -54,6 +58,16 @@ class AndroidMetricsBatchProcessorTest {
     assertNotNull(AppState.getInstance().lifecycleObserver)
   }
 
+  @Test
+  fun `onBackground does not flush before first accepted item`() {
+    val sut = fixture.getSut(useImmediateExecutor = true)
+
+    sut.onBackground()
+
+    assertThat(sut.getProperty<AtomicBoolean>("hasScheduled").get()).isFalse()
+    verify(fixture.client, never()).captureBatchedMetricsEvents(any())
+  }
+
   @Test
   fun `onBackground schedules flush`() {
     val sut = fixture.getSut(useImmediateExecutor = true)

sentry/src/main/java/io/sentry/metrics/MetricsBatchProcessor.java

@@ -34,16 +34,24 @@ public class MetricsBatchProcessor implements IMetricsBatchProcessor {
   private final @NotNull Queue<SentryMetricsEvent> queue;
   private final @NotNull ISentryExecutorService executorService;
   private final @NotNull AtomicBoolean hasScheduled = new AtomicBoolean(false);
+  private volatile boolean hasAcceptedItem = false;
   private volatile boolean isShuttingDown = false;
 
   private final @NotNull ReusableCountLatch pendingCount = new ReusableCountLatch();
 
   public MetricsBatchProcessor(
       final @NotNull SentryOptions options, final @NotNull ISentryClient client) {
+    this(options, client, new SentryExecutorService(options));
+  }
+
+  MetricsBatchProcessor(
+      final @NotNull SentryOptions options,
+      final @NotNull ISentryClient client,
+      final @NotNull ISentryExecutorService executorService) {
     this.options = options;
     this.client = client;
     this.queue = new ConcurrentLinkedQueue<>();
-    this.executorService = new SentryExecutorService(options);
+    this.executorService = executorService;
   }
 
   @Override
@@ -65,21 +73,22 @@ public void add(final @NotNull SentryMetricsEvent metricsEvent) {
     }
     pendingCount.increment();
     queue.offer(metricsEvent);
+    hasAcceptedItem = true;
     maybeSchedule(false);
   }
 
   @SuppressWarnings("FutureReturnValueIgnored")
   @Override
   public void close(final boolean isRestarting) {
     isShuttingDown = true;
-    if (isRestarting) {
+    if (isRestarting && hasAcceptedItem) {
       maybeSchedule(true);
       executorService.submit(() -> executorService.close(options.getShutdownTimeoutMillis()));
-    } else {
-      executorService.close(options.getShutdownTimeoutMillis());
-      while (!queue.isEmpty()) {
-        flushBatch();
-      }
+      return;
+    }
+    executorService.close(options.getShutdownTimeoutMillis());
+    while (!queue.isEmpty()) {
+      flushBatch();
     }
   }
 
@@ -106,6 +115,9 @@ private void maybeSchedule(boolean immediately) {
 
   @Override
   public void flush(long timeoutMillis) {
+    if (!hasAcceptedItem) {
+      return;
+    }
     maybeSchedule(true);
     try {
       pendingCount.waitTillZero(timeoutMillis, TimeUnit.MILLISECONDS);

sentry/src/test/java/io/sentry/metrics/MetricsBatchProcessorTest.kt

@@ -3,6 +3,7 @@ package io.sentry.metrics
 import com.google.common.truth.Truth.assertThat
 import io.sentry.DataCategory
 import io.sentry.ISentryClient
+import io.sentry.ISentryExecutorService
 import io.sentry.SentryMetricsEvent
 import io.sentry.SentryMetricsEvents
 import io.sentry.SentryNanotimeDate
@@ -12,19 +13,106 @@ import io.sentry.clientreport.DiscardReason
 import io.sentry.clientreport.DiscardedEvent
 import io.sentry.protocol.SentryId
 import io.sentry.test.DeferredExecutorService
+import io.sentry.test.getProperty
 import io.sentry.test.injectForField
+import io.sentry.transport.ReusableCountLatch
 import io.sentry.util.JsonSerializationUtils
 import kotlin.test.Test
 import kotlin.test.assertEquals
 import kotlin.test.assertFalse
 import kotlin.test.assertTrue
+import org.mockito.kotlin.any
 import org.mockito.kotlin.argumentCaptor
 import org.mockito.kotlin.atLeast
 import org.mockito.kotlin.mock
+import org.mockito.kotlin.never
 import org.mockito.kotlin.times
 import org.mockito.kotlin.verify
+import org.mockito.kotlin.verifyNoInteractions
 
 class MetricsBatchProcessorTest {
+  @Test
+  fun `constructor does not submit processor work`() {
+    val mockExecutor = mock<ISentryExecutorService>()
+
+    MetricsBatchProcessor(SentryOptions(), mock(), mockExecutor)
+
+    verifyNoInteractions(mockExecutor)
+  }
+
+  @Test
+  fun `empty flush does not submit processor work`() {
+    val mockExecutor = mock<ISentryExecutorService>()
+    val processor = MetricsBatchProcessor(SentryOptions(), mock(), mockExecutor)
+
+    processor.flush(0)
+
+    verifyNoInteractions(mockExecutor)
+  }
+
+  @Test
+  fun `close before first accepted item does not submit processor work`() {
+    val mockExecutor = mock<ISentryExecutorService>()
+    val processor = MetricsBatchProcessor(SentryOptions(), mock(), mockExecutor)
+
+    processor.close(false)
+
+    verify(mockExecutor).close(any())
+    verify(mockExecutor, never()).schedule(any(), any())
+    verify(mockExecutor, never()).submit(any<Runnable>())
+  }
+
+  @Test
+  fun `restart close before first accepted item does not submit processor work`() {
+    val mockExecutor = mock<ISentryExecutorService>()
+    val processor = MetricsBatchProcessor(SentryOptions(), mock(), mockExecutor)
+
+    processor.close(true)
+
+    verify(mockExecutor).close(any())
+    verify(mockExecutor, never()).schedule(any(), any())
+    verify(mockExecutor, never()).submit(any<Runnable>())
+  }
+
+  @Test
+  fun `item rejected during shutdown does not mark processor as used`() {
+    val mockExecutor = mock<ISentryExecutorService>()
+    val processor = MetricsBatchProcessor(SentryOptions(), mock(), mockExecutor)
+    processor.close(false)
+
+    processor.add(metricsEvent("rejected"))
+    processor.flush(0)
+
+    verify(mockExecutor, never()).schedule(any(), any())
+    verify(mockExecutor, never()).submit(any<Runnable>())
+  }
+
+  @Test
+  fun `item rejected due to queue capacity does not mark processor as used`() {
+    val mockExecutor = mock<ISentryExecutorService>()
+    val processor = MetricsBatchProcessor(SentryOptions(), mock(), mockExecutor)
+    val pendingCount = processor.getProperty<ReusableCountLatch>("pendingCount")
+    repeat(MetricsBatchProcessor.MAX_QUEUE_SIZE) { pendingCount.increment() }
+
+    processor.add(metricsEvent("rejected"))
+    processor.flush(0)
+
+    verifyNoInteractions(mockExecutor)
+  }
+
+  @Test
+  fun `flush and restart close submit processor work after first accepted item`() {
+    val mockExecutor = mock<ISentryExecutorService>()
+    val processor = MetricsBatchProcessor(SentryOptions(), mock(), mockExecutor)
+    processor.add(metricsEvent("accepted"))
+
+    processor.flush(0)
+    processor.close(true)
+
+    verify(mockExecutor, times(3)).schedule(any(), any())
+    verify(mockExecutor).submit(any<Runnable>())
+  }
+
   @Test
   fun `schedules another flush after previous flush has run`() {
     val mockClient = mock<ISentryClient>()
@@ -46,6 +134,9 @@ class MetricsBatchProcessorTest {
       .inOrder()
   }
 
+  private fun metricsEvent(name: String) =
+    SentryMetricsEvent(SentryId(), SentryNanotimeDate(), name, "gauge", 1.0)
+
   @Test
   fun `drops metrics events after reaching MAX_QUEUE_SIZE limit`() {
     // given