TIKA-5001: step 2, fetcher/emitter leases and close

CHANGES.txt

@@ -1,4 +1,13 @@
-Release 4.2.1 - unreleased
+Release 4.3.0 - ???
+
+  BREAKING CHANGES AND DEPRECATIONS
+
+   * FetcherManager.getFetcher and EmitterManager.getEmitter return a Lease
+     (AbstractComponentManager.getComponent is now acquire); close it when done with
+     the component and anything it returned. Pipes now closes a
+     Closeable fetcher or emitter when a runtime save or delete replaces it (after the
+     last task using it ends) and on graceful shutdown; nothing closed them before.
+     EmittingUnpackHandler takes the emitter, not the EmitterManager (TIKA-5001).
 
   SERVER, PIPES AND OPERATIONS
 

docs/modules/ROOT/pages/pipes/plugins/writing-a-plugin.adoc

@@ -131,8 +131,13 @@ unknown keys, duplicate keys and numbers-for-enums rejected. Using it is optiona
 boundary is the `Fetcher`, not how it is configured — but see the Jackson rule under
 <<classloading>> before deciding.
 
-IMPORTANT: `Fetcher` implementations must be thread-safe. One instance serves every concurrent
-request against that fetcher id.
+IMPORTANT: `Fetcher` and `Emitter` implementations must be thread-safe. One instance serves
+every concurrent request against that id.
+
+A fetcher or emitter that holds resources (clients, connections, pools) implements
+`java.io.Closeable`. Tika closes it when it is replaced or deleted at runtime, once the last
+task using it ends, and when the pipes server shuts down. A pipes server that is killed
+(timeout, OOM) skips `close()`, so do not rely on it to flush data.
 
 == plugin.properties
 
@@ -404,6 +409,7 @@ java -Dtika.plugin.dev.mode=true ...
   Jackson type is) — never a mix.
 * The zip — not an unpacked directory — dropped in a `plugin-roots` directory.
 * `getName()` unique against every other loaded plugin.
+* `Closeable` on any fetcher or emitter that holds a client or connection.
 
 == Source
 

docs/modules/ROOT/pages/pipes/shared-server-mode.adoc

@@ -91,7 +91,8 @@ load rather than worst-case single document.
 In per-client mode each parse gets its own JVM and a single thread, so thread-safety bugs in
 parsers stay hidden. In shared mode parses run concurrently in one JVM, which can expose
 latent threading bugs in parsers or their dependencies. Strange behaviour that appears only
-in shared mode is most often this.
+in shared mode is most often this. The same holds for fetchers and emitters: one instance
+serves every concurrent request against its id, so shared mode requires thread-safe ones.
 
 == When to Use Shared Mode
 

tika-grpc/src/main/java/org/apache/tika/pipes/grpc/TikaGrpcServerImpl.java

@@ -50,6 +50,7 @@
 import org.apache.tika.pipes.api.fetcher.Fetcher;
 import org.apache.tika.pipes.api.fetcher.FetcherFactory;
 import org.apache.tika.pipes.api.pipesiterator.PipesIteratorFactory;
+import org.apache.tika.pipes.core.Lease;
 import org.apache.tika.pipes.core.PipesConfig;
 import org.apache.tika.pipes.core.PipesException;
 import org.apache.tika.pipes.core.PipesParser;
