Merge branch 'tests-fix-print-pool-threads' into 'dev'

BoxStore: improve thread pool shutdown, error messages #258

See merge request objectbox/objectbox-java!147

Author: greenrobot-team

CHANGELOG.md

@@ -10,6 +10,8 @@ For more insights into what changed in the ObjectBox C++ core, [check the Object
   an "Another BoxStore is still open for this directory" exception. Note that calling `close()` *is recommended* before 
   creating a new instance. [#1201](https://github.com/objectbox/objectbox-java/issues/1201)
 - When using `BoxStoreBuilder.buildDefault()`, don't leak Store when setting as default fails.
+- To help diagnose, print stacks of all threads in the internal thread pool if shutting it down takes too long when
+  closing `BoxStore`.
 
 ## 4.3.1 - 2025-08-12
 

objectbox-java/src/main/java/io/objectbox/BoxStore.java

@@ -21,6 +21,7 @@
 import java.io.Closeable;
 import java.io.File;
 import java.io.IOException;
+import java.io.PrintStream;
 import java.net.MalformedURLException;
 import java.net.URL;
 import java.util.ArrayList;
@@ -81,6 +82,8 @@ public class BoxStore implements Closeable {
 
     /** The ObjectBox database version this Java library is known to work with. */
     private static final String VERSION = "4.3.1-2025-08-02";
+
+    private static final String OBJECTBOX_PACKAGE_NAME = "objectbox";
     private static BoxStore defaultStore;
 
     /** Currently used DB dirs with values from {@link #getCanonicalPath(File)}. */
@@ -238,6 +241,7 @@ public static boolean isSyncServerAvailable() {
 
     native long nativePanicModeRemoveAllObjects(long store, int entityId);
 
+    private final PrintStream errorOutputStream;
     private final File directory;
     private final String canonicalPath;
     /** Reference to the native store. Should probably get through {@link #getNativeStore()} instead. */
@@ -283,6 +287,7 @@ public static boolean isSyncServerAvailable() {
         relinker = builder.relinker;
         NativeLibraryLoader.ensureLoaded();
 
+        errorOutputStream = builder.errorOutputStream;
         directory = builder.directory;
         canonicalPath = getCanonicalPath(directory);
         verifyNotAlreadyOpen(canonicalPath);
@@ -613,7 +618,7 @@ public Transaction beginTx() {
         // Because write TXs are typically not cached, initialCommitCount is not as relevant than for read TXs.
         int initialCommitCount = commitCount;
         if (debugTxWrite) {
-            System.out.println("Begin TX with commit count " + initialCommitCount);
+            getOutput().println("Begin TX with commit count " + initialCommitCount);
         }
         long nativeTx = nativeBeginTx(getNativeStore());
         if (nativeTx == 0) throw new DbException("Could not create native transaction");
@@ -638,7 +643,7 @@ public Transaction beginReadTx() {
         // TODO add multithreaded test for this
         int initialCommitCount = commitCount;
         if (debugTxRead) {
-            System.out.println("Begin read TX with commit count " + initialCommitCount);
+            getOutput().println("Begin read TX with commit count " + initialCommitCount);
         }
         long nativeTx = nativeBeginReadTx(getNativeStore());
         if (nativeTx == 0) throw new DbException("Could not create native read transaction");
@@ -693,12 +698,17 @@ public void close() {
                 // (due to all Java APIs doing closed checks).
                 closed = true;
 
+                // Stop accepting new tasks (async calls, query publishers) on the internal thread pool
+                internalThreadPool().shutdown();
+                // Give running tasks some time to finish, print warnings if they do not to help callers fix their code
+                checkThreadTermination();
+
                 List<Transaction> transactionsToClose;
                 synchronized (transactions) {
                     // Give open transactions some time to close (BoxStore.unregisterTransaction() calls notify),
                     // 1000 ms should be long enough for most small operations and short enough to avoid ANRs on Android.
                     if (hasActiveTransaction()) {
-                        System.out.println("Briefly waiting for active transactions before closing the Store...");
+                        getOutput().println("Briefly waiting for active transactions before closing the Store...");
                         try {
                             // It is fine to hold a lock on BoxStore.this as well as BoxStore.unregisterTransaction()
                             // only synchronizes on "transactions".
@@ -708,7 +718,7 @@ public void close() {
                             // If interrupted, continue with releasing native resources
                         }
                         if (hasActiveTransaction()) {
-                            System.err.println("Transactions are still active:"
+                            getErrorOutput().println("Transactions are still active:"
                                     + " ensure that all database operations are finished before closing the Store!");
                         }
                     }
@@ -726,10 +736,6 @@ public void close() {
                 if (handleToDelete != 0) { // failed before native handle was created?
                     nativeDelete(handleToDelete);
                 }
-
-                // When running the full unit test suite, we had 100+ threads before, hope this helps:
-                threadPool.shutdown();
-                checkThreadTermination();
             }
         }
         if (!oldClosedState) {
@@ -740,24 +746,52 @@ public void close() {
         }
     }
 
-    /** dump thread stacks if pool does not terminate promptly. */
+    /**
+     * Waits briefly for the internal {@link #internalThreadPool()} to terminate. If it does not terminate in time,
+     * prints stack traces of the pool threads.
+     */
     private void checkThreadTermination() {
         try {
-            if (!threadPool.awaitTermination(1, TimeUnit.SECONDS)) {
-                int activeCount = Thread.activeCount();
-                System.err.println("Thread pool not terminated in time; printing stack traces...");
-                Thread[] threads = new Thread[activeCount + 2];
+            if (!internalThreadPool().awaitTermination(1, TimeUnit.SECONDS)) {
+                getErrorOutput().println("ObjectBox thread pool not terminated in time." +
+                        " Ensure all async calls have completed and subscriptions are cancelled before closing the Store." +
+                        "\nDumping stack traces of threads on the pool and any using ObjectBox APIs:" +
+                        "\n=== BEGIN OF DUMP ===");
+                // Note: this may not print any pool threads if other threads are started while enumerating
+                // (and the pool threads do not make it into the threads array).
+                Thread[] threads = new Thread[Thread.activeCount()];
                 int count = Thread.enumerate(threads);
                 for (int i = 0; i < count; i++) {
-                    System.err.println("Thread: " + threads[i].getName());
-                    Thread.dumpStack();
+                    Thread thread = threads[i];
+                    if (shouldDumpThreadStackTrace(thread)) {
+                        getErrorOutput().println("Thread: " + thread.getName());
+                        StackTraceElement[] trace = thread.getStackTrace();
+                        for (StackTraceElement traceElement : trace) {
+                            getErrorOutput().println("\tat " + traceElement);
+                        }
+                    }
                 }
+                getErrorOutput().println("=== END OF DUMP ===");
             }
         } catch (InterruptedException e) {
-            e.printStackTrace();
+            e.printStackTrace(getErrorOutput());
         }
     }
 
+    private boolean shouldDumpThreadStackTrace(Thread thread) {
+        // Dump any threads of the internal thread pool
+        if (thread.getName().startsWith(ObjectBoxThreadPool.THREAD_NAME_PREFIX)) return true;
+
+        // Any other thread might be blocking a thread on the internal pool, so also dump any that appear to use
+        // ObjectBox APIs.
+        StackTraceElement[] trace = thread.getStackTrace();
+        for (StackTraceElement traceElement : trace) {
+            if (traceElement.getClassName().contains(OBJECTBOX_PACKAGE_NAME)) return true;
+        }
+
+        return false;
+    }
+
     /**
      * Danger zone! This will delete all data (files) of this BoxStore!
      * You must call {@link #close()} before and read the docs of that method carefully!
@@ -894,7 +928,7 @@ void txCommitted(Transaction tx, @Nullable int[] entityTypeIdsAffected) {
         synchronized (txCommitCountLock) {
             commitCount++; // Overflow is OK because we check for equality
             if (debugTxWrite) {
-                System.out.println("TX committed. New commit count: " + commitCount + ", entity types affected: " +
+                getOutput().println("TX committed. New commit count: " + commitCount + ", entity types affected: " +
                         (entityTypeIdsAffected != null ? entityTypeIdsAffected.length : 0));
             }
         }
@@ -1013,10 +1047,10 @@ public <T> T callInReadTxWithRetry(Callable<T> callable, int attempts, int initi
                 String diagnose = diagnose();
                 String message = attempt + " of " + attempts + " attempts of calling a read TX failed:";
                 if (logAndHeal) {
-                    System.err.println(message);
+                    getErrorOutput().println(message);
                     e.printStackTrace();
-                    System.err.println(diagnose);
-                    System.err.flush();
+                    getErrorOutput().println(diagnose);
+                    getErrorOutput().flush();
 
                     System.gc();
                     System.runFinalization();
@@ -1121,7 +1155,7 @@ public <R> R callInTxNoException(Callable<R> callable) {
      * See also {@link #runInTx(Runnable)}.
      */
     public void runInTxAsync(final Runnable runnable, @Nullable final TxCallback<Void> callback) {
-        threadPool.submit(() -> {
+        internalScheduleThread(() -> {
             try {
                 runInTx(runnable);
                 if (callback != null) {
@@ -1142,7 +1176,7 @@ public void runInTxAsync(final Runnable runnable, @Nullable final TxCallback<Voi
      * * See also {@link #callInTx(Callable)}.
      */
     public <R> void callInTxAsync(final Callable<R> callable, @Nullable final TxCallback<R> callback) {
-        threadPool.submit(() -> {
+        internalScheduleThread(() -> {
             try {
                 R result = callInTx(callable);
                 if (callback != null) {
@@ -1314,11 +1348,11 @@ public void setDbExceptionListener(@Nullable DbExceptionListener dbExceptionList
 
     @Internal
     public Future<?> internalScheduleThread(Runnable runnable) {
-        return threadPool.submit(runnable);
+        return internalThreadPool().submit(runnable);
     }
 
     @Internal
-    public ExecutorService internalThreadPool() {
+    ExecutorService internalThreadPool() {
         return threadPool;
     }
 
@@ -1337,6 +1371,20 @@ public TxCallback<?> internalFailedReadTxAttemptCallback() {
         return failedReadTxAttemptCallback;
     }
 
+    /**
+     * The output stream to print log messages to. Currently {@link System#out}.
+     */
+    private PrintStream getOutput() {
+        return System.out;
+    }
+
+    /**
+     * The error output stream to print log messages to. This is {@link System#err} by default.
+     */
+    private PrintStream getErrorOutput() {
+        return errorOutputStream;
+    }
+
     void setDebugFlags(int debugFlags) {
         nativeSetDebugFlags(getNativeStore(), debugFlags);
     }

objectbox-java/src/main/java/io/objectbox/BoxStoreBuilder.java

@@ -25,6 +25,7 @@
 import java.io.FileOutputStream;
 import java.io.InputStream;
 import java.io.OutputStream;
+import java.io.PrintStream;
 import java.lang.reflect.Method;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -67,6 +68,12 @@ public class BoxStoreBuilder {
     /** The default maximum size the DB can grow to, which can be overwritten using {@link #maxSizeInKByte}. */
     public static final int DEFAULT_MAX_DB_SIZE_KBYTE = 1024 * 1024;
 
+    /**
+     * The error output stream {@link BoxStore} uses for logging. Defaults to {@link System#err}, but can be customized
+     * for tests.
+     */
+    PrintStream errorOutputStream = System.err;
+
     final byte[] model;
 
     /** BoxStore uses this (not baseDirectory/name) */
@@ -145,6 +152,15 @@ public BoxStoreBuilder(byte[] model) {
         this.model = Arrays.copyOf(model, model.length);
     }
 
+    /**
+     * For testing: set a custom error output stream {@link BoxStore} uses for logging. Defaults to {@link System#err}.
+     */
+    @Internal
+    BoxStoreBuilder setErrorOutput(PrintStream err) {
+        errorOutputStream = err;
+        return this;
+    }
+
     /**
      * Name of the database, which will be used as a directory for database files.
      * You can also specify a base directory for this one using {@link #baseDirectory(File)}.

objectbox-java/src/main/java/io/objectbox/internal/ObjectBoxThreadPool.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 ObjectBox Ltd.
+ * Copyright 2017-2024 ObjectBox Ltd.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -36,6 +36,8 @@
  */
 @Internal
 public class ObjectBoxThreadPool extends ThreadPoolExecutor {
+
+    public static String THREAD_NAME_PREFIX = "ObjectBox-";
     private final BoxStore boxStore;
 
     public ObjectBoxThreadPool(BoxStore boxStore) {
@@ -54,7 +56,7 @@ static class ObjectBoxThreadFactory implements ThreadFactory {
         private static final AtomicInteger POOL_COUNT = new AtomicInteger();
 
         private final ThreadGroup group;
-        private final String namePrefix = "ObjectBox-" + POOL_COUNT.incrementAndGet() + "-Thread-";
+        private final String namePrefix = THREAD_NAME_PREFIX + POOL_COUNT.incrementAndGet() + "-Thread-";
         private final AtomicInteger threadCount = new AtomicInteger();
 
         ObjectBoxThreadFactory() {

tests/objectbox-java-test/src/test/java/io/objectbox/AbstractObjectBoxTest.java

@@ -162,6 +162,10 @@ protected Box<TestEntity> getTestEntityBox() {
 
     @After
     public void tearDown() {
+        // Note: do not collect dangling Cursors and Transactions before store closes (using System.gc()
+        // or System.runFinalization()). Tests should mirror user code and do that themselves (calling close())
+        // or rely on the library (through finalizers or BoxStore.close()).
+
         if (store != null) {
             try {
                 store.close();