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 92884773a2 TIKA-5001: step 1 (#3436)
92884773a2 is described below
commit 92884773a2d9fdd0952f33fab0c4a5390f3a2052
Author: Tim Allison <[email protected]>
AuthorDate: Tue Oct 27 12:08:52 2026 -0400
TIKA-5001: step 1 (#3436)
---
docs/modules/ROOT/pages/configuration/index.adoc | 6 +-
docs/modules/ROOT/pages/pipes/configuration.adoc | 2 +-
docs/modules/ROOT/pages/using-tika/grpc/index.adoc | 3 +-
.../apache/tika/pipes/grpc/TikaGrpcServerTest.java | 74 ++++++++++++++-----
.../tika/pipes/core/AbstractComponentManager.java | 50 ++++++++++---
.../tika/pipes/core/emitter/EmitterManager.java | 1 +
.../tika/pipes/core/fetcher/FetcherManager.java | 1 +
.../pipes/core/emitter/EmitterManagerTest.java | 48 ++++++-------
.../pipes/core/fetcher/FetcherManagerTest.java | 82 +++++++++++++++++-----
9 files changed, 190 insertions(+), 77 deletions(-)
diff --git a/docs/modules/ROOT/pages/configuration/index.adoc
b/docs/modules/ROOT/pages/configuration/index.adoc
index fa363ae605..9012ab191d 100644
--- a/docs/modules/ROOT/pages/configuration/index.adoc
+++ b/docs/modules/ROOT/pages/configuration/index.adoc
@@ -365,9 +365,9 @@ so a value that itself contains `${env:...}` is not
expanded a second time. Pres
at startup with the rest of it; catalog presets and per-request configuration
(`/config`
endpoints, pipes tuples) never do: a request cannot read the server's
environment.
-A reference keeps a secret out of tika-config.json, not out of every file:
with a persistent
-config store (`configStoreType` `file` or Ignite) the resolved fetcher and
emitter configurations
-are written to that store. The default in-memory store writes nothing.
+A persistent config store (`configStoreType` `file` or Ignite) holds only
fetchers and emitters
+registered at runtime (gRPC `SaveFetcher`), as the caller sent them.
Components from the config
+file, with their resolved references, are never written to it.
== Windows file paths
diff --git a/docs/modules/ROOT/pages/pipes/configuration.adoc
b/docs/modules/ROOT/pages/pipes/configuration.adoc
index c16926767b..ce63831dc0 100644
--- a/docs/modules/ROOT/pages/pipes/configuration.adoc
+++ b/docs/modules/ROOT/pages/pipes/configuration.adoc
@@ -256,7 +256,7 @@ For multi-host pipelines (e.g., shared-server clusters) you
can store fetcher/em
|`configStoreType`
|`"memory"`
-|Backend for storing fetcher/emitter configurations. `"memory"` (default) is
in-process; `"ignite"` uses Apache Ignite for shared state across nodes.
+|Backend for storing fetcher/emitter configurations registered at runtime;
those in the config file are never written to it. `"memory"` (default) is
in-process; `"file"` persists to a JSON file; `"ignite"` uses Apache Ignite for
shared state across nodes.
|`configStoreParams`
|`"{}"`
diff --git a/docs/modules/ROOT/pages/using-tika/grpc/index.adoc
b/docs/modules/ROOT/pages/using-tika/grpc/index.adoc
index ca0927e892..0642be675f 100644
--- a/docs/modules/ROOT/pages/using-tika/grpc/index.adoc
+++ b/docs/modules/ROOT/pages/using-tika/grpc/index.adoc
@@ -119,7 +119,8 @@ it is off by default. When `false`, a request carrying
either field is rejected
configuration back (`GetFetcher`, `ListFetchers`, `GetPipesIterator`). Off by
default for two
reasons: mutations change what the server can reach for all subsequent
requests (for example, a
fetcher that escapes a configured base path), and the stored configs returned
by the read RPCs may
-contain passwords, access keys and tokens. When `false`, the mutating RPCs are
rejected with
+contain passwords, access keys and tokens. Fetchers defined in the config file
cannot be
+modified or deleted this way. When `false`, the mutating RPCs are rejected with
`PERMISSION_DENIED`, and the read RPCs return only component identity — id and
class — never the
config.
|===
diff --git
a/tika-grpc/src/test/java/org/apache/tika/pipes/grpc/TikaGrpcServerTest.java
b/tika-grpc/src/test/java/org/apache/tika/pipes/grpc/TikaGrpcServerTest.java
index 8090113fff..7dcb2d04d5 100644
--- a/tika-grpc/src/test/java/org/apache/tika/pipes/grpc/TikaGrpcServerTest.java
+++ b/tika-grpc/src/test/java/org/apache/tika/pipes/grpc/TikaGrpcServerTest.java
@@ -17,6 +17,7 @@
package org.apache.tika.pipes.grpc;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;
@@ -67,6 +68,7 @@ import org.junit.jupiter.api.Nested;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestInstance;
import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.io.TempDir;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -218,7 +220,7 @@ public class TikaGrpcServerTest {
String targetFolder = new File("target").getAbsolutePath();
// create fetchers
for (int i = 0; i < NUM_FETCHERS_TO_CREATE; ++i) {
- String fetcherId = createFetcherId(i);
+ String fetcherId = "saved" + i + ".fs";
SaveFetcherReply reply =
blockingStub.saveFetcher(SaveFetcherRequest
.newBuilder()
.setFetcherId(fetcherId)
@@ -233,7 +235,7 @@ public class TikaGrpcServerTest {
}
// update fetchers
for (int i = 0; i < NUM_FETCHERS_TO_CREATE; ++i) {
- String fetcherId = createFetcherId(i);
+ String fetcherId = "saved" + i + ".fs";
SaveFetcherReply reply =
blockingStub.saveFetcher(SaveFetcherRequest
.newBuilder()
.setFetcherId(fetcherId)
@@ -256,7 +258,7 @@ public class TikaGrpcServerTest {
// get fetchers
for (int i = 0; i < NUM_FETCHERS_TO_CREATE; ++i) {
- String fetcherId = createFetcherId(i);
+ String fetcherId = "saved" + i + ".fs";
GetFetcherReply getFetcherReply =
blockingStub.getFetcher(GetFetcherRequest
.newBuilder()
.setFetcherId(fetcherId)
@@ -267,7 +269,7 @@ public class TikaGrpcServerTest {
// delete fetchers
for (int i = 0; i < NUM_FETCHERS_TO_CREATE; ++i) {
- String fetcherId = createFetcherId(i);
+ String fetcherId = "saved" + i + ".fs";
DeleteFetcherReply deleteFetcherReply =
blockingStub.deleteFetcher(DeleteFetcherRequest
.newBuilder()
.setFetcherId(fetcherId)
@@ -638,6 +640,54 @@ public class TikaGrpcServerTest {
}
}
+ /**
+ * The point of a persistent store: a fetcher saved at runtime reaches the
forked worker.
+ * Config-file fetchers stay out of the store.
+ */
+ @Test
+ public void testSavedFetcherReachesTheForkThroughFileStore(Resources
resources,
+ @TempDir Path tmp) throws Exception {
+ Path store = tmp.resolve("config-store.json");
+ ObjectNode root = (ObjectNode)
OBJECT_MAPPER.readTree(tikaConfigUnlocked.toFile());
+ ObjectNode pipes = (ObjectNode) root.get("pipes");
+ pipes.put("configStoreType", "file");
+ pipes.put("configStoreParams", OBJECT_MAPPER.writeValueAsString(
+ Map.of("path", store.toAbsolutePath().toString())));
+ Path config = tmp.resolve("tika-config-file-store.json");
+ Files.writeString(config, OBJECT_MAPPER.writeValueAsString(root),
StandardCharsets.UTF_8);
+
+ String serverName = InProcessServerBuilder.generateName();
+ Server server =
InProcessServerBuilder.forName(serverName).directExecutor()
+ .addService(newService(config)).build().start();
+ resources.register(server, Duration.ofSeconds(10));
+ ManagedChannel channel =
InProcessChannelBuilder.forName(serverName).directExecutor().build();
+ resources.register(channel, Duration.ofSeconds(10));
+ TikaGrpc.TikaBlockingStub blockingStub =
TikaGrpc.newBlockingStub(channel);
+
+ Path docs = Files.createDirectories(tmp.resolve("docs"));
+ Files.writeString(docs.resolve("saved.html"),
+ "<html><body>saved-fetcher-marker</body></html>",
StandardCharsets.UTF_8);
+ blockingStub.saveFetcher(SaveFetcherRequest.newBuilder()
+ .setFetcherId("saved.fs")
+ .setFetcherType("file-system-fetcher")
+ .setFetcherConfigJson(OBJECT_MAPPER.writeValueAsString(
+ Map.of("basePath", docs.toAbsolutePath().toString())))
+ .build());
+
+ FetchAndParseReply reply =
blockingStub.fetchAndParse(FetchAndParseRequest.newBuilder()
+ .setFetcherId("saved.fs")
+ .setFetchKey("saved.html")
+ .build());
+ assertEquals(PipesResult.RESULT_STATUS.PARSE_SUCCESS.name(),
reply.getStatus(),
+ reply.getErrorMessage());
+ assertTrue(reply.getFieldsMap().values().stream()
+ .anyMatch(v -> v.contains("saved-fetcher-marker")));
+
+ String stored = Files.readString(store, StandardCharsets.UTF_8);
+ assertTrue(stored.contains("saved.fs"), stored);
+ assertFalse(stored.contains("nick1.is.cool.super-fs"), stored);
+ }
+
@Test
public void testBiStream(Resources resources) throws Exception {
String serverName = InProcessServerBuilder.generateName();
@@ -659,20 +709,8 @@ public class TikaGrpcServerTest {
TikaGrpc.TikaBlockingStub blockingStub =
TikaGrpc.newBlockingStub(channel);
TikaGrpc.TikaStub tikaStub = TikaGrpc.newStub(channel);
- String fetcherId = createFetcherId(1);
- String targetFolder = new File("target").getAbsolutePath();
- SaveFetcherReply reply = blockingStub.saveFetcher(SaveFetcherRequest
- .newBuilder()
- .setFetcherId(fetcherId)
- .setFetcherType("file-system-fetcher")
-
.setFetcherConfigJson(OBJECT_MAPPER.writeValueAsString(ImmutableMap
- .builder()
- .put("basePath", targetFolder)
- .put("extractFileSystemMetadata", true)
- .build()))
- .build());
-
- assertEquals(fetcherId, reply.getFetcherId());
+ // from the config file: one saved at runtime is not visible to the
forked worker
+ String fetcherId = "nick1.is.cool.super-fs";
List<FetchAndParseReply> successes = Collections.synchronizedList(new
ArrayList<>());
List<FetchAndParseReply> errors = Collections.synchronizedList(new
ArrayList<>());
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 e93b49a7c9..a6e5bc1b9e 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
@@ -56,6 +56,10 @@ public abstract class AbstractComponentManager<T extends
TikaExtension,
private static final Logger LOG =
LoggerFactory.getLogger(AbstractComponentManager.class);
protected final PluginManager pluginManager;
+ // from the config file: never written to the store, so a persistent store
holds no resolved
+ // ${env:...} secrets, and never replaced or deleted at runtime
+ private final Map<String, ExtensionConfig> configured;
+ // runtime registrations only
private final ConfigStore configStore;
private final Map<String, T> componentCache = new ConcurrentHashMap<>();
private final boolean allowRuntimeModifications;
@@ -77,7 +81,14 @@ public abstract class AbstractComponentManager<T extends
TikaExtension,
} catch (Exception e) {
throw new RuntimeException("Failed to initialize ConfigStore", e);
}
- componentConfigs.forEach(configStore::put);
+ this.configured = Map.copyOf(componentConfigs);
+ for (String id : configured.keySet()) {
+ if (configStore.containsKey(id)) {
+ // most likely a copy an older version persisted; it may hold
resolved secrets
+ LOG.warn("The config store holds a {} '{}' that the config
file also defines; the "
+ + "config file's wins. Remove it from the store.",
getComponentName(), id);
+ }
+ }
this.allowRuntimeModifications = allowRuntimeModifications;
}
@@ -243,7 +254,7 @@ public abstract class AbstractComponentManager<T extends
TikaExtension,
}
// Check if config exists
- ExtensionConfig config = configStore.get(id);
+ ExtensionConfig config = lookup(id);
if (config == null) {
throw createNotFoundException(
"Can't find " + getComponentName() + " for id=" + id +
@@ -270,6 +281,19 @@ public abstract class AbstractComponentManager<T extends
TikaExtension,
}
}
+ private ExtensionConfig lookup(String id) {
+ ExtensionConfig config = configured.get(id);
+ return config != null ? config : configStore.get(id);
+ }
+
+ private void requireNotConfigured(String componentId) throws
TikaConfigException {
+ if (configured.containsKey(componentId)) {
+ throw new TikaConfigException(getComponentName().substring(0,
1).toUpperCase(Locale.ROOT)
+ + getComponentName().substring(1) + " '" + componentId
+ + "' is defined in the server config; change it there");
+ }
+ }
+
/**
* Builds a component instance from its configuration.
*/
@@ -291,6 +315,7 @@ 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.
+ * A component defined in the server config cannot be replaced.
* <p>
* This method is only available if the manager was loaded with
allowRuntimeModifications=true.
* <p>
@@ -321,6 +346,7 @@ public abstract class AbstractComponentManager<T extends
TikaExtension,
} catch (IllegalArgumentException e) {
throw new TikaConfigException(e.getMessage(), e);
}
+ requireNotConfigured(componentId);
config = new ExtensionConfig(componentId, config.name(),
config.json());
String typeName = config.name();
@@ -349,7 +375,8 @@ public abstract class AbstractComponentManager<T extends
TikaExtension,
* Clears the cached instance and removes the configuration.
*
* @param componentId the component ID to delete
- * @throws TikaConfigException if runtime modifications are not allowed or
component not found
+ * @throws TikaConfigException if runtime modifications are not allowed,
the component is not
+ * found, or it is defined in the server config
*/
public synchronized void deleteComponent(String componentId) throws
TikaConfigException {
if (!allowRuntimeModifications) {
@@ -370,6 +397,7 @@ public abstract class AbstractComponentManager<T extends
TikaExtension,
} catch (IllegalArgumentException e) {
throw new TikaConfigException(e.getMessage(), e);
}
+ requireNotConfigured(componentId);
if (!configStore.containsKey(componentId)) {
throw new TikaConfigException(
@@ -390,7 +418,7 @@ public abstract class AbstractComponentManager<T extends
TikaExtension,
* @return the component configuration, or null if not found
*/
public ExtensionConfig getComponentConfig(String componentId) {
- return configStore.get(ComponentIds.normalize(componentId));
+ return lookup(ComponentIds.normalize(componentId));
}
/**
@@ -398,7 +426,7 @@ public abstract class AbstractComponentManager<T extends
TikaExtension,
* bind them, so listing them in a "not found" message only advertises
internals.
*/
public Set<String> getSupported() {
- return configStore.keySet().stream()
+ return getAllIds().stream()
.filter(id -> !ComponentIds.isSystem(id))
.collect(Collectors.toCollection(LinkedHashSet::new));
}
@@ -408,7 +436,9 @@ public abstract class AbstractComponentManager<T extends
TikaExtension,
* anything a caller sees.
*/
public Set<String> getAllIds() {
- return configStore.keySet();
+ Set<String> ids = new LinkedHashSet<>(configured.keySet());
+ ids.addAll(configStore.keySet());
+ return ids;
}
/**
@@ -418,15 +448,15 @@ public abstract class AbstractComponentManager<T extends
TikaExtension,
* @return the single configured component
*/
public T getComponent() throws IOException, TikaException {
- if (configStore.size() != 1) {
+ Set<String> ids = getAllIds();
+ if (ids.size() != 1) {
throw new IllegalArgumentException(
"No-arg get" + getComponentName().substring(0,
1).toUpperCase(Locale.ROOT) +
getComponentName().substring(1) + "() requires exactly 1
configured " +
- getComponentName() + ". Found: " + configStore.size() +
- " (" + configStore.keySet() + ")");
+ getComponentName() + ". Found: " + ids.size() + " (" + ids
+ ")");
}
// Get the single component id and use getComponent(id) for lazy
loading
- String componentId = configStore.keySet().iterator().next();
+ String componentId = ids.iterator().next();
return getComponent(componentId);
}
}
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 256e35b7c2..360eb9bb7d 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
@@ -161,6 +161,7 @@ public class EmitterManager extends
AbstractComponentManager<Emitter, EmitterFac
* 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.
+ * An emitter defined in the server config cannot be replaced.
* <p>
* This method is only available if the EmitterManager was loaded with
* {@link #load(PluginManager, TikaJsonConfig, boolean)} with
allowRuntimeModifications=true.
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 4f4e9a199e..0880d383ef 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
@@ -160,6 +160,7 @@ public class FetcherManager extends
AbstractComponentManager<Fetcher, FetcherFac
* 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.
+ * A fetcher defined in the server config cannot be replaced.
* <p>
* This method is only available if the FetcherManager was loaded with
* {@link #load(PluginManager, TikaJsonConfig, boolean)} with
allowRuntimeModifications=true.
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 18aa5d2fbf..8d8b438275 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
@@ -384,33 +384,22 @@ public class EmitterManagerTest {
TikaPluginManager pluginManager =
TikaPluginManager.load(tikaJsonConfig);
EmitterManager emitterManager = EmitterManager.load(pluginManager,
tikaJsonConfig, true);
+ emitterManager.saveEmitter(fileSystemEmitter("fse2",
tmpDir.resolve("output2")));
+ Emitter originalEmitter = emitterManager.getEmitter("fse2");
- // Update existing emitter with new configuration
- String newConfigJson = String.format(Locale.ROOT, """
- {
- "basePath": "%s",
- "onExists": "REPLACE"
- }
- """, JsonConfigHelper.toJsonPath(tmpDir.resolve("output2")));
- ExtensionConfig updatedConfig = new ExtensionConfig("fse",
"file-system-emitter", newConfigJson);
-
- // Get original emitter instance
- Emitter originalEmitter = emitterManager.getEmitter("fse");
- assertNotNull(originalEmitter);
-
- // Update the emitter config
- emitterManager.saveEmitter(updatedConfig);
-
- // Should still only have 1 emitter
- assertEquals(1, emitterManager.getSupported().size());
- assertTrue(emitterManager.getSupported().contains("fse"));
+ emitterManager.saveEmitter(fileSystemEmitter("fse2",
tmpDir.resolve("output3")));
- // Getting the emitter again should return a NEW instance (cache
cleared)
- Emitter updatedEmitter = emitterManager.getEmitter("fse");
- assertNotNull(updatedEmitter);
-
- // Should be different instance due to re-instantiation
- assertTrue(originalEmitter != updatedEmitter, "Updated emitter should
be a new instance");
+ assertEquals(2, emitterManager.getSupported().size());
+ // the cached instance was dropped
+ assertTrue(originalEmitter != emitterManager.getEmitter("fse2"),
+ "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");
+ assertThrows(TikaConfigException.class,
+ () -> emitterManager.saveEmitter(fileSystemEmitter("fse",
tmpDir.resolve("output2"))));
+ assertThrows(TikaConfigException.class, () ->
emitterManager.deleteEmitter("fse"));
+ assertSame(configured, emitterManager.getEmitter("fse"));
}
@Test
@@ -531,4 +520,13 @@ public class EmitterManagerTest {
assertTrue(exception.getMessage().contains("Runtime modifications are
not allowed"));
}
+
+ private static ExtensionConfig fileSystemEmitter(String id, Path basePath)
{
+ return new ExtensionConfig(id, "file-system-emitter",
String.format(Locale.ROOT, """
+ {
+ "basePath": "%s",
+ "onExists": "REPLACE"
+ }
+ """, JsonConfigHelper.toJsonPath(basePath)));
+ }
}
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 595b5667a4..6fcd6391a1 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
@@ -17,6 +17,7 @@
package org.apache.tika.pipes.core.fetcher;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -28,6 +29,7 @@ import java.nio.file.Path;
import java.util.ArrayList;
import java.util.List;
import java.util.Locale;
+import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -42,6 +44,7 @@ import org.apache.tika.exception.TikaConfigException;
import org.apache.tika.pipes.api.fetcher.Fetcher;
import org.apache.tika.pipes.api.fetcher.FetcherNotFoundException;
import org.apache.tika.pipes.core.PluginsTestHelper;
+import org.apache.tika.pipes.core.config.FileBasedConfigStore;
import org.apache.tika.plugins.ExtensionConfig;
import org.apache.tika.plugins.TikaPluginManager;
import org.apache.tika.serialization.config.JsonConfigHelper;
@@ -375,30 +378,71 @@ public class FetcherManagerTest {
TikaPluginManager pluginManager =
TikaPluginManager.load(tikaJsonConfig);
FetcherManager fetcherManager = FetcherManager.load(pluginManager,
tikaJsonConfig, true);
+ fetcherManager.saveFetcher(fileSystemFetcher("fsf2",
tmpDir.resolve("path2")));
+ Fetcher originalFetcher = fetcherManager.getFetcher("fsf2");
- // Update existing fetcher with new configuration
- String newConfigJson = String.format(Locale.ROOT, """
- {"basePath": "%s"}
- """, JsonConfigHelper.toJsonPath(tmpDir.resolve("path2")));
- ExtensionConfig updatedConfig = new ExtensionConfig("fsf",
"file-system-fetcher", newConfigJson);
+ fetcherManager.saveFetcher(fileSystemFetcher("fsf2",
tmpDir.resolve("path3")));
- // Get original fetcher instance
- Fetcher originalFetcher = fetcherManager.getFetcher("fsf");
- assertNotNull(originalFetcher);
+ assertEquals(2, fetcherManager.getSupported().size());
+ // the cached instance was dropped
+ assertTrue(originalFetcher != fetcherManager.getFetcher("fsf2"),
+ "Updated fetcher should be a new instance");
+ }
- // Update the fetcher config
- fetcherManager.saveFetcher(updatedConfig);
+ @Test
+ public void testConfiguredFetcherCannotBeReplacedOrDeleted(@TempDir Path
tmpDir) throws Exception {
+ Path config = PluginsTestHelper.getFileSystemFetcherConfig(tmpDir);
+ TikaJsonConfig tikaJsonConfig = TikaJsonConfig.load(config);
+ TikaPluginManager pluginManager =
TikaPluginManager.load(tikaJsonConfig);
- // Should still only have 1 fetcher
- assertEquals(1, fetcherManager.getSupported().size());
- assertTrue(fetcherManager.getSupported().contains("fsf"));
+ FetcherManager fetcherManager = FetcherManager.load(pluginManager,
tikaJsonConfig, true);
+ Fetcher configured = fetcherManager.getFetcher("fsf");
- // Getting the fetcher again should return a NEW instance (cache
cleared)
- Fetcher updatedFetcher = fetcherManager.getFetcher("fsf");
- assertNotNull(updatedFetcher);
-
- // Should be different instance due to re-instantiation
- assertTrue(originalFetcher != updatedFetcher, "Updated fetcher should
be a new instance");
+ 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"));
+ }
+
+ @Test
+ public void testConfiguredFetcherIsNotPersisted(@TempDir Path tmpDir)
throws Exception {
+ Path config = PluginsTestHelper.getFileSystemFetcherConfig(tmpDir);
+ TikaJsonConfig tikaJsonConfig = TikaJsonConfig.load(config);
+ TikaPluginManager pluginManager =
TikaPluginManager.load(tikaJsonConfig);
+ Path storeFile = tmpDir.resolve("config-store.json");
+
+ FetcherManager fetcherManager = FetcherManager.load(pluginManager,
tikaJsonConfig, true,
+ new FileBasedConfigStore(storeFile));
+ fetcherManager.saveFetcher(fileSystemFetcher("fsf2",
tmpDir.resolve("path2")));
+
+ assertEquals(Set.of("fsf", "fsf2"), fetcherManager.getSupported());
+ assertEquals(Set.of("fsf2"), new
FileBasedConfigStore(storeFile).keySet());
+ }
+
+ @Test
+ public void testOldStoreCopyOfConfiguredFetcherIsIgnored(@TempDir Path
tmpDir) throws Exception {
+ Path config = PluginsTestHelper.getFileSystemFetcherConfig(tmpDir);
+ TikaJsonConfig tikaJsonConfig = TikaJsonConfig.load(config);
+ TikaPluginManager pluginManager =
TikaPluginManager.load(tikaJsonConfig);
+ Path storeFile = tmpDir.resolve("config-store.json");
+ // what an older version left behind: its copy of the configured
fetcher
+ FileBasedConfigStore legacy = new FileBasedConfigStore(storeFile);
+ legacy.init();
+ legacy.put("fsf", fileSystemFetcher("fsf", tmpDir.resolve("stale")));
+
+ FetcherManager fetcherManager = FetcherManager.load(pluginManager,
tikaJsonConfig, true,
+ new FileBasedConfigStore(storeFile));
+
+ assertFalse(fetcherManager.getConfig("fsf").json().contains("stale"));
+ // left for the operator to remove
+ assertEquals(Set.of("fsf"), new
FileBasedConfigStore(storeFile).keySet());
+ }
+
+ private static ExtensionConfig fileSystemFetcher(String id, Path basePath)
{
+ return new ExtensionConfig(id, "file-system-fetcher",
String.format(Locale.ROOT, """
+ {"basePath": "%s"}
+ """, JsonConfigHelper.toJsonPath(basePath)));
}
@Test
