[Logs 5] Use a separate ExecutorService for log batching (#4379)

* Add Log feature to Java SDK

* Rate limit for log items

* Add options for logs

* Add batch processor for logs

* Use a separate ExecutorService for log batching

Author: adinauer

sentry/src/main/java/io/sentry/logger/LoggerBatchProcessor.java

@@ -1,7 +1,9 @@
 package io.sentry.logger;
 
 import io.sentry.ISentryClient;
+import io.sentry.ISentryExecutorService;
 import io.sentry.ISentryLifecycleToken;
+import io.sentry.SentryExecutorService;
 import io.sentry.SentryLogEvent;
 import io.sentry.SentryLogEvents;
 import io.sentry.SentryOptions;
@@ -22,6 +24,7 @@ public final class LoggerBatchProcessor implements ILoggerBatchProcessor {
   private final @NotNull SentryOptions options;
   private final @NotNull ISentryClient client;
   private final @NotNull Queue<SentryLogEvent> queue;
+  private final @NotNull ISentryExecutorService executorService;
   private volatile @Nullable Future<?> scheduledFlush;
   private static final @NotNull AutoClosableReentrantLock scheduleLock =
       new AutoClosableReentrantLock();
@@ -31,6 +34,7 @@ public LoggerBatchProcessor(
     this.options = options;
     this.client = client;
     this.queue = new ConcurrentLinkedQueue<>();
+    this.executorService = new SentryExecutorService();
   }
 
   @Override
@@ -39,11 +43,14 @@ public void add(final @NotNull SentryLogEvent logEvent) {
     maybeSchedule(false, false);
   }
 
+  @SuppressWarnings("FutureReturnValueIgnored")
   @Override
   public void close(final boolean isRestarting) {
     if (isRestarting) {
       maybeSchedule(true, true);
+      executorService.submit(() -> executorService.close(options.getShutdownTimeoutMillis()));
     } else {
+      executorService.close(options.getShutdownTimeoutMillis());
       while (!queue.isEmpty()) {
         flushBatch();
       }
@@ -58,7 +65,7 @@ private void maybeSchedule(boolean forceSchedule, boolean immediately) {
           || latestScheduledFlush.isDone()
           || latestScheduledFlush.isCancelled()) {
         final int flushAfterMs = immediately ? 0 : FLUSH_AFTER_MS;
-        scheduledFlush = options.getExecutorService().schedule(new BatchRunnable(), flushAfterMs);
+        scheduledFlush = executorService.schedule(new BatchRunnable(), flushAfterMs);
       }
     }
   }