diff --git a/CHANGES.txt b/CHANGES.txt index 9826420af33..5086616d589 100644 --- a/CHANGES.txt +++ b/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 diff --git a/docs/modules/ROOT/pages/pipes/plugins/writing-a-plugin.adoc b/docs/modules/ROOT/pages/pipes/plugins/writing-a-plugin.adoc index deca84bea7a..cf225934100 100644 --- a/docs/modules/ROOT/pages/pipes/plugins/writing-a-plugin.adoc +++ b/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 <> 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 diff --git a/docs/modules/ROOT/pages/pipes/shared-server-mode.adoc b/docs/modules/ROOT/pages/pipes/shared-server-mode.adoc index a4957c7695d..3d5f7e587df 100644 --- a/docs/modules/ROOT/pages/pipes/shared-server-mode.adoc +++ b/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 diff --git a/tika-grpc/src/main/java/org/apache/tika/pipes/grpc/TikaGrpcServerImpl.java b/tika-grpc/src/main/java/org/apache/tika/pipes/grpc/TikaGrpcServerImpl.java index 5fb248918c2..cc9f731460a 100644 --- a/tika-grpc/src/main/java/org/apache/tika/pipes/grpc/TikaGrpcServerImpl.java +++ b/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 responseObserver) { - Fetcher fetcher; - try { - fetcher = fetcherManager.getFetcher(request.getFetcherId()); + String fetcherId; + try (Lease 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 responseObserver) { GetFetcherReply.Builder getFetcherReply = GetFetcherReply.newBuilder(); - try { - Fetcher fetcher = fetcherManager.getFetcher(request.getFetcherId()); - ExtensionConfig config = fetcher.getExtensionConfig(); + try (Lease 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 lease = fetcherManager.getFetcher(fetcherId)) { + ExtensionConfig config = lease.get().getExtensionConfig(); GetFetcherReply.Builder replyBuilder = GetFetcherReply.newBuilder().setFetcherId(config.id()).setFetcherType(config.name()); diff --git a/tika-integration-tests/tika-pipes-es-integration-tests/src/test/java/org/apache/tika/pipes/elasticsearch/tests/ElasticsearchTest.java b/tika-integration-tests/tika-pipes-es-integration-tests/src/test/java/org/apache/tika/pipes/elasticsearch/tests/ElasticsearchTest.java index 8f1b6f6e702..f80ecc9785f 100644 --- a/tika-integration-tests/tika-pipes-es-integration-tests/src/test/java/org/apache/tika/pipes/elasticsearch/tests/ElasticsearchTest.java +++ b/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]; diff --git a/tika-integration-tests/tika-pipes-opensearch-integration-tests/src/test/java/org/apache/tika/pipes/opensearch/tests/OpenSearchTest.java b/tika-integration-tests/tika-pipes-opensearch-integration-tests/src/test/java/org/apache/tika/pipes/opensearch/tests/OpenSearchTest.java index 1d646557fba..0eb0a2a7abe 100644 --- a/tika-integration-tests/tika-pipes-opensearch-integration-tests/src/test/java/org/apache/tika/pipes/opensearch/tests/OpenSearchTest.java +++ b/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"); diff --git a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/AbstractComponentManager.java b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/AbstractComponentManager.java index a6e5bc1b9e8..4e576e6fcff 100644 --- a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/AbstractComponentManager.java +++ b/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. + *

+ * 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 the component type (e.g., Fetcher, Emitter) * @param the factory type for creating components */ public abstract class AbstractComponentManager> { + F extends TikaExtensionFactory> implements Closeable { private static final Logger LOG = LoggerFactory.getLogger(AbstractComponentManager.class); @@ -61,8 +67,9 @@ public abstract class AbstractComponentManager configured; // runtime registrations only private final ConfigStore configStore; - private final Map componentCache = new ConcurrentHashMap<>(); + private final Map> holders = new ConcurrentHashMap<>(); private final boolean allowRuntimeModifications; + private volatile boolean closed = false; protected AbstractComponentManager(PluginManager pluginManager, Map 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 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 holder = holders.get(id); + if (holder == null) { + holder = build(id); + } + Lease lease = holder.tryAcquire(); + if (lease != null) { + return lease; + } + // retired between lookup and acquire: build the replacement } + } - // Check if config exists + private synchronized Holder build(String id) throws IOException, TikaException { + if (closed) { + throw new IllegalStateException(getClass().getSimpleName() + " is closed"); + } + Holder 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 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. *

* This method is only available if the manager was loaded with allowRuntimeModifications=true. @@ -358,9 +387,7 @@ 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); @@ -368,11 +395,12 @@ public synchronized void saveComponent(ExtensionConfig config) throws TikaConfig // 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,12 +469,12 @@ public Set 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 acquire() throws IOException, TikaException { Set ids = getAllIds(); if (ids.size() != 1) { throw new IllegalArgumentException( @@ -455,8 +482,65 @@ public T getComponent() throws IOException, TikaException { 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 { + 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 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); + } + } + } } } diff --git a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/Lease.java b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/Lease.java new file mode 100644 index 00000000000..b3031ef4000 --- /dev/null +++ b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/Lease.java @@ -0,0 +1,53 @@ +/* + * 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.pipes.core; + +import java.util.concurrent.atomic.AtomicBoolean; + +/** + * A hold on a fetcher or emitter instance from {@link AbstractComponentManager}. While any lease + * is open the instance is not closed, even if a runtime save or delete has replaced it. Hold it + * for as long as anything the component returned is in use (a fetched stream is read long after + * {@code fetch()} returns), then close it. Closing twice is a no-op. + * + * @param the component type + */ +public final class Lease implements AutoCloseable { + + private final T component; + private final Runnable release; + private final AtomicBoolean released = new AtomicBoolean(false); + + Lease(T component, Runnable release) { + this.component = component; + this.release = release; + } + + /** + * @return the leased component; do not use it after {@link #close()} + */ + public T get() { + return component; + } + + @Override + public void close() { + if (released.compareAndSet(false, true)) { + release.run(); + } + } +} diff --git a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/async/AsyncEmitter.java b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/async/AsyncEmitter.java index 6e1f9b03c7f..bccf6156f2c 100644 --- a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/async/AsyncEmitter.java +++ b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/async/AsyncEmitter.java @@ -33,6 +33,7 @@ import org.apache.tika.exception.TikaException; import org.apache.tika.pipes.api.emitter.EmitData; import org.apache.tika.pipes.api.emitter.Emitter; +import org.apache.tika.pipes.core.Lease; import org.apache.tika.pipes.core.PipesConfig; import org.apache.tika.pipes.core.emitter.EmitterManager; @@ -126,14 +127,12 @@ private void emitAll() { int emitted = 0; LOG.debug("about to emit {} files, {} estimated bytes", size, estimatedSize); for (Map.Entry> e : map.entrySet()) { - Emitter emitter = null; - try { - emitter = emitterManager.getEmitter(e.getKey()); + try (Lease lease = emitterManager.getEmitter(e.getKey())) { + tryToEmit(e.getKey(), lease.get(), e.getValue()); } catch (IOException | TikaException ex) { LOG.warn("emitter id={} failed on instantiation", e.getKey(), ex); return; } - tryToEmit(e.getKey(), emitter, e.getValue()); emitted += e.getValue().size(); } diff --git a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/async/AsyncProcessor.java b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/async/AsyncProcessor.java index 6d2ab6d27a3..d260c7d0449 100644 --- a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/async/AsyncProcessor.java +++ b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/async/AsyncProcessor.java @@ -87,6 +87,7 @@ public class AsyncProcessor implements Closeable { private volatile int numParserThreadsFinished = 0; private volatile int numEmitterThreadsFinished = 0; private boolean addedEmitterSemaphores = false; + private volatile EmitterManager emitterManager; boolean isShuttingDown = false; /** @@ -188,7 +189,7 @@ private AsyncProcessor(PipesIterator pipesIterator, TikaJsonConfig tikaJsonConfi } } - EmitterManager emitterManager = EmitterManager.load(tikaPluginManager, tikaJsonConfig); + emitterManager = EmitterManager.load(tikaPluginManager, tikaJsonConfig); for (int i = 0; i < asyncConfig.getNumEmitters(); i++) { executorCompletionService.submit( new AsyncEmitter(asyncConfig, emitDatumTuples, emitterManager)); @@ -392,6 +393,9 @@ public synchronized boolean checkActive() throws InterruptedException { public void close() throws IOException { executorService.shutdownNow(); closeServerManagers(); + if (emitterManager != null) { + emitterManager.close(); + } this.pipesReporter.close(); } diff --git a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/emitter/EmitterManager.java b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/emitter/EmitterManager.java index 360eb9bb7db..18746e9d035 100644 --- a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/emitter/EmitterManager.java +++ b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/emitter/EmitterManager.java @@ -29,6 +29,7 @@ import org.apache.tika.pipes.api.emitter.EmitterFactory; import org.apache.tika.pipes.api.emitter.EmitterNotFoundException; import org.apache.tika.pipes.core.AbstractComponentManager; +import org.apache.tika.pipes.core.Lease; import org.apache.tika.plugins.ExtensionConfig; /** @@ -132,35 +133,36 @@ protected TikaException createNotFoundException(String message) { } /** - * Gets an emitter by ID, lazily instantiating it if needed. + * Leases an emitter by ID, lazily instantiating it if needed. Close the lease when the + * emitter and everything it returned are no longer in use. * * @param emitterName the emitter ID - * @return the emitter + * @return a lease on the emitter * @throws EmitterNotFoundException if no emitter with the given ID exists * @throws IOException if there's an error building the emitter * @throws TikaException if there's a configuration error */ - public Emitter getEmitter(String emitterName) throws IOException, TikaException { - return getComponent(emitterName); + public Lease getEmitter(String emitterName) throws IOException, TikaException { + return acquire(emitterName); } /** - * Convenience method that returns an emitter if only one emitter + * Convenience method that leases an emitter if only one emitter * is configured. If 0 or > 1 emitters are configured, this throws an IllegalArgumentException. * - * @return the single configured emitter + * @return a lease on the single configured emitter * @throws IOException if there's an error building the emitter * @throws TikaException if there's a configuration error */ - public Emitter getEmitter() throws IOException, TikaException { - return getComponent(); + public Lease getEmitter() throws IOException, TikaException { + return acquire(); } /** * Dynamically adds or updates an emitter configuration at runtime. * The emitter will not be instantiated until it is first requested via {@link #getEmitter(String)}. * This allows for dynamic configuration without the overhead of immediate instantiation. - * If an emitter with the same ID already exists, it will be replaced and the cached instance cleared. + * If an emitter with the same ID already exists, it will be replaced and its instance retired. * An emitter defined in the server config cannot be replaced. *

* This method is only available if the EmitterManager was loaded with diff --git a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/extractor/EmittingUnpackHandler.java b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/extractor/EmittingUnpackHandler.java index 65cef5fac19..c76459f22ac 100644 --- a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/extractor/EmittingUnpackHandler.java +++ b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/extractor/EmittingUnpackHandler.java @@ -16,7 +16,6 @@ */ package org.apache.tika.pipes.core.extractor; -import java.io.Closeable; import java.io.IOException; import java.io.InputStream; @@ -28,7 +27,6 @@ import org.apache.tika.pipes.api.emitter.EmitKey; import org.apache.tika.pipes.api.emitter.Emitter; import org.apache.tika.pipes.api.emitter.StreamEmitter; -import org.apache.tika.pipes.core.emitter.EmitterManager; import org.apache.tika.pipes.core.emitter.TikaEmitterException; public class EmittingUnpackHandler extends AbstractUnpackHandler { @@ -39,23 +37,23 @@ public class EmittingUnpackHandler extends AbstractUnpackHandler { private static final Metadata METADATA = new Metadata(); private static final ParseContext PARSE_CONTEXT = new ParseContext(); - public EmittingUnpackHandler(FetchEmitTuple fetchEmitTuple, - EmitterManager emitterManager, - ParseContext parseContext) throws TikaException, IOException { + /** + * @param emitter the caller keeps it open (leased) until this handler is closed + */ + public EmittingUnpackHandler(FetchEmitTuple fetchEmitTuple, Emitter emitter, + ParseContext parseContext) throws TikaException { this.containerEmitKey = fetchEmitTuple.getEmitKey(); this.unpackConfig = parseContext.get(UnpackConfig.class); if (this.unpackConfig == null) { throw new TikaConfigException("UnpackConfig must not be null!"); } - Emitter tmpEmitter = - emitterManager.getEmitter(unpackConfig.getEmitter()); - if (! (tmpEmitter instanceof StreamEmitter)) { + if (! (emitter instanceof StreamEmitter)) { throw new TikaConfigException("Emitter " + unpackConfig.getEmitter() + " must implement a StreamEmitter"); } - this.emitter = (StreamEmitter) tmpEmitter; + this.emitter = (StreamEmitter) emitter; } @Override @@ -71,9 +69,7 @@ public void add(int id, Metadata metadata, InputStream inputStream) throws IOExc } @Override - public void close() throws IOException { - if (emitter instanceof Closeable) { - ((Closeable) emitter).close(); - } + public void close() { + // the emitter is leased by the task, not owned here } } diff --git a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/fetcher/FetcherManager.java b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/fetcher/FetcherManager.java index 0880d383ef4..8b2ed5ef4fa 100644 --- a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/fetcher/FetcherManager.java +++ b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/fetcher/FetcherManager.java @@ -29,6 +29,7 @@ import org.apache.tika.pipes.api.fetcher.FetcherFactory; import org.apache.tika.pipes.api.fetcher.FetcherNotFoundException; import org.apache.tika.pipes.core.AbstractComponentManager; +import org.apache.tika.pipes.core.Lease; import org.apache.tika.plugins.ExtensionConfig; /** @@ -131,35 +132,36 @@ protected TikaException createNotFoundException(String message) { } /** - * Gets a fetcher by ID, lazily instantiating it if needed. + * Leases a fetcher by ID, lazily instantiating it if needed. Close the lease when the + * fetcher and everything it returned are no longer in use. * * @param id the fetcher ID - * @return the fetcher + * @return a lease on the fetcher * @throws FetcherNotFoundException if no fetcher with the given ID exists * @throws IOException if there's an error building the fetcher * @throws TikaException if there's a configuration error */ - public Fetcher getFetcher(String id) throws IOException, TikaException { - return getComponent(id); + public Lease getFetcher(String id) throws IOException, TikaException { + return acquire(id); } /** - * Convenience method that returns a fetcher if only one fetcher + * Convenience method that leases a fetcher if only one fetcher * is configured. If 0 or > 1 fetchers are configured, this throws an IllegalArgumentException. * - * @return the single configured fetcher + * @return a lease on the single configured fetcher * @throws IOException if there's an error building the fetcher * @throws TikaException if there's a configuration error */ - public Fetcher getFetcher() throws IOException, TikaException { - return getComponent(); + public Lease getFetcher() throws IOException, TikaException { + return acquire(); } /** * Dynamically adds or updates a fetcher configuration at runtime. * The fetcher will not be instantiated until it is first requested via {@link #getFetcher(String)}. * This allows for dynamic configuration without the overhead of immediate instantiation. - * If a fetcher with the same ID already exists, it will be replaced and the cached instance cleared. + * If a fetcher with the same ID already exists, it will be replaced and its instance retired. * A fetcher defined in the server config cannot be replaced. *

* This method is only available if the FetcherManager was loaded with diff --git a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/ConnectionHandler.java b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/ConnectionHandler.java index 2b16ecf6aff..bfbb0c53e31 100644 --- a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/ConnectionHandler.java +++ b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/ConnectionHandler.java @@ -304,7 +304,9 @@ private void mainLoop() { private PipesWorker createPipesWorker(ArrayBlockingQueue intermediateResult, FetchEmitTuple fetchEmitTuple, ParseContext mergedContext, CountDownLatch countDownLatch) { - FetchHandler fetchHandler = new FetchHandler(resources.getFetcherManager()); + TaskLeases leases = new TaskLeases(resources.getFetcherManager(), + resources.getEmitterManager()); + FetchHandler fetchHandler = new FetchHandler(resources.getFetcherManager(), leases); ParseHandler parseHandler = new ParseHandler(resources.getDetector(), intermediateResult, countDownLatch, resources.getAutoDetectParser(), resources.getRMetaParser(), resources.getDefaultContentHandlerFactory(), pipesConfig.getParseMode()); @@ -312,9 +314,9 @@ private PipesWorker createPipesWorker(ArrayBlockingQueue intermediateR long threshold = (thresholdBytes != null) ? thresholdBytes : EmitStrategyConfig.DEFAULT_DIRECT_EMIT_THRESHOLD_BYTES; EmitHandler emitHandler = new EmitHandler(resources.getDefaultMetadataFilter(), resources.getInferenceDispatcher(), resources.getEmitStrategy(), - resources.getEmitterManager(), threshold); + resources.getEmitterManager(), leases, threshold); return new PipesWorker(fetchEmitTuple, mergedContext, resources.getAutoDetectParser(), - resources.getEmitterManager(), fetchHandler, parseHandler, emitHandler, + leases, fetchHandler, parseHandler, emitHandler, resources.getDefaultMetadataWriteLimiterFactory(), pipesConfig.getParseMode()); } diff --git a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/EmitHandler.java b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/EmitHandler.java index e31fe8cce44..48a9ca7a534 100644 --- a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/EmitHandler.java +++ b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/EmitHandler.java @@ -61,15 +61,18 @@ class EmitHandler { private final InferenceDispatcher inferenceDispatcher; private final EmitStrategy emitStrategy; private final EmitterManager emitterManager; + private final TaskLeases leases; private final long directEmitThresholdBytes; public EmitHandler(MetadataFilter defaultMetadataFilter, InferenceDispatcher inferenceDispatcher, - EmitStrategy emitStrategy, EmitterManager emitterManager, long directEmitThresholdBytes) { + EmitStrategy emitStrategy, EmitterManager emitterManager, TaskLeases leases, + long directEmitThresholdBytes) { this.defaultMetadataFilter = defaultMetadataFilter; this.inferenceDispatcher = inferenceDispatcher; this.emitStrategy = emitStrategy; this.emitterManager = emitterManager; + this.leases = leases; this.directEmitThresholdBytes = directEmitThresholdBytes; } @@ -210,7 +213,7 @@ private PipesResult emit(String taskId, EmitKey emitKey, Emitter emitter = null; try { - emitter = emitterManager.getEmitter(emitKey.getEmitterId()); + emitter = leases.emitter(emitKey.getEmitterId()); } catch (org.apache.tika.pipes.api.emitter.EmitterNotFoundException e) { String noEmitterMsg = "Emitter '" + emitKey.getEmitterId() + "' not found."; // configured ids are operator-facing; log them, don't return them to the caller diff --git a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/FetchHandler.java b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/FetchHandler.java index 418054719e3..5f96cb22fad 100644 --- a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/FetchHandler.java +++ b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/FetchHandler.java @@ -40,9 +40,11 @@ class FetchHandler { private static final BytesFetcher BYTES_FETCHER = new BytesFetcher(); private final FetcherManager fetcherManager; + private final TaskLeases leases; - public FetchHandler(FetcherManager fetcherManager) { + public FetchHandler(FetcherManager fetcherManager, TaskLeases leases) { this.fetcherManager = fetcherManager; + this.leases = leases; } public TisOrResult fetch(FetchEmitTuple fetchEmitTuple, Metadata metadata, ParseContext parseContext) { @@ -68,7 +70,7 @@ private FetcherOrResult getFetcher(FetchEmitTuple t, ParseContext parseContext) return new FetcherOrResult(BYTES_FETCHER, null); } try { - return new FetcherOrResult(fetcherManager.getFetcher(fetcherId), null); + return new FetcherOrResult(leases.fetcher(fetcherId), null); } catch (FetcherNotFoundException e) { String noFetcherMsg = "Fetcher '" + t.getFetchKey().getFetcherId() + "' not found."; // configured ids are operator-facing; log them, don't return them to the caller 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 65bb9eeeac9..84beba799af 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 @@ -523,14 +523,15 @@ public void mainLoop() { private PipesWorker getPipesWorker(ArrayBlockingQueue intermediateResult, FetchEmitTuple fetchEmitTuple, ParseContext mergedContext, CountDownLatch countDownLatch) { - FetchHandler fetchHandler = new FetchHandler(fetcherManager); + TaskLeases leases = new TaskLeases(fetcherManager, emitterManager); + FetchHandler fetchHandler = new FetchHandler(fetcherManager, leases); ParseHandler parseHandler = new ParseHandler(detector, intermediateResult, countDownLatch, autoDetectParser, rMetaParser, defaultContentHandlerFactory, pipesConfig.getParseMode()); Long thresholdBytes = pipesConfig.getEmitStrategy().getThresholdBytes(); long threshold = (thresholdBytes != null) ? thresholdBytes : EmitStrategyConfig.DEFAULT_DIRECT_EMIT_THRESHOLD_BYTES; EmitHandler emitHandler = new EmitHandler(defaultMetadataFilter, inferenceDispatcher, - emitStrategy, emitterManager, threshold); - return new PipesWorker(fetchEmitTuple, mergedContext, autoDetectParser, emitterManager, + emitStrategy, emitterManager, leases, threshold); + return new PipesWorker(fetchEmitTuple, mergedContext, autoDetectParser, leases, fetchHandler, parseHandler, emitHandler, defaultMetadataWriteLimiterFactory, pipesConfig.getParseMode()); } @@ -702,6 +703,13 @@ private void handleCrash(PipesMessageType crashType, String id, Throwable t) { public void close() throws Exception { executorService.shutdownNow(); socket.close(); + // a fetcher or emitter still leased by an interrupted task closes on its release + if (fetcherManager != null) { + fetcherManager.close(); + } + if (emitterManager != null) { + emitterManager.close(); + } defaultMetadataFilter.close(); if (engineRegistry != null) { engineRegistry.close(); diff --git a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/PipesWorker.java b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/PipesWorker.java index ec9415234c5..e6c9e540474 100644 --- a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/PipesWorker.java +++ b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/PipesWorker.java @@ -58,7 +58,6 @@ import org.apache.tika.pipes.api.emitter.Emitter; import org.apache.tika.pipes.api.emitter.StreamEmitter; import org.apache.tika.pipes.core.PipesResults; -import org.apache.tika.pipes.core.emitter.EmitterManager; import org.apache.tika.pipes.core.extractor.EmittingUnpackHandler; import org.apache.tika.pipes.core.extractor.FrictionlessUnpackHandler; import org.apache.tika.pipes.core.extractor.TempFileUnpackHandler; @@ -79,7 +78,7 @@ class PipesWorker implements Callable { private final FetchEmitTuple fetchEmitTuple; private final ParseContext parseContext; private final AutoDetectParser autoDetectParser; - private final EmitterManager emitterManager; + private final TaskLeases leases; private final FetchHandler fetchHandler; private final ParseHandler parseHandler; private final EmitHandler emitHandler; @@ -87,13 +86,13 @@ class PipesWorker implements Callable { private final ParseMode defaultParseMode; public PipesWorker(FetchEmitTuple fetchEmitTuple, ParseContext parseContext, AutoDetectParser autoDetectParser, - EmitterManager emitterManager, FetchHandler fetchHandler, ParseHandler parseHandler, + TaskLeases leases, FetchHandler fetchHandler, ParseHandler parseHandler, EmitHandler emitHandler, MetadataWriteLimiterFactory defaultMetadataWriteLimiterFactory, ParseMode defaultParseMode) { this.fetchEmitTuple = fetchEmitTuple; this.parseContext = parseContext; this.autoDetectParser = autoDetectParser; - this.emitterManager = emitterManager; + this.leases = leases; this.fetchHandler = fetchHandler; this.parseHandler = parseHandler; this.emitHandler = emitHandler; @@ -196,6 +195,8 @@ private PipesResult doCall() throws Exception { LOG.warn("problem closing unpack handler", e); } } + // after the unpack handler, which may still be emitting + leases.close(); } } @@ -223,7 +224,7 @@ private PipesResult zipAndEmitEmbeddedFiles(TempFileUnpackHandler tempHandler, String emitterName = unpackConfig.getEmitter(); Emitter emitter; try { - emitter = emitterManager.getEmitter(emitterName); + emitter = leases.emitter(emitterName); } catch (Exception e) { LOG.warn("Failed to get emitter for zip: {}", emitterName, e); return new PipesResult(PipesResult.RESULT_STATUS.EMITTER_NOT_FOUND, @@ -288,7 +289,7 @@ private PipesResult emitFrictionlessOutput(FrictionlessUnpackHandler frictionles Emitter emitter; try { - emitter = emitterManager.getEmitter(emitterName); + emitter = leases.emitter(emitterName); } catch (Exception e) { LOG.warn("Failed to get emitter for Frictionless output: {}", emitterName, e); return new PipesResult(PipesResult.RESULT_STATUS.EMITTER_NOT_FOUND, @@ -696,7 +697,8 @@ private ParseContext setupParseContext() throws TikaException, IOException { } else { // Regular format, direct emission - use EmittingUnpackHandler parseContext.set(UnpackHandler.class, - new EmittingUnpackHandler(fetchEmitTuple, emitterManager, parseContext)); + new EmittingUnpackHandler(fetchEmitTuple, + leases.emitter(unpackConfig.getEmitter()), parseContext)); } return parseContext; 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 314da6d9602..36dac29b4c3 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 @@ -222,9 +222,14 @@ public MetadataFilter getDefaultMetadataFilter() { return defaultMetadataFilter; } - /** Closes what the config loaded and holds open: the filter chain and the engines. */ + /** + * Closes what the config loaded and holds open: fetchers, emitters, the filter chain and + * the engines. + */ @Override public void close() throws IOException { + fetcherManager.close(); + emitterManager.close(); try { defaultMetadataFilter.close(); } finally { diff --git a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/TaskLeases.java b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/TaskLeases.java new file mode 100644 index 00000000000..70b507e0d0c --- /dev/null +++ b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/TaskLeases.java @@ -0,0 +1,71 @@ +/* + * 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.pipes.core.server; + +import java.io.IOException; +import java.util.HashMap; +import java.util.Map; + +import org.apache.tika.exception.TikaException; +import org.apache.tika.pipes.api.emitter.Emitter; +import org.apache.tika.pipes.api.fetcher.Fetcher; +import org.apache.tika.pipes.core.AbstractComponentManager; +import org.apache.tika.pipes.core.Lease; + +/** + * The fetcher and emitter leases one task holds, one per id, released together when the task + * ends: a runtime save or delete mid-task cannot close an instance the task is still using. + */ +class TaskLeases implements AutoCloseable { + + private final AbstractComponentManager fetcherManager; + private final AbstractComponentManager emitterManager; + private final Map> fetchers = new HashMap<>(); + private final Map> emitters = new HashMap<>(); + + TaskLeases(AbstractComponentManager fetcherManager, + AbstractComponentManager emitterManager) { + this.fetcherManager = fetcherManager; + this.emitterManager = emitterManager; + } + + synchronized Fetcher fetcher(String id) throws IOException, TikaException { + Lease lease = fetchers.get(id); + if (lease == null) { + lease = fetcherManager.acquire(id); + fetchers.put(id, lease); + } + return lease.get(); + } + + synchronized Emitter emitter(String id) throws IOException, TikaException { + Lease lease = emitters.get(id); + if (lease == null) { + lease = emitterManager.acquire(id); + emitters.put(id, lease); + } + return lease.get(); + } + + @Override + public synchronized void close() { + fetchers.values().forEach(Lease::close); + emitters.values().forEach(Lease::close); + fetchers.clear(); + emitters.clear(); + } +} diff --git a/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/ComponentLeaseTest.java b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/ComponentLeaseTest.java new file mode 100644 index 00000000000..c9722fdb5d0 --- /dev/null +++ b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/ComponentLeaseTest.java @@ -0,0 +1,164 @@ +/* + * 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.pipes.core; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotSame; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import java.util.ArrayList; +import java.util.List; +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 org.junit.jupiter.api.Test; + +import org.apache.tika.pipes.api.fetcher.Fetcher; +import org.apache.tika.pipes.core.TrackingFetcherManager.TrackedFetcher; + +public class ComponentLeaseTest { + + @Test + public void testIdleInstanceClosedOnDelete() throws Exception { + TrackingFetcherManager manager = new TrackingFetcherManager(); + manager.saveComponent(TrackingFetcherManager.config("f")); + manager.acquire("f").close(); + TrackedFetcher fetcher = manager.getBuilt().get(0); + assertEquals(0, fetcher.getCloses(), "a released instance stays open while it is current"); + + manager.deleteComponent("f"); + assertEquals(1, fetcher.getCloses()); + } + + @Test + public void testLeasedInstanceClosedOnLastRelease() throws Exception { + TrackingFetcherManager manager = new TrackingFetcherManager(); + manager.saveComponent(TrackingFetcherManager.config("f")); + Lease first = manager.acquire("f"); + Lease second = manager.acquire("f"); + assertSame(first.get(), second.get()); + TrackedFetcher old = (TrackedFetcher) first.get(); + + manager.saveComponent(TrackingFetcherManager.config("f")); + try (Lease replacement = manager.acquire("f")) { + assertNotSame(old, replacement.get(), "an update must not hand out the retired instance"); + } + assertEquals(0, old.getCloses()); + + first.close(); + // a double close must not release the other holder's lease + first.close(); + assertEquals(0, old.getCloses()); + + second.close(); + assertEquals(1, old.getCloses()); + assertEquals(0, ((TrackedFetcher) manager.getBuilt().get(1)).getCloses()); + } + + @Test + public void testCloseRetiresEverything() throws Exception { + TrackingFetcherManager manager = new TrackingFetcherManager("configured"); + manager.saveComponent(TrackingFetcherManager.config("runtime")); + manager.acquire("configured").close(); + Lease held = manager.acquire("runtime"); + TrackedFetcher configured = manager.getBuilt().get(0); + TrackedFetcher runtime = (TrackedFetcher) held.get(); + + manager.close(); + assertEquals(1, configured.getCloses()); + assertEquals(0, runtime.getCloses(), "closed under an in-flight task"); + assertThrows(IllegalStateException.class, () -> manager.acquire("configured")); + + held.close(); + assertEquals(1, runtime.getCloses()); + } + + @Test + public void testConcurrentAcquireBuildsOnce() throws Exception { + TrackingFetcherManager manager = new TrackingFetcherManager("f"); + int threads = 16; + ExecutorService executor = Executors.newFixedThreadPool(threads); + CountDownLatch start = new CountDownLatch(1); + List>> futures = new ArrayList<>(); + try { + for (int i = 0; i < threads; i++) { + futures.add(executor.submit(() -> { + start.await(); + return manager.acquire("f"); + })); + } + start.countDown(); + List> leases = new ArrayList<>(); + for (Future> f : futures) { + leases.add(f.get(10, TimeUnit.SECONDS)); + } + assertEquals(1, manager.getBuilt().size()); + manager.close(); + for (int i = 0; i < leases.size(); i++) { + assertEquals(0, manager.getBuilt().get(0).getCloses()); + leases.get(i).close(); + } + assertEquals(1, manager.getBuilt().get(0).getCloses()); + } finally { + executor.shutdownNow(); + } + } + + @Test + public void testRetireRacingAcquireClosesEachInstanceOnce() throws Exception { + TrackingFetcherManager manager = new TrackingFetcherManager(); + manager.saveComponent(TrackingFetcherManager.config("f")); + ExecutorService executor = Executors.newFixedThreadPool(4); + CountDownLatch start = new CountDownLatch(1); + List> futures = new ArrayList<>(); + try { + for (int i = 0; i < 3; i++) { + futures.add(executor.submit(() -> { + start.await(); + for (int j = 0; j < 2000; j++) { + try (Lease lease = manager.acquire("f")) { + // throws if handed an instance that was already closed + lease.get().fetch("k", null, null); + } + } + return null; + })); + } + futures.add(executor.submit(() -> { + start.await(); + for (int j = 0; j < 500; j++) { + manager.saveComponent(TrackingFetcherManager.config("f")); + } + return null; + })); + start.countDown(); + for (Future f : futures) { + f.get(60, TimeUnit.SECONDS); + } + } finally { + executor.shutdownNow(); + } + manager.close(); + for (TrackedFetcher fetcher : manager.getBuilt()) { + assertEquals(1, fetcher.getCloses()); + } + } +} diff --git a/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/TikaPipesConfigTest.java b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/TikaPipesConfigTest.java index 668bef4ad57..b42fffbbb1c 100644 --- a/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/TikaPipesConfigTest.java +++ b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/TikaPipesConfigTest.java @@ -195,10 +195,10 @@ void testMaxIpcPayloadBytesFromJsonRejectsZero() throws Exception { @Test public void testFetchers() throws Exception { FetcherManager m = FetcherManager.load(getConfigFilePath("fetchers-config.xml")); - Fetcher f1 = m.getFetcher("fs1"); + Fetcher f1 = m.getFetcher("fs1").get(); assertEquals(Paths.get("/my/base/path1"), ((FileSystemFetcher) f1).getBasePath()); - Fetcher f2 = m.getFetcher("fs2"); + Fetcher f2 = m.getFetcher("fs2").get(); assertEquals(Paths.get("/my/base/path2"), ((FileSystemFetcher) f2).getBasePath()); } @@ -231,9 +231,9 @@ public void testNoBasePathFetchers() throws Exception { public void testEmitters() throws Exception { EmitterManager emitterManager = EmitterManager.load(getConfigFilePath("emitters-config.xml")); - Emitter em1 = emitterManager.getEmitter("em1"); + Emitter em1 = emitterManager.getEmitter("em1").get(); assertNotNull(em1); - Emitter em2 = emitterManager.getEmitter("em2"); + Emitter em2 = emitterManager.getEmitter("em2").get(); assertNotNull(em2); } diff --git a/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/TrackingFetcherManager.java b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/TrackingFetcherManager.java new file mode 100644 index 00000000000..fdf40054e1d --- /dev/null +++ b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/TrackingFetcherManager.java @@ -0,0 +1,135 @@ +/* + * 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.pipes.core; + +import java.io.Closeable; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.atomic.AtomicInteger; + +import org.pf4j.PluginManager; + +import org.apache.tika.exception.TikaException; +import org.apache.tika.io.TikaInputStream; +import org.apache.tika.metadata.Metadata; +import org.apache.tika.parser.ParseContext; +import org.apache.tika.pipes.api.fetcher.Fetcher; +import org.apache.tika.pipes.api.fetcher.FetcherFactory; +import org.apache.tika.pipes.api.fetcher.FetcherNotFoundException; +import org.apache.tika.pipes.core.config.InMemoryConfigStore; +import org.apache.tika.plugins.AbstractTikaExtension; +import org.apache.tika.plugins.ExtensionConfig; + +/** A manager of {@link TrackedFetcher}s, no plugins needed; records every instance it builds. */ +public class TrackingFetcherManager extends AbstractComponentManager { + + public static final String TYPE = "tracked"; + + private final List built = new CopyOnWriteArrayList<>(); + + private final FetcherFactory factory = new FetcherFactory() { + @Override + public String getName() { + return TYPE; + } + + @Override + public Fetcher buildExtension(ExtensionConfig config) { + TrackedFetcher fetcher = new TrackedFetcher(config); + built.add(fetcher); + return fetcher; + } + + @Override + public Class getConfigClass() { + return Object.class; + } + }; + + public TrackingFetcherManager(String... configuredIds) { + super(null, configs(configuredIds), true, new InMemoryConfigStore()); + } + + public static ExtensionConfig config(String id) { + return new ExtensionConfig(id, TYPE, "{}"); + } + + private static Map configs(String... ids) { + Map configs = new HashMap<>(); + for (String id : ids) { + configs.put(id, config(id)); + } + return configs; + } + + public List getBuilt() { + return built; + } + + @Override + protected Map getFactories(PluginManager pluginManager) { + return Map.of(TYPE, factory); + } + + @Override + protected String getConfigKey() { + return "fetchers"; + } + + @Override + protected Class getFactoryClass() { + return FetcherFactory.class; + } + + @Override + protected String getComponentName() { + return "fetcher"; + } + + @Override + protected TikaException createNotFoundException(String message) { + return new FetcherNotFoundException(message); + } + + public static class TrackedFetcher extends AbstractTikaExtension implements Fetcher, Closeable { + + private final AtomicInteger closes = new AtomicInteger(); + + TrackedFetcher(ExtensionConfig config) { + super(config); + } + + public int getCloses() { + return closes.get(); + } + + @Override + public TikaInputStream fetch(String fetchKey, Metadata metadata, ParseContext parseContext) { + if (closes.get() > 0) { + throw new IllegalStateException("fetch after close"); + } + return TikaInputStream.get(new byte[0]); + } + + @Override + public void close() { + closes.incrementAndGet(); + } + } +} diff --git a/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/server/EmitHandlerUserMetadataTest.java b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/server/EmitHandlerUserMetadataTest.java index 9afe2f9ef01..8724e74d3a5 100644 --- a/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/server/EmitHandlerUserMetadataTest.java +++ b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/server/EmitHandlerUserMetadataTest.java @@ -37,7 +37,7 @@ public class EmitHandlerUserMetadataTest { private static EmitHandler emitHandler() { // injectUserMetadata touches no instance state; the collaborators are irrelevant - return new EmitHandler(null, null, null, null, 0); + return new EmitHandler(null, null, null, null, null, 0); } @Test diff --git a/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/server/PipesWorkerLeaseTest.java b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/server/PipesWorkerLeaseTest.java new file mode 100644 index 00000000000..e58ed0ce20f --- /dev/null +++ b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/server/PipesWorkerLeaseTest.java @@ -0,0 +1,80 @@ +/* + * 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.pipes.core.server; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import org.apache.tika.io.TikaInputStream; +import org.apache.tika.metadata.Metadata; +import org.apache.tika.parser.ParseContext; +import org.apache.tika.pipes.api.FetchEmitTuple; +import org.apache.tika.pipes.api.ParseMode; +import org.apache.tika.pipes.api.PipesResult; +import org.apache.tika.pipes.api.emitter.EmitKey; +import org.apache.tika.pipes.api.fetcher.FetchKey; +import org.apache.tika.pipes.core.TrackingFetcherManager; +import org.apache.tika.pipes.core.TrackingFetcherManager.TrackedFetcher; + +/** A fetcher deleted mid-task stays open until the task that leased it ends. */ +public class PipesWorkerLeaseTest { + + private static final PipesResult PARSE_FAILED = + new PipesResult(PipesResult.RESULT_STATUS.UNSPECIFIED_CRASH, "parse failed"); + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + public void testFetcherClosedOnlyAfterTask(boolean parseThrows) throws Exception { + TrackingFetcherManager manager = new TrackingFetcherManager(); + manager.saveComponent(TrackingFetcherManager.config("f")); + TaskLeases leases = new TaskLeases(manager, null); + int[] closesDuringParse = {-1}; + ParseHandler parseHandler = new ParseHandler(null, null, null, null, null, null, + ParseMode.RMETA) { + @Override + PipesWorker.ParseDataOrPipesResult parseWithStream(FetchEmitTuple t, + TikaInputStream stream, Metadata metadata, ParseContext ctx) { + try { + manager.deleteComponent("f"); + } catch (Exception e) { + throw new RuntimeException(e); + } + closesDuringParse[0] = ((TrackedFetcher) manager.getBuilt().get(0)).getCloses(); + if (parseThrows) { + throw new IllegalStateException("parser blew up"); + } + return new PipesWorker.ParseDataOrPipesResult(null, PARSE_FAILED); + } + }; + FetchEmitTuple tuple = new FetchEmitTuple("id", new FetchKey("f", "k"), + new EmitKey("e", "k"), new Metadata(), new ParseContext(), + FetchEmitTuple.ON_PARSE_EXCEPTION.EMIT); + PipesWorker worker = new PipesWorker(tuple, new ParseContext(), null, leases, + new FetchHandler(null, leases), parseHandler, null, null, ParseMode.RMETA); + + if (parseThrows) { + assertThrows(IllegalStateException.class, worker::call); + } else { + assertEquals(PARSE_FAILED.status(), worker.call().status()); + } + assertEquals(0, closesDuringParse[0], "closed while the task still held it"); + assertEquals(1, ((TrackedFetcher) manager.getBuilt().get(0)).getCloses()); + } +} diff --git a/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/server/PipesWorkerUnpackCleanupTest.java b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/server/PipesWorkerUnpackCleanupTest.java index cb9b3eb9923..d2cdd236cf5 100644 --- a/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/server/PipesWorkerUnpackCleanupTest.java +++ b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/server/PipesWorkerUnpackCleanupTest.java @@ -51,7 +51,7 @@ public class PipesWorkerUnpackCleanupTest { @ValueSource(booleans = {true, false}) public void testFetchFailure(boolean frictionless) throws Exception { ParseContext context = unpackContext(frictionless); - FetchHandler fetchHandler = new FetchHandler(null) { + FetchHandler fetchHandler = new FetchHandler(null, null) { @Override public TisOrResult fetch(FetchEmitTuple t, Metadata metadata, ParseContext ctx) { return new TisOrResult(null, FETCH_FAILED); @@ -66,7 +66,7 @@ public TisOrResult fetch(FetchEmitTuple t, Metadata metadata, ParseContext ctx) @ValueSource(booleans = {true, false}) public void testParseFailure(boolean frictionless) throws Exception { ParseContext context = unpackContext(frictionless); - FetchHandler fetchHandler = new FetchHandler(null) { + FetchHandler fetchHandler = new FetchHandler(null, null) { @Override public TisOrResult fetch(FetchEmitTuple t, Metadata metadata, ParseContext ctx) { return new TisOrResult(TikaInputStream.get(new byte[0]), null); @@ -103,7 +103,8 @@ private static PipesWorker worker(ParseContext context, FetchHandler fetchHandle FetchEmitTuple tuple = new FetchEmitTuple("id", new FetchKey("fsf", "missing.txt"), new EmitKey("fse", "missing.txt"), new Metadata(), context, FetchEmitTuple.ON_PARSE_EXCEPTION.EMIT); - return new PipesWorker(tuple, context, null, null, fetchHandler, parseHandler, null, + return new PipesWorker(tuple, context, null, new TaskLeases(null, null), fetchHandler, + parseHandler, null, null, ParseMode.UNPACK); } diff --git a/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/core/PluginManagerTest.java b/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/core/PluginManagerTest.java index 738c3c476cc..3ad5ad5f6ad 100644 --- a/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/core/PluginManagerTest.java +++ b/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/core/PluginManagerTest.java @@ -37,7 +37,7 @@ public void testBasic(@TempDir Path tmpDir) throws Exception { TikaPluginManager tikaPluginManager = TikaPluginManager.load(tikaJsonConfig); FetcherManager fetcherManager = FetcherManager.load(tikaPluginManager, tikaJsonConfig); assertEquals(1, fetcherManager.getSupported().size()); - Fetcher f = fetcherManager.getFetcher(); + Fetcher f = fetcherManager.getFetcher().get(); assertEquals("fsf", f.getExtensionConfig().id()); assertEquals("org.apache.tika.pipes.fetcher.fs.FileSystemFetcher", f.getClass().getName()); } diff --git a/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/core/TikaPipesConfigTest.java b/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/core/TikaPipesConfigTest.java index a59335f04fb..a084befe83e 100644 --- a/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/core/TikaPipesConfigTest.java +++ b/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/core/TikaPipesConfigTest.java @@ -22,10 +22,10 @@ public class TikaPipesConfigTest { @Test public void testFetchers() throws Exception { FetcherManager m = FetcherManager.load(getConfigFilePath("fetchers-config.xml")); - Fetcher f1 = m.getFetcher("fs1"); + Fetcher f1 = m.getFetcher("fs1").get(); assertEquals(Paths.get("/my/base/path1"), ((FileSystemFetcher) f1).getBasePath()); - Fetcher f2 = m.getFetcher("fs2"); + Fetcher f2 = m.getFetcher("fs2").get(); assertEquals(Paths.get("/my/base/path2"), ((FileSystemFetcher) f2).getBasePath()); } @@ -58,9 +58,9 @@ public void testNoBasePathFetchers() throws Exception { public void testEmitters() throws Exception { EmitterManager emitterManager = EmitterManager.load(getConfigFilePath("emitters-config.xml")); - Emitter em1 = emitterManager.getEmitter("file-system-emitter-1"); + Emitter em1 = emitterManager.getEmitter("file-system-emitter-1").get(); assertNotNull(em1); - Emitter em2 = emitterManager.getEmitter("file-system-emitter-2"); + Emitter em2 = emitterManager.getEmitter("file-system-emitter-2").get(); assertNotNull(em2); } diff --git a/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/core/emitter/EmitterManagerTest.java b/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/core/emitter/EmitterManagerTest.java index 8d8b438275f..fd16306edf8 100644 --- a/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/core/emitter/EmitterManagerTest.java +++ b/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/core/emitter/EmitterManagerTest.java @@ -83,7 +83,7 @@ public void testEmptyConfig(@TempDir Path tmpDir) throws Exception { // Try to get an emitter when none are configured EmitterNotFoundException exception = assertThrows(EmitterNotFoundException.class, () -> { - emitterManager.getEmitter("any-id"); + emitterManager.getEmitter("any-id").get(); }); assertTrue(exception.getMessage().contains("any-id")); @@ -126,7 +126,7 @@ public void testLazyInstantiation(@TempDir Path tmpDir) throws Exception { assertEquals(2, emitterManager.getSupported().size()); // Request only fse1 - only it should be instantiated - Emitter emitter1 = emitterManager.getEmitter("fse1"); + Emitter emitter1 = emitterManager.getEmitter("fse1").get(); assertNotNull(emitter1); assertEquals("fse1", emitter1.getExtensionConfig().id()); @@ -134,7 +134,7 @@ public void testLazyInstantiation(@TempDir Path tmpDir) throws Exception { assertTrue(emitterManager.getSupported().contains("fse2")); // Now request fse2 - Emitter emitter2 = emitterManager.getEmitter("fse2"); + Emitter emitter2 = emitterManager.getEmitter("fse2").get(); assertNotNull(emitter2); assertEquals("fse2", emitter2.getExtensionConfig().id()); } @@ -148,9 +148,9 @@ public void testCaching(@TempDir Path tmpDir) throws Exception { EmitterManager emitterManager = EmitterManager.load(pluginManager, tikaJsonConfig); // Get the same emitter multiple times - Emitter emitter1 = emitterManager.getEmitter("fse"); - Emitter emitter2 = emitterManager.getEmitter("fse"); - Emitter emitter3 = emitterManager.getEmitter("fse"); + Emitter emitter1 = emitterManager.getEmitter("fse").get(); + Emitter emitter2 = emitterManager.getEmitter("fse").get(); + Emitter emitter3 = emitterManager.getEmitter("fse").get(); // Should be the exact same instance (reference equality) assertSame(emitter1, emitter2); @@ -179,7 +179,7 @@ public void testThreadSafety(@TempDir Path tmpDir) throws Exception { startLatch.await(); // All threads try to get the emitter at once - return emitterManager.getEmitter("fse"); + return emitterManager.getEmitter("fse").get(); } finally { doneLatch.countDown(); } @@ -216,7 +216,7 @@ public void testUnknownEmitterId(@TempDir Path tmpDir) throws Exception { EmitterManager emitterManager = EmitterManager.load(pluginManager, tikaJsonConfig); EmitterNotFoundException exception = assertThrows(EmitterNotFoundException.class, () -> { - emitterManager.getEmitter("non-existent-emitter"); + emitterManager.getEmitter("non-existent-emitter").get(); }); assertTrue(exception.getMessage().contains("non-existent-emitter")); @@ -299,7 +299,7 @@ public void testGetSingleEmitter(@TempDir Path tmpDir) throws Exception { EmitterManager emitterManager = EmitterManager.load(pluginManager, tikaJsonConfig); // When only one emitter exists, no-arg getEmitter() should work - Emitter emitter = emitterManager.getEmitter(); + Emitter emitter = emitterManager.getEmitter().get(); assertNotNull(emitter); assertEquals("fse", emitter.getExtensionConfig().id()); } @@ -337,7 +337,7 @@ public void testGetSingleEmitterWithMultipleConfigured(@TempDir Path tmpDir) thr // When multiple emitters exist, no-arg getEmitter() should fail IllegalArgumentException exception = assertThrows(IllegalArgumentException.class, () -> { - emitterManager.getEmitter(); + emitterManager.getEmitter().get(); }); assertTrue(exception.getMessage().contains("exactly 1")); @@ -372,7 +372,7 @@ public void testSaveEmitter(@TempDir Path tmpDir) throws Exception { assertTrue(emitterManager.getSupported().contains("fse2")); // Emitter should be lazily instantiated when requested - Emitter emitter2 = emitterManager.getEmitter("fse2"); + Emitter emitter2 = emitterManager.getEmitter("fse2").get(); assertNotNull(emitter2); assertEquals("fse2", emitter2.getExtensionConfig().id()); } @@ -385,21 +385,21 @@ public void testSaveEmitterDuplicate(@TempDir Path tmpDir) throws Exception { EmitterManager emitterManager = EmitterManager.load(pluginManager, tikaJsonConfig, true); emitterManager.saveEmitter(fileSystemEmitter("fse2", tmpDir.resolve("output2"))); - Emitter originalEmitter = emitterManager.getEmitter("fse2"); + Emitter originalEmitter = emitterManager.getEmitter("fse2").get(); emitterManager.saveEmitter(fileSystemEmitter("fse2", tmpDir.resolve("output3"))); assertEquals(2, emitterManager.getSupported().size()); // the cached instance was dropped - assertTrue(originalEmitter != emitterManager.getEmitter("fse2"), + assertTrue(originalEmitter != emitterManager.getEmitter("fse2").get(), "Updated emitter should be a new instance"); // one from the config file is not the caller's to replace or delete - Emitter configured = emitterManager.getEmitter("fse"); + Emitter configured = emitterManager.getEmitter("fse").get(); assertThrows(TikaConfigException.class, () -> emitterManager.saveEmitter(fileSystemEmitter("fse", tmpDir.resolve("output2")))); assertThrows(TikaConfigException.class, () -> emitterManager.deleteEmitter("fse")); - assertSame(configured, emitterManager.getEmitter("fse")); + assertSame(configured, emitterManager.getEmitter("fse").get()); } @Test @@ -460,7 +460,7 @@ public void testSaveEmitterLazyInstantiation(@TempDir Path tmpDir) throws Except assertEquals(5, emitterManager.getSupported().size()); // Request only fse3 - only it should be instantiated - Emitter emitter3 = emitterManager.getEmitter("fse3"); + Emitter emitter3 = emitterManager.getEmitter("fse3").get(); assertNotNull(emitter3); assertEquals("fse3", emitter3.getExtensionConfig().id()); diff --git a/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/core/fetcher/FetcherManagerTest.java b/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/core/fetcher/FetcherManagerTest.java index 6fcd6391a13..441d012f39c 100644 --- a/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/core/fetcher/FetcherManagerTest.java +++ b/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/core/fetcher/FetcherManagerTest.java @@ -86,7 +86,7 @@ public void testEmptyConfig(@TempDir Path tmpDir) throws Exception { // Try to get a fetcher when none are configured FetcherNotFoundException exception = assertThrows(FetcherNotFoundException.class, () -> { - fetcherManager.getFetcher("any-id"); + fetcherManager.getFetcher("any-id").get(); }); assertTrue(exception.getMessage().contains("any-id")); @@ -127,7 +127,7 @@ public void testLazyInstantiation(@TempDir Path tmpDir) throws Exception { assertEquals(2, fetcherManager.getSupported().size()); // Request only fsf1 - only it should be instantiated - Fetcher fetcher1 = fetcherManager.getFetcher("fsf1"); + Fetcher fetcher1 = fetcherManager.getFetcher("fsf1").get(); assertNotNull(fetcher1); assertEquals("fsf1", fetcher1.getExtensionConfig().id()); @@ -135,7 +135,7 @@ public void testLazyInstantiation(@TempDir Path tmpDir) throws Exception { assertTrue(fetcherManager.getSupported().contains("fsf2")); // Now request fsf2 - Fetcher fetcher2 = fetcherManager.getFetcher("fsf2"); + Fetcher fetcher2 = fetcherManager.getFetcher("fsf2").get(); assertNotNull(fetcher2); assertEquals("fsf2", fetcher2.getExtensionConfig().id()); } @@ -149,9 +149,9 @@ public void testCaching(@TempDir Path tmpDir) throws Exception { FetcherManager fetcherManager = FetcherManager.load(pluginManager, tikaJsonConfig); // Get the same fetcher multiple times - Fetcher fetcher1 = fetcherManager.getFetcher("fsf"); - Fetcher fetcher2 = fetcherManager.getFetcher("fsf"); - Fetcher fetcher3 = fetcherManager.getFetcher("fsf"); + Fetcher fetcher1 = fetcherManager.getFetcher("fsf").get(); + Fetcher fetcher2 = fetcherManager.getFetcher("fsf").get(); + Fetcher fetcher3 = fetcherManager.getFetcher("fsf").get(); // Should be the exact same instance (reference equality) assertSame(fetcher1, fetcher2); @@ -180,7 +180,7 @@ public void testThreadSafety(@TempDir Path tmpDir) throws Exception { startLatch.await(); // All threads try to get the fetcher at once - return fetcherManager.getFetcher("fsf"); + return fetcherManager.getFetcher("fsf").get(); } finally { doneLatch.countDown(); } @@ -217,7 +217,7 @@ public void testUnknownFetcherId(@TempDir Path tmpDir) throws Exception { FetcherManager fetcherManager = FetcherManager.load(pluginManager, tikaJsonConfig); FetcherNotFoundException exception = assertThrows(FetcherNotFoundException.class, () -> { - fetcherManager.getFetcher("non-existent-fetcher"); + fetcherManager.getFetcher("non-existent-fetcher").get(); }); assertTrue(exception.getMessage().contains("non-existent-fetcher")); @@ -298,7 +298,7 @@ public void testGetSingleFetcher(@TempDir Path tmpDir) throws Exception { FetcherManager fetcherManager = FetcherManager.load(pluginManager, tikaJsonConfig); // When only one fetcher exists, no-arg getFetcher() should work - Fetcher fetcher = fetcherManager.getFetcher(); + Fetcher fetcher = fetcherManager.getFetcher().get(); assertNotNull(fetcher); assertEquals("fsf", fetcher.getExtensionConfig().id()); } @@ -334,7 +334,7 @@ public void testGetSingleFetcherWithMultipleConfigured(@TempDir Path tmpDir) thr // When multiple fetchers exist, no-arg getFetcher() should fail IllegalArgumentException exception = assertThrows(IllegalArgumentException.class, () -> { - fetcherManager.getFetcher(); + fetcherManager.getFetcher().get(); }); assertTrue(exception.getMessage().contains("requires exactly 1")); @@ -366,7 +366,7 @@ public void testSaveFetcher(@TempDir Path tmpDir) throws Exception { assertTrue(fetcherManager.getSupported().contains("fsf2")); // Fetcher should be lazily instantiated when requested - Fetcher fetcher2 = fetcherManager.getFetcher("fsf2"); + Fetcher fetcher2 = fetcherManager.getFetcher("fsf2").get(); assertNotNull(fetcher2); assertEquals("fsf2", fetcher2.getExtensionConfig().id()); } @@ -379,13 +379,13 @@ public void testSaveFetcherDuplicate(@TempDir Path tmpDir) throws Exception { FetcherManager fetcherManager = FetcherManager.load(pluginManager, tikaJsonConfig, true); fetcherManager.saveFetcher(fileSystemFetcher("fsf2", tmpDir.resolve("path2"))); - Fetcher originalFetcher = fetcherManager.getFetcher("fsf2"); + Fetcher originalFetcher = fetcherManager.getFetcher("fsf2").get(); fetcherManager.saveFetcher(fileSystemFetcher("fsf2", tmpDir.resolve("path3"))); assertEquals(2, fetcherManager.getSupported().size()); // the cached instance was dropped - assertTrue(originalFetcher != fetcherManager.getFetcher("fsf2"), + assertTrue(originalFetcher != fetcherManager.getFetcher("fsf2").get(), "Updated fetcher should be a new instance"); } @@ -396,13 +396,13 @@ public void testConfiguredFetcherCannotBeReplacedOrDeleted(@TempDir Path tmpDir) TikaPluginManager pluginManager = TikaPluginManager.load(tikaJsonConfig); FetcherManager fetcherManager = FetcherManager.load(pluginManager, tikaJsonConfig, true); - Fetcher configured = fetcherManager.getFetcher("fsf"); + Fetcher configured = fetcherManager.getFetcher("fsf").get(); TikaConfigException e = assertThrows(TikaConfigException.class, () -> fetcherManager.saveFetcher(fileSystemFetcher("fsf", tmpDir.resolve("path2")))); assertTrue(e.getMessage().contains("defined in the server config"), e.getMessage()); assertThrows(TikaConfigException.class, () -> fetcherManager.deleteFetcher("fsf")); - assertSame(configured, fetcherManager.getFetcher("fsf")); + assertSame(configured, fetcherManager.getFetcher("fsf").get()); } @Test @@ -500,7 +500,7 @@ public void testSaveFetcherLazyInstantiation(@TempDir Path tmpDir) throws Except assertEquals(5, fetcherManager.getSupported().size()); // Request only fsf3 - only it should be instantiated - Fetcher fetcher3 = fetcherManager.getFetcher("fsf3"); + Fetcher fetcher3 = fetcherManager.getFetcher("fsf3").get(); assertNotNull(fetcher3); assertEquals("fsf3", fetcher3.getExtensionConfig().id()); diff --git a/tika-pipes/tika-pipes-plugins/tika-pipes-http/src/test/java/org/apache/tika/pipes/fetcher/http/HttpFetcherTest.java b/tika-pipes/tika-pipes-plugins/tika-pipes-http/src/test/java/org/apache/tika/pipes/fetcher/http/HttpFetcherTest.java index fde8f185eb1..b6bda58bc7d 100644 --- a/tika-pipes/tika-pipes-plugins/tika-pipes-http/src/test/java/org/apache/tika/pipes/fetcher/http/HttpFetcherTest.java +++ b/tika-pipes/tika-pipes-plugins/tika-pipes-http/src/test/java/org/apache/tika/pipes/fetcher/http/HttpFetcherTest.java @@ -267,7 +267,7 @@ public void testRedirect() throws Exception { String url = "https://t.co/cvfkWAEIxw?amp=1"; ByteArrayOutputStream bos = new ByteArrayOutputStream(); Metadata metadata = new Metadata(); - HttpFetcher httpFetcher = (HttpFetcher) getFetcherManager("tika-config-http.json").getFetcher("http"); + HttpFetcher httpFetcher = (HttpFetcher) getFetcherManager("tika-config-http.json").getFetcher("http").get(); try (InputStream is = httpFetcher.fetch(url, metadata, new ParseContext())) { IOUtils.copy(is, bos); } @@ -281,7 +281,7 @@ public void testRange() throws Exception { long start = 969596307; long end = start + 1408 - 1; Metadata metadata = new Metadata(); - HttpFetcher httpFetcher = (HttpFetcher) getFetcherManager("tika-config-http.json").getFetcher("http"); + HttpFetcher httpFetcher = (HttpFetcher) getFetcherManager("tika-config-http.json").getFetcher("http").get(); try (TemporaryResources tmp = new TemporaryResources()) { Path tmpPath = tmp.createTempFile(metadata); try (InputStream is = httpFetcher.fetch(url, start, end, metadata)) { @@ -304,7 +304,7 @@ FetcherManager getFetcherManager(String path) throws Exception { @Test public void testVerifySslReachesClientFactory() throws Exception { HttpFetcher fetcher = (HttpFetcher) getFetcherManager("tika-config-http.json") - .getFetcher("http-fetcher-1"); + .getFetcher("http-fetcher-1").get(); assertTrue(fetcher.getHttpFetcherConfig().isVerifySsl(), "config default must be verify-on"); assertTrue(new HttpFetcherConfig().isVerifySsl(), "bare config default must be verify-on"); @@ -322,7 +322,7 @@ public void testVerifySslReachesClientFactory() throws Exception { } private void mockClientResponse(final HttpResponse response) throws Exception { - httpFetcher = (HttpFetcher) getFetcherManager("tika-config-http.json").getFetcher("http-fetcher-1"); + httpFetcher = (HttpFetcher) getFetcherManager("tika-config-http.json").getFetcher("http-fetcher-1").get(); final HttpClient httpClient = Mockito.mock(HttpClient.class); final HttpClientFactory clientFactory = Mockito.mock(HttpClientFactory.class);