From fb1642320d444ed90a1450a1ab862513b2b6d7ba Mon Sep 17 00:00:00 2001 From: tallison Date: Tue, 27 Oct 2026 14:31:33 -0400 Subject: [PATCH] TIKA-5007: make JDBCEmitter thread-safe --- CHANGES.txt | 4 ++ .../tika/pipes/emitter/jdbc/JDBCEmitter.java | 10 +-- .../pipes/emitter/jdbc/JDBCEmitterTest.java | 62 +++++++++++++++++++ 3 files changed, 72 insertions(+), 4 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 9826420af33..a9cb193b8c3 100644 --- a/CHANGES.txt +++ b/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 diff --git a/tika-pipes/tika-pipes-plugins/tika-pipes-jdbc/src/main/java/org/apache/tika/pipes/emitter/jdbc/JDBCEmitter.java b/tika-pipes/tika-pipes-plugins/tika-pipes-jdbc/src/main/java/org/apache/tika/pipes/emitter/jdbc/JDBCEmitter.java index 9b4ab0f4117..1541755e04b 100644 --- a/tika-pipes/tika-pipes-plugins/tika-pipes-jdbc/src/main/java/org/apache/tika/pipes/emitter/jdbc/JDBCEmitter.java +++ b/tika-pipes/tika-pipes-plugins/tika-pipes-jdbc/src/main/java/org/apache/tika/pipes/emitter/jdbc/JDBCEmitter.java @@ -75,8 +75,8 @@ * *

* 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. *

*/ public class JDBCEmitter extends AbstractEmitter implements Closeable { @@ -185,7 +185,9 @@ public void emit(List emitData) throws IOException { } } - private void emitWithRetry(String emitKey, List metadataList) throws IOException { + // one lock for bind, batch, commit and reconnect: they share the connection and statement + private synchronized void emitWithRetry(String emitKey, List 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(); diff --git a/tika-pipes/tika-pipes-plugins/tika-pipes-jdbc/src/test/java/org/apache/tika/pipes/emitter/jdbc/JDBCEmitterTest.java b/tika-pipes/tika-pipes-plugins/tika-pipes-jdbc/src/test/java/org/apache/tika/pipes/emitter/jdbc/JDBCEmitterTest.java index 7dfcfaa452a..1c9760f3828 100644 --- a/tika-pipes/tika-pipes-plugins/tika-pipes-jdbc/src/test/java/org/apache/tika/pipes/emitter/jdbc/JDBCEmitterTest.java +++ b/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 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> 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++) {