TIKA-4595: Dynamic Fetcher/Emitter Management with ConfigStore Support (#2489)

* TIKA-4595: Add dynamic fetcher/emitter management API to PipesClient

- Added SAVE_FETCHER, DELETE_FETCHER, LIST_FETCHERS, GET_FETCHER commands
- Added SAVE_EMITTER, DELETE_EMITTER, LIST_EMITTERS, GET_EMITTER commands
- Implemented PipesClient public API methods for runtime configuration
- Implemented PipesServer command handlers
- Added deleteComponent() and getComponentConfig() to AbstractComponentManager
- Added wrapper methods to FetcherManager and EmitterManager
- Added remove() method to ConfigStore interface and implementations
- All tests passing

* Fix gRPC saveFetcher to propagate to PipesClient's forked PipesServer

- saveFetcher now calls both fetcherManager.saveFetcher() and pipesClient.saveFetcher()
- This ensures fetchers are available in the forked PipesServer process
- Implemented deleteFetcher to call both managers as well
- Fixes FetcherNotFoundException when using dynamic fetchers via gRPC

The issue was that fetchers saved via gRPC were only stored in the gRPC
server's FetcherManager, but when pipesClient.process() forks a new
PipesServer process, that process has its own FetcherManager and doesn't
have access to the dynamically created fetchers. Now both are updated.

* Add file-based ConfigStore for shared config between gRPC and PipesServer

- Created FileBasedConfigStore that persists to JSON file
- Created FileBasedConfigStoreFactory with @extension annotation
- Updated PipesServer.initializeResources() to create and use ConfigStore
- Both gRPC server and forked PipesServer can now share fetcher configs via file

This enables dynamic fetcher management across JVM processes:
1. gRPC saves fetcher → writes to config file
2. PipesServer starts → reads from same file
3. Both JVMs share the same fetcher configuration

* Handle 'file' ConfigStore type as built-in (not plugin)

- Added direct handling for 'file' type in ConfigStoreFactory.createConfigStore()
- File-based store is in core, not a plugin, so needs special handling
- Avoids ClassNotFoundException when trying to load 'file' as a class name
- Also added remove() method to IgniteConfigStore for interface compliance

* Make ExtensionConfig Serializable for socket communication

ExtensionConfig is sent over sockets between PipesClient and PipesServer,
so it needs to implement Serializable. Records can implement Serializable
and all fields (String, String, String) are already serializable.

Fixes NotSerializableException when calling saveFetcher via gRPC.

* Add embedded Ignite server architecture

- Created IgniteStoreServer class that runs as embedded server
- TikaGrpcServer starts Ignite server on startup (if ignite ConfigStore configured)
- IgniteConfigStore now acts as client-only (clientMode=true)
- No external Ignite dependency needed in Docker/Kubernetes
- Server runs in background daemon thread within tika-grpc process
- Clients (gRPC + forked PipesServer) connect to embedded server

Architecture:
  ┌─────────────────────────────────┐
  │      tika-grpc Process         │
  │  ┌──────────────────────────┐  │
  │  │  IgniteStoreServer       │  │
  │  │  (server mode, daemon)   │  │
  │  └────────▲─────────────────┘  │
  │           │                     │
  │  ┌────────┴─────────────────┐  │
  │  │ IgniteConfigStore        │  │
  │  │ (client mode)            │  │
  │  └──────────────────────────┘  │
  └─────────────────────────────────┘
           ▲
           │ (client connection)
           │
  ┌────────┴─────────────────┐
  │  PipesServer (forked)    │
  │  IgniteConfigStore       │
  │  (client mode)           │
  └──────────────────────────┘

* Configure Ignite work directory to /tmp

- Set workDirectory to /tmp/ignite-work in IgniteStoreServer
- Set workDirectory to /tmp/ignite-work in IgniteConfigStore
- Avoids 'Work directory does not exist and cannot be created: /work' error
- Uses system property ignite.work.dir if set, defaults to /tmp/ignite-work
- Ensures Ignite can write to work directory in Docker containers

* Use /var/cache/tika/ignite-work for Ignite work directory

- Changed from /tmp/ignite-work to /var/cache/tika/ignite-work
- Aligns with Tika's standard cache location
- /var/cache/tika is already used for plugins and other Tika cache data

* Use plugin classloader for Ignite server startup

- Find Ignite plugin's classloader from plugin manager
- Load IgniteStoreServer and CacheMode using plugin classloader
- Fixes NoClassDefFoundError for H2 classes
- Ensures all Ignite dependencies (including H2) are available
- Plugin classloader has all dependencies from lib/ directory

* Enable peer class loading in Ignite client

- Set setPeerClassLoadingEnabled(true) in IgniteConfigStore
- Must match server configuration
- Fixes: Remote node has peer class loading enabled flag different from local
- Both server and client now have peerClassLoading=true

* Disable peer class loading to fix classloader conflicts

- Set setPeerClassLoadingEnabled(false) on both server and client
- Fixes ClassCastException due to class loaded by different classloaders
- Server uses plugin classloader, client uses app classloader
- Peer class loading causes the same class to be in both, creating conflicts
- We don't need peer class loading for our use case

* Move Ignite dependencies directly into tika-grpc

- Made tika-pipes-ignite a required (non-optional) dependency of tika-grpc
- Added ignite.version and h2.version properties
- Removed reflection-based classloader lookup
- Direct instantiation of IgniteStoreServer
- Avoids all PF4J plugin classloader issues
- Ignite classes now on main classpath

* Disable Ignite Object Input Filter autoconfiguration

- Set IGNITE_ENABLE_OBJECT_INPUT_FILTER_AUTOCONFIGURATION=false
- Fixes: Failed to autoconfigure Ignite Object Input Filter
- Ignite was conflicting with existing serialization filter
- Apply in both IgniteStoreServer and IgniteConfigStore

* Add Ignite filter autoconfiguration disable to client too

* Add Ignite as built-in ConfigStore type

- Handle 'ignite' type directly in ConfigStoreFactory
- Load IgniteConfigStoreFactory via reflection
- Works in forked PipesServer without plugin system
- Matches pattern used for 'file' type
- Fixes: ClassNotFoundException: ignite in forked process

* Add PipesIterator management API with save, get, and delete operations

* Fix Ignite tests to use temp directory and server mode

* Fix TikaGrpcServerTest - update assertion for delete operation now that it's implemented

Author: nddipiazza

tika-grpc/pom.xml

@@ -38,6 +38,8 @@
     <asarkar-grpc-test.version>2.0.0</asarkar-grpc-test.version>
     <awaitility.version>4.3.0</awaitility.version>
     <j2objc-annotations.version>3.1</j2objc-annotations.version>
+    <ignite.version>2.17.0</ignite.version>
+    <h2.version>1.4.197</h2.version>
   </properties>
 
   <dependencyManagement>
@@ -232,7 +234,6 @@
       <groupId>org.apache.tika</groupId>
       <artifactId>tika-pipes-ignite</artifactId>
       <version>${project.version}</version>
-      <optional>true</optional>
     </dependency>
     <dependency>
       <groupId>org.apache.tika</groupId>

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

@@ -39,16 +39,22 @@
 
 import org.apache.tika.DeleteFetcherReply;
 import org.apache.tika.DeleteFetcherRequest;
+import org.apache.tika.DeletePipesIteratorReply;
+import org.apache.tika.DeletePipesIteratorRequest;
 import org.apache.tika.FetchAndParseReply;
 import org.apache.tika.FetchAndParseRequest;
 import org.apache.tika.GetFetcherConfigJsonSchemaReply;
 import org.apache.tika.GetFetcherConfigJsonSchemaRequest;
 import org.apache.tika.GetFetcherReply;
 import org.apache.tika.GetFetcherRequest;
+import org.apache.tika.GetPipesIteratorReply;
+import org.apache.tika.GetPipesIteratorRequest;
 import org.apache.tika.ListFetchersReply;
 import org.apache.tika.ListFetchersRequest;
 import org.apache.tika.SaveFetcherReply;
 import org.apache.tika.SaveFetcherRequest;
+import org.apache.tika.SavePipesIteratorReply;
+import org.apache.tika.SavePipesIteratorRequest;
 import org.apache.tika.TikaGrpc;
 import org.apache.tika.config.ConfigContainer;
 import org.apache.tika.config.loader.TikaJsonConfig;
@@ -131,11 +137,43 @@ private ConfigStore createConfigStore() throws TikaConfigException {
         ExtensionConfig storeConfig = new ExtensionConfig(
             configStoreType, configStoreType, configStoreParams);
 
+        // If using Ignite, start the embedded server first
+        if ("ignite".equalsIgnoreCase(configStoreType)) {
+            startIgniteServer(storeConfig);
+        }
+
         return ConfigStoreFactory.createConfigStore(
                 pluginManager,
                 configStoreType,
                 storeConfig);
     }
+    
+    private void startIgniteServer(ExtensionConfig config) {
+        try {
+            LOG.info("Starting embedded Ignite server for ConfigStore");
+            
+            // Parse config to get Ignite settings
+            com.fasterxml.jackson.databind.ObjectMapper mapper = new com.fasterxml.jackson.databind.ObjectMapper();
+            com.fasterxml.jackson.databind.JsonNode params = mapper.readTree(config.json());
+            
+            String cacheName = params.has("cacheName") ? params.get("cacheName").asText() : "tika-config-store";
+            String cacheMode = params.has("cacheMode") ? params.get("cacheMode").asText() : "REPLICATED";
+            String instanceName = params.has("igniteInstanceName") ? params.get("igniteInstanceName").asText() : "TikaIgniteServer";
+            
+            // Direct instantiation - no reflection needed
+            org.apache.ignite.cache.CacheMode mode = org.apache.ignite.cache.CacheMode.valueOf(cacheMode);
+            org.apache.tika.pipes.ignite.server.IgniteStoreServer server = 
+                new org.apache.tika.pipes.ignite.server.IgniteStoreServer(cacheName, mode, instanceName);
+            
+            server.startAsync();
+            
+            LOG.info("Embedded Ignite server started successfully");
+            
+        } catch (Exception e) {
+            LOG.error("Failed to start embedded Ignite server", e);
+            throw new RuntimeException("Failed to start Ignite server", e);
+        }
+    }
 
     @Override
     public void fetchAndParseServerSideStreaming(FetchAndParseRequest request,
@@ -225,7 +263,12 @@ public void saveFetcher(SaveFetcherRequest request,
         try {
             String factoryName = findFactoryNameForClass(request.getFetcherClass());
             ExtensionConfig config = new ExtensionConfig(request.getFetcherId(), factoryName, request.getFetcherConfigJson());
+            
+            // Save to gRPC server's fetcher manager (for schema queries, etc.)
             fetcherManager.saveFetcher(config);
+            
+            // Also save to PipesClient so it propagates to the forked PipesServer
+            pipesClient.saveFetcher(config);
         } catch (Exception e) {
             throw new RuntimeException(e);
         }
@@ -331,7 +374,113 @@ public void getFetcherConfigJsonSchema(GetFetcherConfigJsonSchemaRequest request
     }
 
     private boolean deleteFetcher(String id) {
-        LOG.warn("Deleting fetchers is not supported in the current implementation");
-        return false;
+        try {
+            // Delete from gRPC server's fetcher manager
+            fetcherManager.deleteFetcher(id);
+            
+            // Also delete from PipesClient so it propagates to the forked PipesServer
+            pipesClient.deleteFetcher(id);
+            
+            LOG.info("Successfully deleted fetcher: {}", id);
+            return true;
+        } catch (Exception e) {
+            LOG.error("Failed to delete fetcher: {}", id, e);
+            return false;
+        }
+    }
+    
+    // ========== PipesIterator RPC Methods ==========
+    
+    @Override
+    public void savePipesIterator(SavePipesIteratorRequest request,
+                                  StreamObserver<SavePipesIteratorReply> responseObserver) {
+        try {
+            String iteratorId = request.getIteratorId();
+            String iteratorClass = request.getIteratorClass();
+            String iteratorConfigJson = request.getIteratorConfigJson();
+            
+            LOG.info("Saving pipes iterator: id={}, class={}", iteratorId, iteratorClass);
+            
+            ExtensionConfig config = new ExtensionConfig(iteratorId, iteratorClass, iteratorConfigJson);
+            
+            // Save via PipesClient so it propagates to the forked PipesServer
+            pipesClient.savePipesIterator(config);
+            
+            SavePipesIteratorReply reply = SavePipesIteratorReply.newBuilder()
+                    .setMessage("Pipes iterator saved successfully")
+                    .build();
+            responseObserver.onNext(reply);
+            responseObserver.onCompleted();
+            
+            LOG.info("Successfully saved pipes iterator: {}", iteratorId);
+            
+        } catch (Exception e) {
+            LOG.error("Failed to save pipes iterator", e);
+            responseObserver.onError(io.grpc.Status.INTERNAL
+                    .withDescription("Failed to save pipes iterator: " + e.getMessage())
+                    .withCause(e)
+                    .asRuntimeException());
+        }
+    }
+    
+    @Override
+    public void getPipesIterator(GetPipesIteratorRequest request,
+                                 StreamObserver<GetPipesIteratorReply> responseObserver) {
+        try {
+            String iteratorId = request.getIteratorId();
+            LOG.info("Getting pipes iterator: {}", iteratorId);
+            
+            ExtensionConfig config = pipesClient.getPipesIteratorConfig(iteratorId);
+            
+            if (config == null) {
+                responseObserver.onError(io.grpc.Status.NOT_FOUND
+                        .withDescription("Pipes iterator not found: " + iteratorId)
+                        .asRuntimeException());
+                return;
+            }
+            
+            GetPipesIteratorReply reply = GetPipesIteratorReply.newBuilder()
+                    .setIteratorId(config.id())
+                    .setIteratorClass(config.name())
+                    .setIteratorConfigJson(config.json())
+                    .build();
+            responseObserver.onNext(reply);
+            responseObserver.onCompleted();
+            
+            LOG.info("Successfully retrieved pipes iterator: {}", iteratorId);
+            
+        } catch (Exception e) {
+            LOG.error("Failed to get pipes iterator", e);
+            responseObserver.onError(io.grpc.Status.INTERNAL
+                    .withDescription("Failed to get pipes iterator: " + e.getMessage())
+                    .withCause(e)
+                    .asRuntimeException());
+        }
+    }
+    
+    @Override
+    public void deletePipesIterator(DeletePipesIteratorRequest request,
+                                    StreamObserver<DeletePipesIteratorReply> responseObserver) {
+        try {
+            String iteratorId = request.getIteratorId();
+            LOG.info("Deleting pipes iterator: {}", iteratorId);
+            
+            pipesClient.deletePipesIterator(iteratorId);
+            
+            DeletePipesIteratorReply reply = DeletePipesIteratorReply.newBuilder()
+                    .setMessage("Pipes iterator deleted successfully")
+                    .build();
+            responseObserver.onNext(reply);
+            responseObserver.onCompleted();
+            
+            LOG.info("Successfully deleted pipes iterator: {}", iteratorId);
+            
+        } catch (Exception e) {
+            LOG.error("Failed to delete pipes iterator", e);
+            responseObserver.onError(io.grpc.Status.INTERNAL
+                    .withDescription("Failed to delete pipes iterator: " + e.getMessage())
+                    .withCause(e)
+                    .asRuntimeException());
+        }
     }
 }

tika-grpc/src/main/proto/tika.proto

@@ -59,6 +59,19 @@ service Tika   {
     Get the Fetcher Config schema for a given fetcher class.
   */
   rpc GetFetcherConfigJsonSchema(GetFetcherConfigJsonSchemaRequest) returns (GetFetcherConfigJsonSchemaReply) {}
+  
+  /* 
+    Save a pipes iterator to the iterator store.   
+  */
+  rpc SavePipesIterator(SavePipesIteratorRequest) returns (SavePipesIteratorReply) {}
+  /* 
+    Get a pipes iterator's data from the iterator store.   
+  */
+  rpc GetPipesIterator(GetPipesIteratorRequest) returns (GetPipesIteratorReply) {}
+  /* 
+    Delete a pipes iterator from the iterator store.   
+  */
+  rpc DeletePipesIterator(DeletePipesIteratorRequest) returns (DeletePipesIteratorReply) {}
 }
 
 message SaveFetcherRequest {
@@ -143,3 +156,43 @@ message GetFetcherConfigJsonSchemaReply {
   // The json schema that describes the fetcher config in string format.
   string fetcher_config_json_schema = 1;
 }
+
+// ========== PipesIterator Messages ==========
+
+message SavePipesIteratorRequest {
+  // A unique identifier for each pipes iterator. If this already exists, operation will overwrite existing.
+  string iterator_id = 1;
+  // The full java class name of the pipes iterator class.
+  string iterator_class = 2;
+  // JSON string of the pipes iterator config object.
+  string iterator_config_json = 3;
+}
+
+message SavePipesIteratorReply {
+  // Status message
+  string message = 1;
+}
+
+message GetPipesIteratorRequest {
+  // The pipes iterator ID to retrieve
+  string iterator_id = 1;
+}
+
+message GetPipesIteratorReply {
+  // The pipes iterator ID
+  string iterator_id = 1;
+  // The full java class name of the pipes iterator
+  string iterator_class = 2;
+  // JSON string of the pipes iterator config object
+  string iterator_config_json = 3;
+}
+
+message DeletePipesIteratorRequest {
+  // The pipes iterator ID to delete
+  string iterator_id = 1;
+}
+
+message DeletePipesIteratorReply {
+  // Status message
+  string message = 1;
+}

tika-grpc/src/test/java/org/apache/tika/pipes/grpc/TikaGrpcServerTest.java

@@ -184,15 +184,15 @@ public void testFetcherCrud(Resources resources) throws Exception {
             assertEquals(FileSystemFetcher.class.getName(), getFetcherReply.getFetcherClass());
         }
 
-        // delete fetchers - note: delete is not currently supported
+        // delete fetchers
         for (int i = 0; i < NUM_FETCHERS_TO_CREATE; ++i) {
             String fetcherId = createFetcherId(i);
             DeleteFetcherReply deleteFetcherReply = blockingStub.deleteFetcher(DeleteFetcherRequest
                     .newBuilder()
                     .setFetcherId(fetcherId)
                     .build());
-            // Delete is not supported, so this will return false
-            Assertions.assertFalse(deleteFetcherReply.getSuccess());
+            // Delete is now supported and should succeed
+            Assertions.assertTrue(deleteFetcherReply.getSuccess());
         }
     }
 

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

@@ -309,6 +309,47 @@ public synchronized void saveComponent(ExtensionConfig config) throws TikaConfig
         configStore.put(componentId, config);
     }
 
+    /**
+     * Deletes a component configuration by ID.
+     * 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
+     */
+    public synchronized void deleteComponent(String componentId) throws TikaConfigException {
+        if (!allowRuntimeModifications) {
+            throw new TikaConfigException(
+                    "Runtime modifications are not allowed. " + getClass().getSimpleName() +
+                    " must be loaded with allowRuntimeModifications=true to use delete" +
+                    getComponentName().substring(0, 1).toUpperCase(Locale.ROOT) + getComponentName().substring(1) + "()");
+        }
+
+        if (componentId == null) {
+            throw new IllegalArgumentException("Component ID cannot be null");
+        }
+
+        if (!configStore.containsKey(componentId)) {
+            throw new TikaConfigException(
+                    getComponentName().substring(0, 1).toUpperCase(Locale.ROOT) + 
+                    getComponentName().substring(1) + " with ID '" + componentId + "' not found");
+        }
+
+        // Clear cache and remove config
+        componentCache.remove(componentId);
+        configStore.remove(componentId);
+        LOG.debug("Deleted {} config: id={}", getComponentName(), componentId);
+    }
+
+    /**
+     * Gets the configuration for a specific component by ID.
+     *
+     * @param componentId the component ID
+     * @return the component configuration, or null if not found
+     */
+    public ExtensionConfig getComponentConfig(String componentId) {
+        return configStore.get(componentId);
+    }
+
     /**
      * Returns the set of supported component IDs.
      */