@@ -383,16 +384,16 @@ private ParseContext buildRequestParseContext(FetchAndParseRequest request,
 
     private void fetchAndParseImpl(FetchAndParseRequest request, ParseContext parseContext,
                                    StreamObserver<FetchAndParseReply> responseObserver) {
-        Fetcher fetcher;
-        try {
-            fetcher = fetcherManager.getFetcher(request.getFetcherId());
+        String fetcherId;
+        try (Lease<Fetcher> lease = fetcherManager.getFetcher(request.getFetcherId())) {
+            fetcherId = lease.get().getExtensionConfig().id();
         } catch (TikaException | IOException e) {
             throw new RuntimeException("Could not find fetcher with name " + request.getFetcherId(), e);
         }
 
         Metadata tikaMetadata = new Metadata();
         try {
-            PipesResult pipesResult = pipesParser.parse(new FetchEmitTuple(request.getFetchKey(), new FetchKey(fetcher.getExtensionConfig().id(), request.getFetchKey()),
+            PipesResult pipesResult = pipesParser.parse(new FetchEmitTuple(request.getFetchKey(), new FetchKey(fetcherId, request.getFetchKey()),
                     new EmitKey(), tikaMetadata, parseContext, FetchEmitTuple.ON_PARSE_EXCEPTION.SKIP));
             FetchAndParseReply.Builder fetchReplyBuilder =
                     FetchAndParseReply.newBuilder()
@@ -483,9 +484,8 @@ static Status notFoundStatus(String fetcherId) {
     public void getFetcher(GetFetcherRequest request,
                            StreamObserver<GetFetcherReply> responseObserver) {
         GetFetcherReply.Builder getFetcherReply = GetFetcherReply.newBuilder();
-        try {
-            Fetcher fetcher = fetcherManager.getFetcher(request.getFetcherId());
-            ExtensionConfig config = fetcher.getExtensionConfig();
+        try (Lease<Fetcher> lease = fetcherManager.getFetcher(request.getFetcherId())) {
+            ExtensionConfig config = lease.get().getExtensionConfig();
 
             getFetcherReply.setFetcherId(config.id());
             getFetcherReply.setFetcherType(config.name());
@@ -512,9 +512,8 @@ public void listFetchers(ListFetchersRequest request,
         // The config may carry secrets; only include it once component management is enabled.
         boolean includeConfig = tikaGrpcConfig.isAllowComponentManagement();
         for (String fetcherId : fetcherManager.getSupported()) {
-            try {
-                Fetcher fetcher = fetcherManager.getFetcher(fetcherId);
-                ExtensionConfig config = fetcher.getExtensionConfig();
+            try (Lease<Fetcher> lease = fetcherManager.getFetcher(fetcherId)) {
+                ExtensionConfig config = lease.get().getExtensionConfig();
 
                 GetFetcherReply.Builder replyBuilder = GetFetcherReply.newBuilder().setFetcherId(config.id()).setFetcherType(config.name());
 

tika-integration-tests/tika-pipes-es-integration-tests/src/test/java/org/apache/tika/pipes/elasticsearch/tests/ElasticsearchTest.java

@@ -400,7 +400,7 @@ public void testUpsert(
         Emitter emitter = EmitterManager
                 .load(TikaPluginManager.load(tikaJsonConfig),
                         tikaJsonConfig)
-                .getEmitter();
+                .getEmitter().get();
         Metadata metadata = new Metadata();
         metadata.set("mime", "mimeA");
         metadata.set("title", "titleA");
@@ -484,7 +484,7 @@ public void testVectorSearch(
             TikaJsonConfig tikaJsonConfig = TikaJsonConfig.load(pluginsConfigFile);
             Emitter emitter = EmitterManager
                     .load(TikaPluginManager.load(tikaJsonConfig), tikaJsonConfig)
-                    .getEmitter();
+                    .getEmitter().get();
 
             for (Object[] doc : docs) {
                 String id = (String) doc[0];

tika-integration-tests/tika-pipes-opensearch-integration-tests/src/test/java/org/apache/tika/pipes/opensearch/tests/OpenSearchTest.java

@@ -373,7 +373,7 @@ public void testUpsert(@TempDir Path pipesDirectory, @TempDir Path testDocDirect
 
         TikaJsonConfig tikaJsonConfig = TikaJsonConfig.load(pluginsConfigFile);
         Emitter emitter = EmitterManager
-                .load(TikaPluginManager.load(tikaJsonConfig), tikaJsonConfig).getEmitter();
+                .load(TikaPluginManager.load(tikaJsonConfig), tikaJsonConfig).getEmitter().get();
         Metadata metadata = new Metadata();
         metadata.set("mime", "mimeA");
         metadata.set("title", "titleA");

tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/AbstractComponentManager.java

@@ -16,6 +16,7 @@
  */
 package org.apache.tika.pipes.core;
 
+import java.io.Closeable;
 import java.io.IOException;
 import java.util.HashMap;
 import java.util.Iterator;
@@ -46,12 +47,17 @@
 /**
  * Abstract base class for managing Tika components (Fetchers, Emitters, etc.).
  * Provides lazy instantiation, early validation, and optional runtime modifications.
+ * <p>
+ * The manager owns every instance it builds. Callers {@link #acquire(String)} a {@link Lease}
+ * and close it when done. A runtime save or delete retires the instance it replaces: no new
+ * lease can get it, and it is closed (if it implements {@link AutoCloseable}) when its last
+ * lease is released, at once if none is open. {@link #close()} retires every instance.
  *
  * @param <T> the component type (e.g., Fetcher, Emitter)
  * @param <F> the factory type for creating components
  */
 public abstract class AbstractComponentManager<T extends TikaExtension,
-                                                F extends TikaExtensionFactory<T>> {
+                                                F extends TikaExtensionFactory<T>> implements Closeable {
 
     private static final Logger LOG = LoggerFactory.getLogger(AbstractComponentManager.class);
 
@@ -61,8 +67,9 @@ public abstract class AbstractComponentManager<T extends TikaExtension,
     private final Map<String, ExtensionConfig> configured;
     // runtime registrations only
     private final ConfigStore configStore;
-    private final Map<String, T> componentCache = new ConcurrentHashMap<>();
+    private final Map<String, Holder<T>> holders = new ConcurrentHashMap<>();
     private final boolean allowRuntimeModifications;
+    private volatile boolean closed = false;
 
     protected AbstractComponentManager(PluginManager pluginManager,
                                       Map<String, ExtensionConfig> componentConfigs,
@@ -240,44 +247,66 @@ private static String toJsonString(final JsonNode node) throws TikaConfigExcepti
     }
 
     /**
-     * Gets a component by ID, lazily instantiating it if needed.
+     * Leases a component by ID, lazily instantiating it if needed. Close the lease when done
+     * with the component and with everything it returned.
      */
-    public T getComponent(String id) throws IOException, TikaException {
+    public Lease<T> acquire(String id) throws IOException, TikaException {
         // Canonicalize on the way in so lookup agrees with what load() stored. __ is not refused
         // here: the host's own components resolve through this method.
         id = ComponentIds.normalize(id);
-
-        // Check cache first (fast path, no synchronization)
-        T component = componentCache.get(id);
-        if (component != null) {
-            return component;
+        while (true) {
+            Holder<T> holder = holders.get(id);
+            if (holder == null) {
+                holder = build(id);
+            }
+            Lease<T> lease = holder.tryAcquire();
+            if (lease != null) {
+                return lease;
+            }
+            // retired between lookup and acquire: build the replacement
         }
+    }
 
-        // Check if config exists
+    private synchronized Holder<T> build(String id) throws IOException, TikaException {
+        if (closed) {
+            throw new IllegalStateException(getClass().getSimpleName() + " is closed");
+        }
+        Holder<T> holder = holders.get(id);
+        if (holder != null) {
+            return holder;
+        }
         ExtensionConfig config = lookup(id);
         if (config == null) {
             throw createNotFoundException(
                     "Can't find " + getComponentName() + " for id=" + id +
                     ". Available: " + getSupported());
         }
+        try {
+            holder = new Holder<>(buildComponent(config), getComponentName(), id);
+        } catch (TikaConfigException e) {
+            throw new IOException("Failed to build " + getComponentName() + ": " + id, e);
+        }
+        holders.put(id, holder);
+        LOG.debug("Lazily instantiated {}: {}", getComponentName(), id);
+        return holder;
+    }
 
-        // Synchronized block to ensure only one thread builds the component
-        synchronized (this) {
-            // Double-check in case another thread built it while we were waiting
-            component = componentCache.get(id);
-            if (component != null) {
-                return component;
-            }
+    private synchronized void retire(String id) {
+        Holder<T> holder = holders.remove(id);
+        if (holder != null) {
+            holder.retire();
+        }
+    }
 
-            // Build the component
-            try {
-                component = buildComponent(config);
-                componentCache.put(id, component);
-                LOG.debug("Lazily instantiated {}: {}", getComponentName(), id);
-                return component;
-            } catch (TikaConfigException e) {
-                throw new IOException("Failed to build " + getComponentName() + ": " + id, e);
-            }
+    /**
+     * Retires every instance: each is closed now if no lease is open, else when its last lease
+     * is released. Later calls to {@link #acquire(String)} fail.
+     */
+    @Override
+    public synchronized void close() {
+        closed = true;
+        for (String id : Set.copyOf(holders.keySet())) {
+            retire(id);
         }
     }
 
@@ -313,8 +342,8 @@ private T buildComponent(ExtensionConfig config) throws TikaConfigException, IOE
 
     /**
      * Dynamically adds or updates a component configuration at runtime.
-     * The component will not be instantiated until it is first requested via {@link #getComponent(String)}.
-     * If a component with the same ID already exists, it will be replaced and the cached instance cleared.
+     * The component will not be instantiated until it is first requested via {@link #acquire(String)}.
+     * If a component with the same ID already exists, it will be replaced and its instance retired.
      * A component defined in the server config cannot be replaced.
      * <p>
      * This method is only available if the manager was loaded with allowRuntimeModifications=true.
@@ -358,21 +387,20 @@ public synchronized void saveComponent(ExtensionConfig config) throws TikaConfig
                     ". Available: " + factories.keySet());
         }
 
-        // If updating existing component, clear the cache so it gets re-instantiated
         if (configStore.containsKey(componentId)) {
-            componentCache.remove(componentId);
             LOG.debug("Updating existing {} config: id={}, type={}", getComponentName(), componentId, typeName);
         } else {
             LOG.debug("Creating new {} config: id={}, type={}", getComponentName(), componentId, typeName);
         }
 
         // Store config without instantiating
         configStore.put(componentId, config);
+        retire(componentId);
     }
 
     /**
      * Deletes a component configuration by ID.
-     * Clears the cached instance and removes the configuration.
+     * Removes the configuration and retires the instance.
      *
      * @param componentId the component ID to delete
      * @throws TikaConfigException if runtime modifications are not allowed, the component is not
@@ -405,9 +433,8 @@ public synchronized void deleteComponent(String componentId) throws TikaConfigEx
                     getComponentName().substring(1) + " with ID '" + componentId + "' not found");
         }
 
-        // Clear cache and remove config
-        componentCache.remove(componentId);
         configStore.remove(componentId);
+        retire(componentId);
         LOG.debug("Deleted {} config: id={}", getComponentName(), componentId);
     }
 
@@ -442,21 +469,78 @@ public Set<String> getAllIds() {
     }
 
     /**
-     * Convenience method that returns a component if only one component
+     * Convenience method that leases a component if only one component
      * is configured. If 0 or > 1 components are configured, this throws an IllegalArgumentException.
      *
-     * @return the single configured component
+     * @return a lease on the single configured component
      */
-    public T getComponent() throws IOException, TikaException {
+    public Lease<T> acquire() throws IOException, TikaException {
         Set<String> ids = getAllIds();
         if (ids.size() != 1) {
             throw new IllegalArgumentException(
                     "No-arg get" + getComponentName().substring(0, 1).toUpperCase(Locale.ROOT) +
                     getComponentName().substring(1) + "() requires exactly 1 configured " +
                     getComponentName() + ". Found: " + ids.size() + " (" + ids + ")");
         }
-        // Get the single component id and use getComponent(id) for lazy loading
-        String componentId = ids.iterator().next();
-        return getComponent(componentId);
+        return acquire(ids.iterator().next());
+    }
+
+    /**
+     * One built instance. Closed exactly once, when it is retired and no lease is open,
+     * whichever comes last.
+     */
+    private static final class Holder<T> {
+        private final T component;
+        private final String componentName;
+        private final String id;
+        private int refs = 0;
+        private boolean retired = false;
+
+        Holder(T component, String componentName, String id) {
+            this.component = component;
+            this.componentName = componentName;
+            this.id = id;
+        }
+
+        synchronized Lease<T> tryAcquire() {
+            if (retired) {
+                return null;
+            }
+            refs++;
+            return new Lease<>(component, this::release);
+        }
+
+        private void release() {
+            synchronized (this) {
+                if (--refs > 0 || !retired) {
+                    return;
+                }
+            }
+            closeComponent();
+        }
+
+        void retire() {
+            synchronized (this) {
+                if (retired) {
+                    return;
+                }
+                retired = true;
+                if (refs > 0) {
+                    return;
+                }
+            }
+            closeComponent();
+        }
+
+        private void closeComponent() {
+            if (component instanceof AutoCloseable closeable) {
+                try {
+                    closeable.close();
+                    LOG.debug("Closed {}: {}", componentName, id);
+                } catch (Exception e) {
+                    LOG.warn("Failed to close {} '{}'", componentName, id, e);
+                }
+            }
+        }
     }
 }