This is an automated email from the ASF dual-hosted git repository.
tballison pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/tika.git
The following commit(s) were added to refs/heads/main by this push:
new 143a15ca6b TIKA-5001: step 2, fetcher/emitter leases and close (#3440)
143a15ca6b is described below
commit 143a15ca6b38abe7e46656b9c28dee589018d1d9
Author: Tim Allison <[email protected]>
AuthorDate: Wed Oct 28 08:52:42 2026 -0400
TIKA-5001: step 2, fetcher/emitter leases and close (#3440)
---
CHANGES.txt | 11 +-
.../ROOT/pages/pipes/plugins/writing-a-plugin.adoc | 10 +-
.../ROOT/pages/pipes/shared-server-mode.adoc | 3 +-
.../apache/tika/pipes/grpc/TikaGrpcServerImpl.java | 19 ++-
.../elasticsearch/tests/ElasticsearchTest.java | 4 +-
.../pipes/opensearch/tests/OpenSearchTest.java | 2 +-
.../tika/pipes/core/AbstractComponentManager.java | 162 +++++++++++++++-----
.../java/org/apache/tika/pipes/core/Lease.java | 53 +++++++
.../apache/tika/pipes/core/async/AsyncEmitter.java | 7 +-
.../tika/pipes/core/async/AsyncProcessor.java | 6 +-
.../tika/pipes/core/emitter/EmitterManager.java | 20 +--
.../core/extractor/EmittingUnpackHandler.java | 22 ++-
.../tika/pipes/core/fetcher/FetcherManager.java | 20 +--
.../tika/pipes/core/server/ConnectionHandler.java | 8 +-
.../apache/tika/pipes/core/server/EmitHandler.java | 7 +-
.../tika/pipes/core/server/FetchHandler.java | 6 +-
.../apache/tika/pipes/core/server/PipesServer.java | 14 +-
.../apache/tika/pipes/core/server/PipesWorker.java | 16 +-
.../pipes/core/server/SharedServerResources.java | 7 +-
.../apache/tika/pipes/core/server/TaskLeases.java | 71 +++++++++
.../apache/tika/pipes/core/ComponentLeaseTest.java | 164 +++++++++++++++++++++
.../tika/pipes/core/TikaPipesConfigTest.java | 8 +-
.../tika/pipes/core/TrackingFetcherManager.java | 135 +++++++++++++++++
.../core/server/EmitHandlerUserMetadataTest.java | 2 +-
.../pipes/core/server/PipesWorkerLeaseTest.java | 80 ++++++++++
.../core/server/PipesWorkerUnpackCleanupTest.java | 7 +-
.../apache/tika/pipes/core/PluginManagerTest.java | 2 +-
.../tika/pipes/core/TikaPipesConfigTest.java | 8 +-
.../pipes/core/emitter/EmitterManagerTest.java | 32 ++--
.../pipes/core/fetcher/FetcherManagerTest.java | 32 ++--
.../tika/pipes/fetcher/http/HttpFetcherTest.java | 8 +-
31 files changed, 787 insertions(+), 159 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 9826420af3..5086616d58 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 deca84bea7..cf22593410 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
<<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
diff --git a/docs/modules/ROOT/pages/pipes/shared-server-mode.adoc
b/docs/modules/ROOT/pages/pipes/shared-server-mode.adoc
index a4957c7695..3d5f7e587d 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 5fb248918c..cc9f731460 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.FetchKey;
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 @@ class TikaGrpcServerImpl extends TikaGrpc.TikaImplBase {
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 @@ class TikaGrpcServerImpl extends TikaGrpc.TikaImplBase {
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 @@ class TikaGrpcServerImpl extends TikaGrpc.TikaImplBase {
// 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());
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 8f1b6f6e70..f80ecc9785 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 class ElasticsearchTest {
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 class ElasticsearchTest {
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 1d646557fb..0eb0a2a7ab 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 class OpenSearchTest {
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 a6e5bc1b9e..4e576e6fcf 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 @@ import org.apache.tika.plugins.TikaExtensionFactory;
/**
* 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 @@ public abstract class AbstractComponentManager<T extends
TikaExtension,
}
/**
- * 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 @@ public abstract class AbstractComponentManager<T extends
TikaExtension,
/**
* 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,9 +387,7 @@ public abstract class AbstractComponentManager<T extends
TikaExtension,
". 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 abstract class AbstractComponentManager<T extends
TikaExtension,
// 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 abstract class AbstractComponentManager<T extends
TikaExtension,
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 abstract class AbstractComponentManager<T extends
TikaExtension,
}
/**
- * 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(
@@ -455,8 +482,65 @@ public abstract class AbstractComponentManager<T extends
TikaExtension,
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);
+ }
+ }
+ }
}
}
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 0000000000..b3031ef400
--- /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 <T> the component type
+ */
+public final class Lease<T> 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 6e1f9b03c7..bccf6156f2 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.slf4j.LoggerFactory;
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 @@ public class AsyncEmitter implements Callable<Integer> {
int emitted = 0;
LOG.debug("about to emit {} files, {} estimated bytes", size,
estimatedSize);
for (Map.Entry<String, List<EmitData>> e : map.entrySet()) {
- Emitter emitter = null;
- try {
- emitter = emitterManager.getEmitter(e.getKey());
+ try (Lease<Emitter> 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 6d2ab6d27a..d260c7d044 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 @@ public class AsyncProcessor implements Closeable {
}
}
- 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 class AsyncProcessor implements Closeable {
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 360eb9bb7d..18746e9d03 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.Emitter;
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 @@ public class EmitterManager extends
AbstractComponentManager<Emitter, EmitterFac
}
/**
- * 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<Emitter> 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<Emitter> 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.
* <p>
* 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 65cef5fac1..c76459f22a 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.FetchEmitTuple;
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 class EmittingUnpackHandler extends
AbstractUnpackHandler {
}
@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 0880d383ef..8b2ed5ef4f 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.Fetcher;
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 @@ public class FetcherManager extends
AbstractComponentManager<Fetcher, FetcherFac
}
/**
- * 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<Fetcher> 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<Fetcher> 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.
* <p>
* 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 2b16ecf6af..bfbb0c53e3 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 @@ public class ConnectionHandler implements Runnable,
Closeable {
private PipesWorker createPipesWorker(ArrayBlockingQueue<Metadata>
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 @@ public class ConnectionHandler implements Runnable,
Closeable {
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 e31fe8cce4..48a9ca7a53 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 @@ class EmitHandler {
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 418054719e..5f96cb22fa 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 @@ class FetchHandler {
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 65bb9eeeac..84beba799a 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 class PipesServer implements AutoCloseable {
private PipesWorker getPipesWorker(ArrayBlockingQueue<Metadata>
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 @@ public class PipesServer implements AutoCloseable {
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 ec9415234c..e6c9e54047 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.EmitKey;
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<PipesResult> {
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<PipesResult> {
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 @@ class PipesWorker implements Callable<PipesResult> {
LOG.warn("problem closing unpack handler", e);
}
}
+ // after the unpack handler, which may still be emitting
+ leases.close();
}
}
@@ -223,7 +224,7 @@ class PipesWorker implements Callable<PipesResult> {
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 @@ class PipesWorker implements Callable<PipesResult> {
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 @@ class PipesWorker implements Callable<PipesResult> {
} 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 314da6d960..36dac29b4c 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 class SharedServerResources implements Closeable {
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 0000000000..70b507e0d0
--- /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<Fetcher, ?> fetcherManager;
+ private final AbstractComponentManager<Emitter, ?> emitterManager;
+ private final Map<String, Lease<Fetcher>> fetchers = new HashMap<>();
+ private final Map<String, Lease<Emitter>> emitters = new HashMap<>();
+
+ TaskLeases(AbstractComponentManager<Fetcher, ?> fetcherManager,
+ AbstractComponentManager<Emitter, ?> emitterManager) {
+ this.fetcherManager = fetcherManager;
+ this.emitterManager = emitterManager;
+ }
+
+ synchronized Fetcher fetcher(String id) throws IOException, TikaException {
+ Lease<Fetcher> 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<Emitter> 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 0000000000..c9722fdb5d
--- /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<Fetcher> first = manager.acquire("f");
+ Lease<Fetcher> second = manager.acquire("f");
+ assertSame(first.get(), second.get());
+ TrackedFetcher old = (TrackedFetcher) first.get();
+
+ manager.saveComponent(TrackingFetcherManager.config("f"));
+ try (Lease<Fetcher> 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<Fetcher> 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<Future<Lease<Fetcher>>> futures = new ArrayList<>();
+ try {
+ for (int i = 0; i < threads; i++) {
+ futures.add(executor.submit(() -> {
+ start.await();
+ return manager.acquire("f");
+ }));
+ }
+ start.countDown();
+ List<Lease<Fetcher>> leases = new ArrayList<>();
+ for (Future<Lease<Fetcher>> 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<Future<?>> 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<Fetcher> 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 668bef4ad5..b42fffbbb1 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 @@ public class TikaPipesConfigTest extends TikaTest {
@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 class TikaPipesConfigTest extends TikaTest {
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 0000000000..fdf40054e1
--- /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<Fetcher,
FetcherFactory> {
+
+ public static final String TYPE = "tracked";
+
+ private final List<TrackedFetcher> 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<String, ExtensionConfig> configs(String... ids) {
+ Map<String, ExtensionConfig> configs = new HashMap<>();
+ for (String id : ids) {
+ configs.put(id, config(id));
+ }
+ return configs;
+ }
+
+ public List<TrackedFetcher> getBuilt() {
+ return built;
+ }
+
+ @Override
+ protected Map<String, FetcherFactory> getFactories(PluginManager
pluginManager) {
+ return Map.of(TYPE, factory);
+ }
+
+ @Override
+ protected String getConfigKey() {
+ return "fetchers";
+ }
+
+ @Override
+ protected Class<FetcherFactory> 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 9afe2f9ef0..8724e74d3a 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 0000000000..e58ed0ce20
--- /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 cb9b3eb992..d2cdd236cf 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 class PipesWorkerUnpackCleanupTest {
@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 @@ public class PipesWorkerUnpackCleanupTest {
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 738c3c476c..3ad5ad5f6a 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 class PluginManagerTest {
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 a59335f04f..a084befe83 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 class TikaPipesConfigTest {
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 8d8b438275..fd16306edf 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 class EmitterManagerTest {
// 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 class EmitterManagerTest {
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 class EmitterManagerTest {
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 class EmitterManagerTest {
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 class EmitterManagerTest {
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 class EmitterManagerTest {
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 class EmitterManagerTest {
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 class EmitterManagerTest {
// 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 class EmitterManagerTest {
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 class EmitterManagerTest {
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 class EmitterManagerTest {
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 6fcd6391a1..441d012f39 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 class FetcherManagerTest {
// 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 class FetcherManagerTest {
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 class FetcherManagerTest {
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 class FetcherManagerTest {
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 class FetcherManagerTest {
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 class FetcherManagerTest {
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 class FetcherManagerTest {
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 class FetcherManagerTest {
// 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 class FetcherManagerTest {
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 class FetcherManagerTest {
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 class FetcherManagerTest {
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 class FetcherManagerTest {
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 fde8f185eb..b6bda58bc7 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 @@ class HttpFetcherTest extends TikaTest {
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 @@ class HttpFetcherTest extends TikaTest {
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 @@ class HttpFetcherTest extends TikaTest {
@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 @@ class HttpFetcherTest extends TikaTest {
}
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);
