TIKA-5001: step 1

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
 

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`
 |`"{}"`

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.
 |===

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.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 void testFetcherCrud(Resources resources) throws Exception {
         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 void testFetcherCrud(Resources resources) throws Exception {
         }
         // 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 void testFetcherCrud(Resources resources) throws Exception {
 
         // 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 void testFetcherCrud(Resources resources) throws Exception {
 
         // 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 void onCompleted() {
         }
     }
 
+    /**
+     * 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 void testBiStream(Resources resources) throws Exception {
         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<>());

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 @@ protected AbstractComponentManager(PluginManager pluginManager,
         } 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 T getComponent(String id) throws IOException, TikaException {
         }
 
         // 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 T getComponent(String id) throws IOException, TikaException {
         }
     }
 
+    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 @@ private T buildComponent(ExtensionConfig config) throws TikaConfigException, IOE
      * Dynamically adds or updates a component configuration at runtime.
      * The component will not be instantiated until it is first requested via {@link #getComponent(String)}.
      * If a component with the same ID already exists, it will be replaced and the cached instance cleared.
+     * 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 synchronized void saveComponent(ExtensionConfig config) throws TikaConfig
         } 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 synchronized void saveComponent(ExtensionConfig config) throws TikaConfig
      * 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 synchronized void deleteComponent(String componentId) throws TikaConfigEx
         } catch (IllegalArgumentException e) {
             throw new TikaConfigException(e.getMessage(), e);
         }
+        requireNotConfigured(componentId);
 
         if (!configStore.containsKey(componentId)) {
             throw new TikaConfigException(
@@ -390,15 +418,15 @@ public synchronized void deleteComponent(String componentId) throws TikaConfigEx
      * @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));
     }
 
     /**
      * Returns the component IDs a caller may name. Host components are omitted: a request cannot
      * 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 Set<String> getSupported() {
      * 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 Set<String> getAllIds() {
      * @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);
     }
 }

tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/emitter/EmitterManager.java

@@ -161,6 +161,7 @@ public Emitter getEmitter() throws IOException, TikaException {
      * 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.

tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/fetcher/FetcherManager.java

@@ -160,6 +160,7 @@ public Fetcher getFetcher() throws IOException, TikaException {
      * 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.

tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/core/emitter/EmitterManagerTest.java

@@ -384,33 +384,22 @@ public void testSaveEmitterDuplicate(@TempDir Path tmpDir) throws Exception {
         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 void testSaveEmitterNotAllowedExplicit(@TempDir Path tmpDir) throws Excep
 
         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)));
+    }
 }