diff --git a/jena-db/jena-tdb2/src/main/java/org/apache/jena/tdb2/loader/main/DataToTuples.java b/jena-db/jena-tdb2/src/main/java/org/apache/jena/tdb2/loader/main/DataToTuples.java
index a0886a9..24b30bb 100644
--- a/jena-db/jena-tdb2/src/main/java/org/apache/jena/tdb2/loader/main/DataToTuples.java
+++ b/jena-db/jena-tdb2/src/main/java/org/apache/jena/tdb2/loader/main/DataToTuples.java
@@ -18,10 +18,13 @@
 
 package org.apache.jena.tdb2.loader.main;
 
+import static org.apache.jena.tdb2.loader.main.PhasedOps.acquire;
+
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.Semaphore;
 import java.util.function.Consumer;
 import java.util.function.Function;
 
@@ -101,6 +104,9 @@
         }
     }
 
+    /** Semaphore for the other thread to indicate it has finished. */
+    private final Semaphore termination = new Semaphore(0);
+
     @Override
     public void startBulk() {
         thread = new Thread(()->action());
@@ -109,11 +115,7 @@
 
     @Override
     public void finishBulk() {
-        try {
-            thread.join();
-        } catch (InterruptedException e) {
-            throw new BulkLoaderException("InterruptedException", e);
-        }
+        acquire(termination);
     }
 
     // Triples.
@@ -155,6 +157,7 @@
         }
         transaction.end();
         CoLib.finish(coordinator);
+        termination.release();
     }
 
     //@Override
diff --git a/jena-db/jena-tdb2/src/main/java/org/apache/jena/tdb2/loader/main/Indexer.java b/jena-db/jena-tdb2/src/main/java/org/apache/jena/tdb2/loader/main/Indexer.java
index 02cd56f..54aef8f 100644
--- a/jena-db/jena-tdb2/src/main/java/org/apache/jena/tdb2/loader/main/Indexer.java
+++ b/jena-db/jena-tdb2/src/main/java/org/apache/jena/tdb2/loader/main/Indexer.java
@@ -18,6 +18,8 @@
 
 package org.apache.jena.tdb2.loader.main;
 
+import static org.apache.jena.tdb2.loader.main.PhasedOps.acquire;
+
 import java.util.Arrays;
 import java.util.List;
 import java.util.concurrent.ArrayBlockingQueue;
@@ -25,7 +27,6 @@
 import java.util.concurrent.Semaphore;
 
 import org.apache.jena.atlas.lib.ArrayUtils;
-import org.apache.jena.atlas.lib.Timer;
 import org.apache.jena.atlas.lib.tuple.Tuple;
 import org.apache.jena.atlas.logging.Log;
 import org.apache.jena.dboe.transaction.txn.Transaction;
@@ -65,16 +66,6 @@
         }
     }
 
-    private static long acquire(Semaphore semaphore, int numPermits) {
-        return Timer.time(()->{
-            try { semaphore.acquire(numPermits); }
-            catch (InterruptedException e) {
-                Log.error(Indexer.class, "Interrupted", e);
-                throw new RuntimeException(e);
-            }
-        });
-    }
-
     /** Return a function that delivers multiple {@code List<Tuple<NodeId>>>} to this indexer */
     public Destination<Tuple<NodeId>> index() {
         return this::index;
diff --git a/jena-db/jena-tdb2/src/main/java/org/apache/jena/tdb2/loader/main/PhasedOps.java b/jena-db/jena-tdb2/src/main/java/org/apache/jena/tdb2/loader/main/PhasedOps.java
index 7ee8d3e..852127e 100644
--- a/jena-db/jena-tdb2/src/main/java/org/apache/jena/tdb2/loader/main/PhasedOps.java
+++ b/jena-db/jena-tdb2/src/main/java/org/apache/jena/tdb2/loader/main/PhasedOps.java
@@ -19,11 +19,14 @@
 package org.apache.jena.tdb2.loader.main;
 
 import java.util.*;
+import java.util.concurrent.Semaphore;
 import java.util.function.Consumer;
 import java.util.function.IntFunction;
 import java.util.stream.Stream;
 
+import org.apache.jena.atlas.lib.Timer;
 import org.apache.jena.atlas.lib.tuple.Tuple;
+import org.apache.jena.atlas.logging.Log;
 import org.apache.jena.tdb2.loader.BulkLoaderException;
 import org.apache.jena.tdb2.loader.base.MonitorOutput;
 import org.apache.jena.tdb2.loader.base.ProgressMonitor;
@@ -37,6 +40,22 @@
  */
 class PhasedOps {
 
+    /** Acquire one permit from a semaphore. Return the time spent waiting. */
+    /* package */ static long acquire(Semaphore termination) {
+        return acquire(termination, 1);
+    }
+
+    /** Acquire permits from a semaphore. Return the time spent waiting. */
+    /* package */ static long acquire(Semaphore semaphore, int numPermits) {
+        return Timer.time(()->{
+            try { semaphore.acquire(numPermits); }
+            catch (InterruptedException e) {
+                Log.error(Indexer.class, "Interrupted", e);
+                throw new RuntimeException(e);
+            }
+        });
+    }
+
     static Map<String, TupleIndex> indexMap(DatasetGraphTDB dsgtdb) {
         Map<String, TupleIndex> indexMap = new HashMap<>();
         // All triple/quad indexes.