TIKA-5007: make JDBCEmitter thread-safe

CHANGES.txt

@@ -6,6 +6,10 @@ Release 4.2.1 - unreleased
      for EKS IAM roles for service accounts (IRSA); the plugin now bundles the AWS STS
      module it needs (TIKA-5006).
 
+   * The JDBC emitter is thread-safe: concurrent emits (shared-server mode, several
+     async emitter threads) shared one connection and prepared statement unguarded,
+     mixing values across rows and committing each other's batches (TIKA-5007).
+
   OTHER CHANGES
 
    * PEM-armored CMS/PKCS7 (-----BEGIN PKCS7----- or -----BEGIN CMS-----) is parsed like

tika-pipes/tika-pipes-plugins/tika-pipes-jdbc/src/main/java/org/apache/tika/pipes/emitter/jdbc/JDBCEmitter.java

@@ -75,8 +75,8 @@
  * </pre>
  * <p>
  * This is only an initial, basic implementation of an emitter for JDBC.
- * It is currently NOT thread safe because of the shared prepared statement,
- * and depending on the jdbc implementation because of the shared connection.
+ * It is thread safe: emits share one connection and prepared statement, so they are
+ * serialized, one transaction at a time.
  * </p>
  */
 public class JDBCEmitter extends AbstractEmitter implements Closeable {
@@ -185,7 +185,9 @@ public void emit(List<? extends EmitData> emitData) throws IOException {
         }
     }
 
-    private void emitWithRetry(String emitKey, List<Metadata> metadataList) throws IOException {
+    // one lock for bind, batch, commit and reconnect: they share the connection and statement
+    private synchronized void emitWithRetry(String emitKey, List<Metadata> metadataList)
+            throws IOException {
         int tries = 0;
         Exception ex = null;
         while (tries++ <= config.maxRetries()) {
@@ -409,7 +411,7 @@ private void updateBoolean(PreparedStatement insertStatement, int i, String val)
     }
 
     @Override
-    public void close() throws IOException {
+    public synchronized void close() throws IOException {
         try {
             if (insertStatement != null) {
                 insertStatement.close();

tika-pipes/tika-pipes-plugins/tika-pipes-jdbc/src/test/java/org/apache/tika/pipes/emitter/jdbc/JDBCEmitterTest.java

@@ -37,6 +37,11 @@
 import java.util.Locale;
 import java.util.Map;
 import java.util.TimeZone;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.node.ObjectNode;
@@ -397,6 +402,63 @@ public void testVarcharTruncation(@TempDir Path tmpDir) throws Exception {
         assertEquals(3, rows);
     }
 
+    /**
+     * Shared-server mode hands one emitter to every thread (TIKA-5007): concurrent emits must
+     * not mix one document's values into another's row or commit each other's batches.
+     */
+    @Test
+    public void testConcurrentEmits(@TempDir Path tmpDir) throws Exception {
+        String connectionString = "jdbc:h2:file:" + tmpDir.resolve("h2").toAbsolutePath();
+        LinkedHashMap<String, String> keys = new LinkedHashMap<>();
+        keys.put("k1", "string");
+        keys.put("k2", "int");
+        ObjectNode configNode = createConfigNode(connectionString,
+                "insert into test (path, k1, k2) values (?,?,?)",
+                "create table test (path varchar(64) primary key, k1 varchar(64), k2 integer)",
+                null, "first_only", keys);
+        JDBCEmitter emitter = JDBCEmitter.build(new ExtensionConfig("test-jdbc", "jdbc-emitter",
+                MAPPER.writeValueAsString(configNode)));
+
+        int threads = 8;
+        int perThread = 200;
+        ExecutorService executor = Executors.newFixedThreadPool(threads);
+        CountDownLatch start = new CountDownLatch(1);
+        List<Future<?>> futures = new ArrayList<>();
+        try {
+            for (int t = 0; t < threads; t++) {
+                int base = t * perThread;
+                futures.add(executor.submit(() -> {
+                    start.await();
+                    for (int i = base; i < base + perThread; i++) {
+                        emitter.emit("id" + i, List.of(m("k1", "v" + i, "k2", Integer.toString(i))),
+                                new ParseContext());
+                    }
+                    return null;
+                }));
+            }
+            start.countDown();
+            for (Future<?> f : futures) {
+                f.get(2, TimeUnit.MINUTES);
+            }
+        } finally {
+            executor.shutdownNow();
+            emitter.close();
+        }
+
+        try (Connection connection = DriverManager.getConnection(connectionString);
+                Statement st = connection.createStatement();
+                ResultSet rs = st.executeQuery("select path, k1, k2 from test")) {
+            int rows = 0;
+            while (rs.next()) {
+                int k2 = rs.getInt(3);
+                assertEquals("id" + k2, rs.getString(1));
+                assertEquals("v" + k2, rs.getString(2));
+                rows++;
+            }
+            assertEquals(threads * perThread, rows);
+        }
+    }
+
     private Metadata m(String... strings) {
         Metadata metadata = new Metadata();
         for (int i = 0; i < strings.length; i++) {