This is an automated email from the ASF dual-hosted git repository.
tballison pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/tika.git
The following commit(s) were added to refs/heads/main by this push:
new 235cb9be53 TIKA-5007: make JDBCEmitter thread-safe (#3444)
235cb9be53 is described below
commit 235cb9be532cd61093fbb862da2f10713c9e6b99
Author: Tim Allison <[email protected]>
AuthorDate: Wed Oct 28 09:03:46 2026 -0400
TIKA-5007: make JDBCEmitter thread-safe (#3444)
---
CHANGES.txt | 4 ++
.../tika/pipes/emitter/jdbc/JDBCEmitter.java | 10 ++--
.../tika/pipes/emitter/jdbc/JDBCEmitterTest.java | 62 ++++++++++++++++++++++
3 files changed, 72 insertions(+), 4 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 6b8df9399d..7969560e69 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -22,6 +22,10 @@ Release 4.3.0 - ???
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 9b4ab0f411..1541755e04 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 @@ import org.apache.tika.utils.TikaDates;
* </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 class JDBCEmitter extends AbstractEmitter implements
Closeable {
}
}
- 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 @@ public class JDBCEmitter extends AbstractEmitter implements
Closeable {
}
@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 7dfcfaa452..1c9760f382 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.List;
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 class JDBCEmitterTest {
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++) {
