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 4d74019db5 TIKA-4889: engines are closed with the config that loaded
them (#3169)
4d74019db5 is described below
commit 4d74019db593ddeb867b418b8b699311a991c284
Author: Tim Allison <[email protected]>
AuthorDate: Mon Sep 14 20:23:03 2026 -0400
TIKA-4889: engines are closed with the config that loaded them (#3169)
---
CHANGES.txt | 6 ++
.../ROOT/pages/configuration/inference.adoc | 5 ++
.../org/apache/tika/parser/inference/Engine.java | 13 ++++-
.../tika/parser/inference/EngineRegistry.java | 33 ++++++++++-
.../tika/parser/inference/EngineRegistryTest.java | 64 ++++++++++++++++++++++
.../tika/inference/OpenAIEmbeddingEngine.java | 3 +-
.../apache/tika/pipes/core/server/PipesServer.java | 7 +++
.../pipes/core/server/SharedServerResources.java | 23 +++++++-
.../tika/config/loader/EngineLoaderTest.java | 5 ++
.../org/apache/tika/config/loader/TestEngine.java | 10 ++++
10 files changed, 160 insertions(+), 9 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index b14be90515..90d23060af 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,5 +1,11 @@
Release 4.1.0 - unreleased
+ * Engines are closed with the config that loaded them: Engine is
+ Closeable (no-op by default), EngineRegistry closes every engine once,
+ and the pipes server closes the registry at shutdown, so an
+ openai-embedding-engine or a VLM used as a text recognizer releases its
+ HTTP client (TIKA-4889).
+
* Per-request switch for text recognizers: {"parse-context":
{"text-recognizers": {"enabled": false}}} runs no recognizer or annotator
from the list for that parse, in every container, and OCR_ONLY no longer
diff --git a/docs/modules/ROOT/pages/configuration/inference.adoc
b/docs/modules/ROOT/pages/configuration/inference.adoc
index fd2a982ee8..cb2f65cdc1 100644
--- a/docs/modules/ROOT/pages/configuration/inference.adoc
+++ b/docs/modules/ROOT/pages/configuration/inference.adoc
@@ -49,6 +49,11 @@
xref:configuration/index.adoc#text-recognizers[`text-recognizers`], so an OCR en
configured in one place. A binding cannot use a recognizer for a task it does
not fit; `embed`
on Tesseract fails config load.
+An engine lives as long as the config that loaded it. `Engine` is `Closeable`;
an engine that
+holds a client or a native handle releases it in `close()`, and the registry
closes every
+engine once when the pipes server shuts down. Library users holding a
`TikaLoader` close
+`loader.get(EngineRegistry.class)` themselves.
+
== `inference`: what runs on what
A list of bindings. Each names an engine, says what it is fed, and what to ask:
diff --git
a/tika-core/src/main/java/org/apache/tika/parser/inference/Engine.java
b/tika-core/src/main/java/org/apache/tika/parser/inference/Engine.java
index ba71a2d0b9..cca39ece09 100644
--- a/tika-core/src/main/java/org/apache/tika/parser/inference/Engine.java
+++ b/tika-core/src/main/java/org/apache/tika/parser/inference/Engine.java
@@ -16,12 +16,21 @@
*/
package org.apache.tika.parser.inference;
+import java.io.Closeable;
+import java.io.IOException;
+
/**
* A model or service an inference binding calls: one endpoint or one local
binary with its
* settings, named in the {@code "engines"} map. An engine knows nothing about
documents; a
- * binding says what it is fed and what to ask.
+ * binding says what it is fed and what to ask. An engine lives as long as the
config that
+ * loaded it: {@link EngineRegistry#close()} closes every engine once at
shutdown.
*
* @since Apache Tika 4.1
*/
-public interface Engine {
+public interface Engine extends Closeable {
+
+ /** Releases what the engine holds (a client, a native handle); nothing by
default. */
+ @Override
+ default void close() throws IOException {
+ }
}
diff --git
a/tika-core/src/main/java/org/apache/tika/parser/inference/EngineRegistry.java
b/tika-core/src/main/java/org/apache/tika/parser/inference/EngineRegistry.java
index 39a57d3cc7..33944a33c3 100644
---
a/tika-core/src/main/java/org/apache/tika/parser/inference/EngineRegistry.java
+++
b/tika-core/src/main/java/org/apache/tika/parser/inference/EngineRegistry.java
@@ -16,12 +16,22 @@
*/
package org.apache.tika.parser.inference;
+import java.io.Closeable;
+import java.io.IOException;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.Map;
-/** The {@code "engines"} map: user-chosen name to engine, built once at
config load. */
-public final class EngineRegistry {
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * The {@code "engines"} map: user-chosen name to engine, built once at config
load and
+ * closed once with it.
+ */
+public final class EngineRegistry implements Closeable {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(EngineRegistry.class);
private final Map<String, Engine> engines;
@@ -36,4 +46,23 @@ public final class EngineRegistry {
public Map<String, Engine> getEngines() {
return engines;
}
+
+ /** Closes every engine; one failure does not skip the rest, the first is
rethrown. */
+ @Override
+ public void close() throws IOException {
+ IOException first = null;
+ for (Map.Entry<String, Engine> e : engines.entrySet()) {
+ try {
+ e.getValue().close();
+ } catch (IOException | RuntimeException ex) {
+ LOG.warn("engine {} failed to close", e.getKey(), ex);
+ if (first == null) {
+ first = ex instanceof IOException io ? io : new
IOException(ex);
+ }
+ }
+ }
+ if (first != null) {
+ throw first;
+ }
+ }
}
diff --git
a/tika-core/src/test/java/org/apache/tika/parser/inference/EngineRegistryTest.java
b/tika-core/src/test/java/org/apache/tika/parser/inference/EngineRegistryTest.java
new file mode 100644
index 0000000000..7ec3aa81c5
--- /dev/null
+++
b/tika-core/src/test/java/org/apache/tika/parser/inference/EngineRegistryTest.java
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.tika.parser.inference;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import org.junit.jupiter.api.Test;
+
+public class EngineRegistryTest {
+
+ static final class Recording implements Engine {
+ final IOException failure;
+ boolean closed;
+
+ Recording(IOException failure) {
+ this.failure = failure;
+ }
+
+ @Override
+ public void close() throws IOException {
+ closed = true;
+ if (failure != null) {
+ throw failure;
+ }
+ }
+ }
+
+ @Test
+ public void testCloseReachesEveryEngineAndRethrowsTheFirstFailure() {
+ Recording a = new Recording(null);
+ Recording b = new Recording(new IOException("b down"));
+ Recording c = new Recording(new IOException("c down"));
+ Map<String, Engine> engines = new LinkedHashMap<>();
+ engines.put("a", a);
+ engines.put("b", b);
+ engines.put("c", c);
+ engines.put("d", new Engine() { });
+
+ IOException thrown = assertThrows(IOException.class,
+ () -> new EngineRegistry(engines).close());
+ assertEquals("b down", thrown.getMessage());
+ assertTrue(a.closed && b.closed && c.closed, "a failure does not skip
the rest");
+ }
+}
diff --git
a/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/OpenAIEmbeddingEngine.java
b/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/OpenAIEmbeddingEngine.java
index 517844eac4..a6d431d3f9 100644
---
a/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/OpenAIEmbeddingEngine.java
+++
b/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/OpenAIEmbeddingEngine.java
@@ -16,7 +16,6 @@
*/
package org.apache.tika.inference;
-import java.io.Closeable;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Base64;
@@ -44,7 +43,7 @@ import org.apache.tika.utils.StringUtils;
* {@code {"image": "data:<mime>;base64,..."}} inputs and reads the vectors
back by index.
*/
@TikaComponent(name = "openai-embedding-engine", spi = false)
-public class OpenAIEmbeddingEngine implements EmbeddingEngine, Initializable,
Closeable {
+public class OpenAIEmbeddingEngine implements EmbeddingEngine, Initializable {
private static final ObjectMapper MAPPER = new ObjectMapper();
diff --git
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/PipesServer.java
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/PipesServer.java
index a900caad76..df5f37e3f6 100644
---
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/PipesServer.java
+++
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/PipesServer.java
@@ -62,6 +62,7 @@ import
org.apache.tika.metadata.writelimiter.MetadataWriteLimiterFactory;
import org.apache.tika.parser.AutoDetectParser;
import org.apache.tika.parser.ParseContext;
import org.apache.tika.parser.RecursiveParserWrapper;
+import org.apache.tika.parser.inference.EngineRegistry;
import org.apache.tika.parser.inference.InferenceDispatcher;
import org.apache.tika.pipes.api.FetchEmitTuple;
import org.apache.tika.pipes.api.PipesResult;
@@ -186,6 +187,7 @@ public class PipesServer implements AutoCloseable {
private final PipesConfig pipesConfig;
private final Socket socket;
private final MetadataFilter defaultMetadataFilter;
+ private final EngineRegistry engineRegistry;
private final InferenceDispatcher inferenceDispatcher;
private final ContentHandlerFactory defaultContentHandlerFactory;
private final MetadataWriteLimiterFactory
defaultMetadataWriteLimiterFactory;
@@ -259,6 +261,7 @@ public class PipesServer implements AutoCloseable {
this.socket = socket;
socket.setSoTimeout((int) pipesConfig.getSocketTimeoutMillis());
this.defaultMetadataFilter = tikaLoader.loadMetadataFilters();
+ this.engineRegistry = tikaLoader.get(EngineRegistry.class);
this.inferenceDispatcher = tikaLoader.get(InferenceDispatcher.class);
this.defaultContentHandlerFactory =
tikaLoader.loadContentHandlerFactory();
this.defaultMetadataWriteLimiterFactory =
configContext.get(MetadataWriteLimiterFactory.class);
@@ -407,6 +410,7 @@ public class PipesServer implements AutoCloseable {
} finally {
connectionPool.shutdownNow();
connectionPool.awaitTermination(10, TimeUnit.SECONDS);
+ resources.close();
LOG.debug("Shared server shutdown complete");
}
}
@@ -715,6 +719,9 @@ public class PipesServer implements AutoCloseable {
executorService.shutdownNow();
socket.close();
defaultMetadataFilter.close();
+ if (engineRegistry != null) {
+ engineRegistry.close();
+ }
}
private void exit(int exitCode) {
diff --git
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/SharedServerResources.java
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/SharedServerResources.java
index 80aa8e4c18..6f5b623694 100644
---
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/SharedServerResources.java
+++
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/SharedServerResources.java
@@ -16,6 +16,7 @@
*/
package org.apache.tika.pipes.core.server;
+import java.io.Closeable;
import java.io.IOException;
import org.xml.sax.SAXException;
@@ -32,6 +33,7 @@ import
org.apache.tika.metadata.writelimiter.MetadataWriteLimiterFactory;
import org.apache.tika.parser.AutoDetectParser;
import org.apache.tika.parser.ParseContext;
import org.apache.tika.parser.RecursiveParserWrapper;
+import org.apache.tika.parser.inference.EngineRegistry;
import org.apache.tika.parser.inference.InferenceDispatcher;
import org.apache.tika.pipes.core.EmitStrategy;
import org.apache.tika.pipes.core.PipesConfig;
@@ -53,7 +55,7 @@ import org.apache.tika.sax.ContentHandlerFactory;
* <li>MetadataFilter, ContentHandlerFactory - stateless</li>
* </ul>
*/
-public class SharedServerResources {
+public class SharedServerResources implements Closeable {
private final TikaLoader tikaLoader;
private final PipesConfig pipesConfig;
@@ -64,6 +66,7 @@ public class SharedServerResources {
private final EmitterManager emitterManager;
private final MetadataFilter defaultMetadataFilter;
private final InferenceDispatcher inferenceDispatcher;
+ private final EngineRegistry engineRegistry;
private final ContentHandlerFactory defaultContentHandlerFactory;
private final MetadataWriteLimiterFactory
defaultMetadataWriteLimiterFactory;
private final EmitStrategy emitStrategy;
@@ -75,7 +78,7 @@ public class SharedServerResources {
AutoDetectParser autoDetectParser, Detector
detector,
RecursiveParserWrapper rMetaParser,
FetcherManager fetcherManager,
EmitterManager emitterManager,
MetadataFilter defaultMetadataFilter,
- InferenceDispatcher inferenceDispatcher,
+ InferenceDispatcher inferenceDispatcher,
EngineRegistry engineRegistry,
ContentHandlerFactory
defaultContentHandlerFactory,
MetadataWriteLimiterFactory
defaultMetadataWriteLimiterFactory,
EmitStrategy emitStrategy, ConfigStore
configStore,
@@ -90,6 +93,7 @@ public class SharedServerResources {
this.emitterManager = emitterManager;
this.defaultMetadataFilter = defaultMetadataFilter;
this.inferenceDispatcher = inferenceDispatcher;
+ this.engineRegistry = engineRegistry;
this.defaultContentHandlerFactory = defaultContentHandlerFactory;
this.defaultMetadataWriteLimiterFactory =
defaultMetadataWriteLimiterFactory;
this.emitStrategy = emitStrategy;
@@ -128,6 +132,7 @@ public class SharedServerResources {
// Load filters and factories
MetadataFilter metadataFilter = tikaLoader.loadMetadataFilters();
+ EngineRegistry engineRegistry = tikaLoader.get(EngineRegistry.class);
InferenceDispatcher inferenceDispatcher =
tikaLoader.get(InferenceDispatcher.class);
ContentHandlerFactory contentHandlerFactory =
tikaLoader.loadContentHandlerFactory();
ParseContext configContext = tikaLoader.loadParseContext();
@@ -142,7 +147,7 @@ public class SharedServerResources {
return new SharedServerResources(tikaLoader, pipesConfig,
autoDetectParser, detector,
rMetaParser, fetcherManager, emitterManager, metadataFilter,
inferenceDispatcher,
- contentHandlerFactory,
+ engineRegistry, contentHandlerFactory,
metadataWriteLimiterFactory, emitStrategy, configStore,
ExceptionReporting.get(configContext), presetRegistry);
}
@@ -217,6 +222,18 @@ public class SharedServerResources {
return defaultMetadataFilter;
}
+ /** Closes what the config loaded and holds open: the filter chain and the
engines. */
+ @Override
+ public void close() throws IOException {
+ try {
+ defaultMetadataFilter.close();
+ } finally {
+ if (engineRegistry != null) {
+ engineRegistry.close();
+ }
+ }
+ }
+
/** Null when no {@code "inference"} bindings are configured. */
public InferenceDispatcher getInferenceDispatcher() {
return inferenceDispatcher;
diff --git
a/tika-serialization/src/test/java/org/apache/tika/config/loader/EngineLoaderTest.java
b/tika-serialization/src/test/java/org/apache/tika/config/loader/EngineLoaderTest.java
index 034083f947..59f94446e6 100644
---
a/tika-serialization/src/test/java/org/apache/tika/config/loader/EngineLoaderTest.java
+++
b/tika-serialization/src/test/java/org/apache/tika/config/loader/EngineLoaderTest.java
@@ -20,6 +20,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import java.nio.file.Files;
import java.nio.file.Path;
@@ -51,6 +52,10 @@ public class EngineLoaderTest {
assertEquals("second", ((TestEngine) engines.get("two")).getLabel());
assertNull(engines.get("three"));
assertEquals(2, engines.getEngines().size());
+
+ engines.close();
+ assertTrue(((TestEngine) engines.get("one")).isClosed());
+ assertTrue(((TestEngine) engines.get("two")).isClosed());
}
@Test
diff --git
a/tika-serialization/src/test/java/org/apache/tika/config/loader/TestEngine.java
b/tika-serialization/src/test/java/org/apache/tika/config/loader/TestEngine.java
index d5559c7b00..4ca1c3b6d6 100644
---
a/tika-serialization/src/test/java/org/apache/tika/config/loader/TestEngine.java
+++
b/tika-serialization/src/test/java/org/apache/tika/config/loader/TestEngine.java
@@ -24,6 +24,16 @@ import org.apache.tika.parser.inference.Engine;
public class TestEngine implements Engine {
private String label = "";
+ private boolean closed;
+
+ @Override
+ public void close() {
+ closed = true;
+ }
+
+ public boolean isClosed() {
+ return closed;
+ }
public String getLabel() {
return label;
