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 ddc99e2b88 TIKA-5001 step3 (#3441)
ddc99e2b88 is described below
commit ddc99e2b8879001771d70091e9a4d2b9b7f3c8c9
Author: Tim Allison <[email protected]>
AuthorDate: Thu Oct 29 13:54:51 2026 -0400
TIKA-5001 step3 (#3441)
* TIKA-5001: step 2, fetcher/emitter leases and close
* TIKA-5001: step 3, propagate config store changes to running forks
* TIKA-5001: e2e uses file config store; count only success statuses
* TIKA-5001: e2e limit path requires a success; stop leaked grpc server
---
CHANGES.txt | 9 ++
docs/modules/ROOT/pages/pipes/configuration.adoc | 2 +-
docs/modules/ROOT/pages/using-tika/grpc/index.adoc | 16 ++-
.../org/apache/tika/pipes/ExternalTestBase.java | 35 ++++-
.../pipes/filesystem/FileSystemFetcherTest.java | 6 +-
.../tika/pipes/filesystem/HandlerTypeTest.java | 1 +
.../tika/pipes/ignite/IgniteConfigStoreTest.java | 7 +-
.../tika-grpc/src/test/resources/tika-config.json | 2 +
.../apache/tika/pipes/grpc/TikaGrpcServerImpl.java | 16 +++
.../apache/tika/pipes/grpc/TikaGrpcServerTest.java | 75 +++++++++--
.../tika/pipes/ignite/IgniteConfigStore.java | 53 ++++++--
.../tika/pipes/ignite/IgniteConfigStoreTest.java | 27 ++++
.../tika/pipes/core/AbstractComponentManager.java | 38 ++++++
.../org/apache/tika/pipes/core/PipesConfig.java | 2 +-
.../apache/tika/pipes/core/config/ConfigStore.java | 11 ++
.../pipes/core/config/FileBasedConfigStore.java | 26 ++++
.../tika/pipes/core/StorePropagationTest.java | 143 +++++++++++++++++++++
.../tika/pipes/core/TrackingFetcherManager.java | 7 +-
18 files changed, 438 insertions(+), 38 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 69bc2cf552..2a8e7586d8 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -9,6 +9,10 @@ Release 4.3.0 - ???
last task using it ends) and on graceful shutdown; nothing closed them
before.
EmittingUnpackHandler takes the emitter, not the EmitterManager
(TIKA-5001).
+ * tika-grpc refuses SaveFetcher with configStoreType memory
(FAILED_PRECONDITION):
+ the forks never saw such a fetcher, so every fetch through it failed. Use
file or
+ ignite (TIKA-5001).
+
SERVER, PIPES AND OPERATIONS
* The tika-server full and tika-grpc Docker images no longer install
@@ -42,6 +46,11 @@ Release 4.3.0 - ???
OTHER CHANGES
+ * A runtime fetcher or emitter save or delete reaches running pipes workers
at
+ their next fetch, through a change token on file and ignite stores
+ (ConfigStore.changeToken); before, a worker kept the instance it first
built
+ until it restarted (TIKA-5001).
+
* PEM-armored CMS/PKCS7 (-----BEGIN PKCS7----- or -----BEGIN CMS-----) is
parsed like
DER: Pkcs7Parser and the opt-in pkcs-7-detector report the smime-type and
extract
signed content; -----BEGIN CMS----- is now detected (TIKA-5002).
diff --git a/docs/modules/ROOT/pages/pipes/configuration.adoc
b/docs/modules/ROOT/pages/pipes/configuration.adoc
index ce63831dc0..eeeac19c3a 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 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.
+|Backend for storing fetcher/emitter configurations registered at runtime;
those in the config file are never written to it. `"memory"` (default) is
in-process and not seen by the forked workers; `"file"` persists to a JSON
file; `"ignite"` uses Apache Ignite for shared state across nodes. With `file`
or `ignite`, a runtime save or delete reaches running workers at their next
fetch.
|`configStoreParams`
|`"{}"`
diff --git a/docs/modules/ROOT/pages/using-tika/grpc/index.adoc
b/docs/modules/ROOT/pages/using-tika/grpc/index.adoc
index 0642be675f..f1e378f60e 100644
--- a/docs/modules/ROOT/pages/using-tika/grpc/index.adoc
+++ b/docs/modules/ROOT/pages/using-tika/grpc/index.adoc
@@ -126,11 +126,17 @@ config.
|===
Fetches run in the forked pipes workers, and each worker builds its own config
store from the
-same settings. With the default in-memory `configStoreType`, a fetcher saved
through
-`SaveFetcher` exists only in the gRPC JVM and every `FetchAndParse` that names
it fails with
-`FETCHER_NOT_FOUND`. When `allowComponentManagement` is `true`, set
`configStoreType` to `file`
-or `ignite` (see
xref:pipes/configuration.adoc#_distributed_config_store[Distributed Config
-Store]) so the workers see the same store.
+same settings. A fetcher saved to the default in-memory `configStoreType`
would exist only in the
+gRPC JVM, so `SaveFetcher` is refused with `FAILED_PRECONDITION`. To save
fetchers at runtime,
+set `configStoreType` to `file` or `ignite` (see
+xref:pipes/configuration.adoc#_distributed_config_store[Distributed Config
Store]) so the
+workers see the same store.
+
+A save or delete takes effect for fetches that start after it, in every
worker. A fetch already
+running finishes with the fetcher it started with, even if that fetcher was
just deleted or its
+credentials replaced; the old instance is closed when the last such fetch
ends. Each fetch through
+a runtime fetcher reads the store's change marker, so while an Ignite store is
unreachable those
+fetches fail with `FETCHER_INITIALIZATION_EXCEPTION`; fetchers from the config
file are unaffected.
Enable these only for trusted callers over a secured channel:
diff --git
a/tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/ExternalTestBase.java
b/tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/ExternalTestBase.java
index 5ebe3d569d..8ce26f09c9 100644
---
a/tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/ExternalTestBase.java
+++
b/tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/ExternalTestBase.java
@@ -23,6 +23,7 @@ import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
+import java.net.ServerSocket;
import java.net.URL;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
@@ -34,7 +35,9 @@ import java.util.HashSet;
import java.util.List;
import java.util.Locale;
import java.util.Set;
+import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import java.util.regex.Pattern;
import java.util.stream.Stream;
import java.util.zip.ZipEntry;
@@ -55,6 +58,7 @@ import org.testcontainers.containers.output.Slf4jLogConsumer;
import org.testcontainers.containers.wait.strategy.Wait;
import org.testcontainers.junit.jupiter.Testcontainers;
+import org.apache.tika.pipes.api.PipesResult;
import org.apache.tika.pipes.grpc.proto.FetchAndParseReply;
import org.apache.tika.pipes.grpc.proto.ListFetchersRequest;
import org.apache.tika.pipes.grpc.proto.TikaGrpc;
@@ -104,7 +108,8 @@ public abstract class ExternalTestBase {
private static void startLocalGrpcServer() throws Exception {
LOG.info("Starting local tika-grpc server using Maven exec");
-
+ assertPortFree(GRPC_PORT);
+
Path tikaGrpcDir = findTikaGrpcDirectory();
Path configFile = localConfig("tika-config.json");
@@ -275,14 +280,25 @@ public abstract class ExternalTestBase {
void close() {
if (USE_LOCAL_SERVER && localGrpcProcess != null) {
LOG.info("Stopping local gRPC server");
+ // mvn exec's java child outlives mvn and keeps the port; snapshot
before it reparents
+ List<ProcessHandle> children =
localGrpcProcess.descendants().toList();
localGrpcProcess.destroy();
+ children.forEach(ProcessHandle::destroy);
try {
if (!localGrpcProcess.waitFor(10, TimeUnit.SECONDS)) {
localGrpcProcess.destroyForcibly();
}
+ for (ProcessHandle child : children) {
+ try {
+ child.onExit().get(10, TimeUnit.SECONDS);
+ } catch (ExecutionException | TimeoutException e) {
+ child.destroyForcibly();
+ }
+ }
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
localGrpcProcess.destroyForcibly();
+ children.forEach(ProcessHandle::destroyForcibly);
}
} else if (composeContainer != null) {
composeContainer.close();
@@ -327,6 +343,23 @@ public abstract class ExternalTestBase {
LOG.info("Finished downloading and extracting govdocs1 files");
}
+ /**
+ * The readiness probe accepts any server on the port, so a leftover one
would silently
+ * answer this test's requests.
+ */
+ public static void assertPortFree(int port) {
+ try (ServerSocket ignored = new ServerSocket(port)) {
+ // free
+ } catch (IOException e) {
+ throw new IllegalStateException("Port " + port + " is already in
use, likely a "
+ + "leftover tika-grpc server from an earlier run; stop it
and rerun", e);
+ }
+ }
+
+ public static boolean isSuccess(FetchAndParseReply reply) {
+ return
PipesResult.RESULT_STATUS.valueOf(reply.getStatus()).isSuccess();
+ }
+
public static void assertAllFilesFetched(Path baseDir,
List<FetchAndParseReply> successes,
List<FetchAndParseReply> errors) {
Set<String> allFetchKeys = new HashSet<>();
diff --git
a/tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/filesystem/FileSystemFetcherTest.java
b/tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/filesystem/FileSystemFetcherTest.java
index 197209276d..0c38496e8e 100644
---
a/tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/filesystem/FileSystemFetcherTest.java
+++
b/tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/filesystem/FileSystemFetcherTest.java
@@ -82,7 +82,7 @@ class FileSystemFetcherTest extends ExternalTestBase {
public void onNext(FetchAndParseReply fetchAndParseReply) {
LOG.debug("Reply from fetch-and-parse - key={}, status={}",
fetchAndParseReply.getFetchKey(),
fetchAndParseReply.getStatus());
- if
("FETCH_AND_PARSE_EXCEPTION".equals(fetchAndParseReply.getStatus())) {
+ if (!ExternalTestBase.isSuccess(fetchAndParseReply)) {
errors.add(fetchAndParseReply);
} else {
successes.add(fetchAndParseReply);
@@ -145,8 +145,8 @@ class FileSystemFetcherTest extends ExternalTestBase {
LOG.info("Processed {} documents (limit was {})", totalProcessed,
maxDocs);
Assertions.assertTrue(totalProcessed <= maxDocs,
"Should not process more than " + maxDocs + " documents");
- Assertions.assertTrue(totalProcessed > 0,
- "Should have processed at least one document");
+ Assertions.assertNotEquals(0, successes.size(),
+ () -> "Should have some successful fetches; errors: " +
errors);
}
LOG.info("Test completed successfully - {} successes, {} errors",
diff --git
a/tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/filesystem/HandlerTypeTest.java
b/tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/filesystem/HandlerTypeTest.java
index d78fe6ab0a..71f9242d38 100644
---
a/tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/filesystem/HandlerTypeTest.java
+++
b/tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/filesystem/HandlerTypeTest.java
@@ -106,6 +106,7 @@ class HandlerTypeTest {
private static void startLocalGrpcServer() throws Exception {
LOG.info("Starting local tika-grpc server with Ignite config for
HandlerType test");
+ ExternalTestBase.assertPortFree(GRPC_PORT);
Path currentDir = Path.of("").toAbsolutePath();
Path tikaRootDir = currentDir;
diff --git
a/tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/ignite/IgniteConfigStoreTest.java
b/tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/ignite/IgniteConfigStoreTest.java
index d8c5101089..1e2dcd5652 100644
---
a/tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/ignite/IgniteConfigStoreTest.java
+++
b/tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/ignite/IgniteConfigStoreTest.java
@@ -109,6 +109,7 @@ class IgniteConfigStoreTest {
private static void startLocalGrpcServer() throws Exception {
LOG.info("Starting local tika-grpc server using Maven");
+ ExternalTestBase.assertPortFree(GRPC_PORT);
Path currentDir = Path.of("").toAbsolutePath();
Path tikaRootDir = currentDir;
@@ -428,7 +429,7 @@ class IgniteConfigStoreTest {
public void onNext(FetchAndParseReply fetchAndParseReply) {
LOG.debug("Reply from fetch-and-parse - key={},
status={}",
fetchAndParseReply.getFetchKey(),
fetchAndParseReply.getStatus());
- if
("FETCH_AND_PARSE_EXCEPTION".equals(fetchAndParseReply.getStatus())) {
+ if (!ExternalTestBase.isSuccess(fetchAndParseReply)) {
errors.add(fetchAndParseReply);
} else {
successes.add(fetchAndParseReply);
@@ -498,8 +499,8 @@ class IgniteConfigStoreTest {
totalProcessed, maxDocs);
Assertions.assertTrue(totalProcessed <= maxDocs,
"Should not process more than " + maxDocs + " documents");
- Assertions.assertTrue(totalProcessed > 0,
- "Should have processed at least one document");
+ Assertions.assertNotEquals(0, successes.size(),
+ () -> "Should have some successful fetches; errors: " +
errors);
}
LOG.info("Ignite ConfigStore test completed successfully - {}
successes, {} errors",
diff --git a/tika-e2e-tests/tika-grpc/src/test/resources/tika-config.json
b/tika-e2e-tests/tika-grpc/src/test/resources/tika-config.json
index 0529991d3d..d48b36bf15 100644
--- a/tika-e2e-tests/tika-grpc/src/test/resources/tika-config.json
+++ b/tika-e2e-tests/tika-grpc/src/test/resources/tika-config.json
@@ -4,6 +4,8 @@
],
"pipes": {
"numClients": 1,
+ "configStoreType": "file",
+ "configStoreParams": "{\"path\": \"target/tika-e2e-config-store.json\"}",
"forkedJvmArgs": [
"--add-opens=java.base/jdk.internal.access=ALL-UNNAMED",
"--add-opens=java.base/jdk.internal.misc=ALL-UNNAMED",
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 9735759b06..c9eb605b24 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
@@ -57,6 +57,7 @@ import org.apache.tika.pipes.core.PipesParser;
import org.apache.tika.pipes.core.config.ConfigStore;
import org.apache.tika.pipes.core.config.ConfigStoreFactory;
import org.apache.tika.pipes.core.config.DefaultPluginsDir;
+import org.apache.tika.pipes.core.config.InMemoryConfigStore;
import org.apache.tika.pipes.core.fetcher.FetcherManager;
import org.apache.tika.pipes.grpc.proto.DeleteFetcherReply;
import org.apache.tika.pipes.grpc.proto.DeleteFetcherRequest;
@@ -92,6 +93,10 @@ class TikaGrpcServerImpl extends TikaGrpc.TikaImplBase {
public static final JsonSchemaGenerator JSON_SCHEMA_GENERATOR = new
JsonSchemaGenerator(OBJECT_MAPPER);
private static final String PIPES_ITERATOR_PREFIX = "pipesIterator:";
+ // the pipes workers are separate JVMs, each with its own empty in-memory
store
+ static final String MEMORY_STORE_SAVE_REFUSED = "SaveFetcher is refused: a
fetcher saved "
+ + "to the in-memory config store never reaches the pipes workers,
so every fetch "
+ + "through it would fail. Set pipes.configStoreType to 'file' or
'ignite'.";
private static final String IGNITE_STORE_SERVER_CLASS =
"org.apache.tika.pipes.ignite.server.IgniteStoreServer";
PipesConfig pipesConfig;
@@ -167,6 +172,11 @@ class TikaGrpcServerImpl extends TikaGrpc.TikaImplBase {
}
this.configStore = createConfigStore();
+ if (tikaGrpcConfig.isAllowComponentManagement()
+ && configStore instanceof InMemoryConfigStore) {
+ LOG.warn("allowComponentManagement is on with configStoreType
'memory': {}",
+ MEMORY_STORE_SAVE_REFUSED);
+ }
fetcherManager = FetcherManager.load(pluginManager, tikaJsonConfig,
tikaGrpcConfig.isAllowComponentManagement(), this.configStore);
@@ -448,6 +458,12 @@ class TikaGrpcServerImpl extends TikaGrpc.TikaImplBase {
if (denyComponentManagement(responseObserver)) {
return;
}
+ if (configStore instanceof InMemoryConfigStore) {
+ responseObserver.onError(io.grpc.Status.FAILED_PRECONDITION
+ .withDescription(MEMORY_STORE_SAVE_REFUSED)
+ .asRuntimeException());
+ return;
+ }
String fetcherType = request.getFetcherType();
if (!isRegisteredFetcherType(fetcherType)) {
responseObserver.onError(io.grpc.Status.INVALID_ARGUMENT
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 e4e28e9c55..5ca2787249 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
@@ -201,13 +201,13 @@ public class TikaGrpcServerTest {
static final int NUM_FETCHERS_TO_CREATE = 10;
@Test
- public void testFetcherCrud(Resources resources) throws Exception {
+ public void testFetcherCrud(Resources resources, @TempDir Path tmp) throws
Exception {
String serverName = InProcessServerBuilder.generateName();
Server server = InProcessServerBuilder
.forName(serverName)
.directExecutor()
- .addService(newService(tikaConfigUnlocked))
+ .addService(newService(fileStoreConfig(tmp)))
.build()
.start();
resources.register(server, Duration.ofSeconds(10));
@@ -287,6 +287,32 @@ public class TikaGrpcServerTest {
return "nick" + i + ".is.cool.super-fs";
}
+ @Test
+ public void testSaveFetcherRefusedWithMemoryStore(Resources resources)
throws Exception {
+ // the forks have their own empty in-memory stores; "success" here
meant every fetch failed
+ TikaGrpc.TikaBlockingStub blockingStub = startServer(resources,
tikaConfigUnlocked);
+ StatusRuntimeException ex =
Assertions.assertThrows(StatusRuntimeException.class, () ->
+ blockingStub.saveFetcher(SaveFetcherRequest
+ .newBuilder()
+ .setFetcherId("saved.fs")
+ .setFetcherType("file-system-fetcher")
+ .setFetcherConfigJson("{}")
+ .build()));
+ assertEquals(Status.Code.FAILED_PRECONDITION,
ex.getStatus().getCode());
+ }
+
+ /** The unlocked config with a file config store at {@code
tmp/config-store.json}. */
+ private static Path fileStoreConfig(Path tmp) throws Exception {
+ 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",
tmp.resolve("config-store.json").toAbsolutePath().toString())));
+ Path config = tmp.resolve("tika-config-file-store.json");
+ Files.writeString(config, OBJECT_MAPPER.writeValueAsString(root),
StandardCharsets.UTF_8);
+ return config;
+ }
+
@Test
public void testComponentManagementDeniedByDefault(Resources resources)
throws Exception {
TikaGrpc.TikaBlockingStub blockingStub = startServer(resources,
tikaConfig);
@@ -643,20 +669,15 @@ public class TikaGrpcServerTest {
}
/**
- * The point of a persistent store: a fetcher saved at runtime reaches the
forked worker.
+ * The point of a persistent store: a fetcher saved at runtime reaches the
forked worker,
+ * and so do later updates and deletes, though the fork already built the
old instance.
* 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);
+ Path config = fileStoreConfig(tmp);
String serverName = InProcessServerBuilder.generateName();
Server server =
InProcessServerBuilder.forName(serverName).directExecutor()
@@ -688,6 +709,40 @@ public class TikaGrpcServerTest {
String stored = Files.readString(store, StandardCharsets.UTF_8);
assertTrue(stored.contains("saved.fs"), stored);
assertFalse(stored.contains("nick1.is.cool.super-fs"), stored);
+
+ // the fork cached the first instance; the update must replace it
there too
+ Path docs2 = Files.createDirectories(tmp.resolve("docs2"));
+ Files.writeString(docs2.resolve("moved.html"),
+ "<html><body>updated-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", docs2.toAbsolutePath().toString())))
+ .build());
+ reply = blockingStub.fetchAndParse(FetchAndParseRequest.newBuilder()
+ .setFetcherId("saved.fs")
+ .setFetchKey("moved.html")
+ .build());
+ assertEquals(PipesResult.RESULT_STATUS.PARSE_SUCCESS.name(),
reply.getStatus(),
+ reply.getErrorMessage());
+ assertTrue(reply.getFieldsMap().values().stream()
+ .anyMatch(v -> v.contains("updated-fetcher-marker")));
+ reply = blockingStub.fetchAndParse(FetchAndParseRequest.newBuilder()
+ .setFetcherId("saved.fs")
+ .setFetchKey("saved.html")
+ .build());
+ assertEquals(PipesResult.RESULT_STATUS.FETCH_EXCEPTION.name(),
reply.getStatus(),
+ "the fork still used the old basePath");
+
+ blockingStub.deleteFetcher(DeleteFetcherRequest.newBuilder()
+ .setFetcherId("saved.fs")
+ .build());
+ Assertions.assertThrows(StatusRuntimeException.class, () ->
+ blockingStub.fetchAndParse(FetchAndParseRequest.newBuilder()
+ .setFetcherId("saved.fs")
+ .setFetchKey("moved.html")
+ .build()));
}
@Test
diff --git
a/tika-pipes/tika-pipes-config-store-ignite/src/main/java/org/apache/tika/pipes/ignite/IgniteConfigStore.java
b/tika-pipes/tika-pipes-config-store-ignite/src/main/java/org/apache/tika/pipes/ignite/IgniteConfigStore.java
index 9e7c2909d2..1985aa49b1 100644
---
a/tika-pipes/tika-pipes-config-store-ignite/src/main/java/org/apache/tika/pipes/ignite/IgniteConfigStore.java
+++
b/tika-pipes/tika-pipes-config-store-ignite/src/main/java/org/apache/tika/pipes/ignite/IgniteConfigStore.java
@@ -19,6 +19,7 @@ package org.apache.tika.pipes.ignite;
import java.util.HashSet;
import java.util.Locale;
import java.util.Set;
+import java.util.UUID;
import org.apache.ignite.Ignite;
import org.apache.ignite.table.KeyValueView;
@@ -45,6 +46,8 @@ public class IgniteConfigStore implements ConfigStore {
private static final Logger LOG =
LoggerFactory.getLogger(IgniteConfigStore.class);
private static final String DEFAULT_TABLE_NAME = "tika_config_store";
private static final String DEFAULT_INSTANCE_NAME =
"TikaIgniteConfigStore";
+ // '#' is outside ComponentIds.LEGAL_ID, so no component can collide with
this row
+ static final String CHANGE_TOKEN_KEY = "#change-token";
private Ignite ignite;
private KeyValueView<String, ExtensionConfigDTO> kvView;
@@ -120,6 +123,34 @@ public class IgniteConfigStore implements ConfigStore {
LOG.error("Failed to put config with id: {}", id, e);
throw new RuntimeException("Failed to put config", e);
}
+ touchChangeToken();
+ }
+
+ /**
+ * A random value rewritten after every put and remove, so every process
sharing the
+ * table sees each change; random rather than a counter, so writers need
no CAS.
+ */
+ @Override
+ public Object changeToken() {
+ checkInitialized();
+ try {
+ ExtensionConfigDTO dto = kvView.get(null, CHANGE_TOKEN_KEY);
+ return dto != null ? dto.getJson() : "";
+ } catch (Exception e) {
+ LOG.error("Failed to read the change token", e);
+ throw new RuntimeException("Failed to read the change token", e);
+ }
+ }
+
+ // after the change itself: a reader that sees the new token must see the
new config
+ private void touchChangeToken() {
+ try {
+ kvView.put(null, CHANGE_TOKEN_KEY,
+ new ExtensionConfigDTO(CHANGE_TOKEN_KEY,
UUID.randomUUID().toString()));
+ } catch (Exception e) {
+ LOG.error("Failed to write the change token", e);
+ throw new RuntimeException("Failed to write the change token", e);
+ }
}
@Override
@@ -155,6 +186,7 @@ public class IgniteConfigStore implements ConfigStore {
// by position: column lookup by name folds in the default
locale too
keys.add(resultSet.next().stringValue(0));
}
+ keys.remove(CHANGE_TOKEN_KEY);
return keys;
} catch (Exception e) {
LOG.error("Failed to get keySet", e);
@@ -164,29 +196,24 @@ public class IgniteConfigStore implements ConfigStore {
@Override
public int size() {
- checkInitialized();
- try {
- var resultSet = ignite.sql().execute(null, "SELECT COUNT(*) as cnt
FROM " + quotedName(tableName));
- if (resultSet.hasNext()) {
- return (int) resultSet.next().longValue(0);
- }
- return 0;
- } catch (Exception e) {
- LOG.error("Failed to get size", e);
- throw new RuntimeException("Failed to get size", e);
- }
+ return keySet().size();
}
@Override
public ExtensionConfig remove(String id) {
checkInitialized();
+ ExtensionConfigDTO removed;
try {
- ExtensionConfigDTO removed = kvView.getAndRemove(null, id);
- return removed != null ? new ExtensionConfig(id,
removed.getName(), removed.getJson()) : null;
+ removed = kvView.getAndRemove(null, id);
} catch (Exception e) {
LOG.error("Failed to remove config with id: {}", id, e);
throw new RuntimeException("Failed to remove config", e);
}
+ if (removed == null) {
+ return null;
+ }
+ touchChangeToken();
+ return new ExtensionConfig(id, removed.getName(), removed.getJson());
}
public void close() {
diff --git
a/tika-pipes/tika-pipes-config-store-ignite/src/test/java/org/apache/tika/pipes/ignite/IgniteConfigStoreTest.java
b/tika-pipes/tika-pipes-config-store-ignite/src/test/java/org/apache/tika/pipes/ignite/IgniteConfigStoreTest.java
index 2f2d6ce6b1..e91ef33ba3 100644
---
a/tika-pipes/tika-pipes-config-store-ignite/src/test/java/org/apache/tika/pipes/ignite/IgniteConfigStoreTest.java
+++
b/tika-pipes/tika-pipes-config-store-ignite/src/test/java/org/apache/tika/pipes/ignite/IgniteConfigStoreTest.java
@@ -18,6 +18,7 @@ package org.apache.tika.pipes.ignite;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -131,6 +132,32 @@ public class IgniteConfigStoreTest {
assertFalse(store.keySet().contains("id3"));
}
+ @Test
+ public void testChangeTokenSeenByAnotherClient() throws Exception {
+ IgniteConfigStore other = new IgniteConfigStore();
+ other.init();
+ try {
+ Object before = other.changeToken();
+ store.put("id1", new ExtensionConfig("id1", "type1", "{}"));
+ Object afterPut = other.changeToken();
+ assertNotEquals(before, afterPut);
+
+ store.put("id1", new ExtensionConfig("id1", "type1",
"{\"updated\":true}"));
+ Object afterUpdate = other.changeToken();
+ assertNotEquals(afterPut, afterUpdate);
+
+ store.remove("nonexistent");
+ assertEquals(afterUpdate, other.changeToken(), "a no-op remove is
not a change");
+
+ store.remove("id1");
+ assertNotEquals(afterUpdate, other.changeToken());
+ assertTrue(other.keySet().isEmpty(), "the token row is not a
component");
+ assertEquals(0, other.size());
+ } finally {
+ other.close();
+ }
+ }
+
@Test
public void testGetNonExistent() {
assertNull(store.get("nonexistent"));
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 4e576e6fcf..f17ede0b4b 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
@@ -23,6 +23,7 @@ import java.util.Iterator;
import java.util.LinkedHashSet;
import java.util.Locale;
import java.util.Map;
+import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;
@@ -70,6 +71,8 @@ public abstract class AbstractComponentManager<T extends
TikaExtension,
private final Map<String, Holder<T>> holders = new ConcurrentHashMap<>();
private final boolean allowRuntimeModifications;
private volatile boolean closed = false;
+ // the store state the cached runtime instances were built from
+ private volatile Object changeToken;
protected AbstractComponentManager(PluginManager pluginManager,
Map<String, ExtensionConfig>
componentConfigs,
@@ -88,6 +91,7 @@ public abstract class AbstractComponentManager<T extends
TikaExtension,
} catch (Exception e) {
throw new RuntimeException("Failed to initialize ConfigStore", e);
}
+ this.changeToken = configStore.changeToken();
this.configured = Map.copyOf(componentConfigs);
for (String id : configured.keySet()) {
if (configStore.containsKey(id)) {
@@ -254,6 +258,10 @@ public abstract class AbstractComponentManager<T extends
TikaExtension,
// 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);
+ // the store cannot change a config-file component, so its lookups
never depend on it
+ if (!configured.containsKey(id)) {
+ retireIfStoreChanged();
+ }
while (true) {
Holder<T> holder = holders.get(id);
if (holder == null) {
@@ -291,6 +299,36 @@ public abstract class AbstractComponentManager<T extends
TikaExtension,
return holder;
}
+ /**
+ * Another process (the gRPC server, another node) saved or deleted a
runtime component:
+ * retire every cached runtime instance so the next lease rebuilds from
the store.
+ * Config-file instances are untouched; the store cannot change them.
+ */
+ private void retireIfStoreChanged() throws IOException {
+ Object token;
+ try {
+ token = configStore.changeToken();
+ } catch (RuntimeException e) {
+ // fail this lookup, not the task's process: an unchecked
exception crashes a fork
+ throw new IOException("Can't check the config store for changes",
e);
+ }
+ if (Objects.equals(token, changeToken)) {
+ return;
+ }
+ synchronized (this) {
+ if (Objects.equals(token, changeToken)) {
+ return;
+ }
+ changeToken = token;
+ for (String id : Set.copyOf(holders.keySet())) {
+ if (!configured.containsKey(id)) {
+ retire(id);
+ }
+ }
+ }
+ LOG.debug("Config store changed; retired cached runtime {}s",
getComponentName());
+ }
+
private synchronized void retire(String id) {
Holder<T> holder = holders.remove(id);
if (holder != null) {
diff --git
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PipesConfig.java
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PipesConfig.java
index e7c79bb63b..5b4369de8a 100644
---
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PipesConfig.java
+++
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PipesConfig.java
@@ -177,7 +177,7 @@ public class PipesConfig {
/**
* Type of ConfigStore to use for distributed state management.
- * Options: "memory" (default), "ignite"
+ * Options: "memory" (default), "file", "ignite"
*/
private String configStoreType = "memory";
diff --git
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/config/ConfigStore.java
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/config/ConfigStore.java
index 5c90fa0b7f..c497218761 100644
---
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/config/ConfigStore.java
+++
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/config/ConfigStore.java
@@ -46,6 +46,17 @@ public interface ConfigStore extends TikaExtension {
// Default implementation does nothing
}
+ /**
+ * A value that changes whenever the stored configurations may have
changed, including
+ * changes made by another process; compared with {@code equals}. Runtime
components
+ * built from an older state are retired when it changes, so it must be
cheap: it is
+ * read on every component lookup. Null (the default) for a store no other
process
+ * shares, whose changes the owning manager already sees.
+ */
+ default Object changeToken() {
+ return null;
+ }
+
/**
* Stores a configuration.
*
diff --git
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/config/FileBasedConfigStore.java
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/config/FileBasedConfigStore.java
index b43b1bf2cd..a902ff9949 100644
---
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/config/FileBasedConfigStore.java
+++
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/config/FileBasedConfigStore.java
@@ -19,12 +19,16 @@ package org.apache.tika.pipes.core.config;
import java.io.IOException;
import java.nio.file.FileSystems;
import java.nio.file.Files;
+import java.nio.file.NoSuchFileException;
import java.nio.file.Path;
import java.nio.file.StandardCopyOption;
import java.nio.file.attribute.FileAttribute;
import java.nio.file.attribute.PosixFilePermission;
import java.nio.file.attribute.PosixFilePermissions;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
import java.util.EnumSet;
+import java.util.HexFormat;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
@@ -82,6 +86,26 @@ public class FileBasedConfigStore implements ConfigStore {
}
}
+ /**
+ * A digest of the file's bytes. Not mtime or the file key: coarse
timestamps and inode
+ * reuse can hide an edit that keeps the size, and the file is small and
already re-read
+ * on every lookup.
+ */
+ @Override
+ public Object changeToken() {
+ try {
+ return HexFormat.of().formatHex(
+
MessageDigest.getInstance("SHA-256").digest(Files.readAllBytes(configFile)));
+ } catch (NoSuchFileException e) {
+ return "";
+ } catch (IOException e) {
+ LOG.warn("Failed to read {}", configFile, e);
+ return "";
+ } catch (NoSuchAlgorithmException e) {
+ throw new IllegalStateException("SHA-256 not available", e);
+ }
+ }
+
@Override
public synchronized void put(String id, ExtensionConfig config) {
cache.put(id, config);
@@ -140,6 +164,8 @@ public class FileBasedConfigStore implements ConfigStore {
private synchronized void loadFromFile() throws IOException {
if (!Files.exists(configFile)) {
+ // the file is the shared state; a deleted file means an empty
store
+ cache.clear();
return;
}
diff --git
a/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/StorePropagationTest.java
b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/StorePropagationTest.java
new file mode 100644
index 0000000000..e0054edb1d
--- /dev/null
+++
b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/StorePropagationTest.java
@@ -0,0 +1,143 @@
+/*
+ * 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.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import org.apache.tika.pipes.api.fetcher.Fetcher;
+import org.apache.tika.pipes.api.fetcher.FetcherNotFoundException;
+import org.apache.tika.pipes.core.TrackingFetcherManager.TrackedFetcher;
+import org.apache.tika.pipes.core.config.FileBasedConfigStore;
+import org.apache.tika.pipes.core.config.InMemoryConfigStore;
+import org.apache.tika.plugins.ExtensionConfig;
+
+/**
+ * A save or delete in one process reaches another process's manager through a
shared file
+ * store: the gRPC server and a pipes fork, each with its own store instance
on one file.
+ */
+public class StorePropagationTest {
+
+ @TempDir
+ Path tmp;
+
+ private TrackingFetcherManager manager(String... configuredIds) throws
Exception {
+ FileBasedConfigStore store = new
FileBasedConfigStore(tmp.resolve("store.json"));
+ store.init();
+ return new TrackingFetcherManager(store, configuredIds);
+ }
+
+ @Test
+ public void testUpdateAndDeleteReachAnotherProcess() throws Exception {
+ TrackingFetcherManager parent = manager();
+ TrackingFetcherManager fork = manager();
+ parent.saveComponent(TrackingFetcherManager.config("f"));
+
+ Fetcher first;
+ try (Lease<Fetcher> lease = fork.acquire("f")) {
+ first = lease.get();
+ }
+ try (Lease<Fetcher> lease = fork.acquire("f")) {
+ assertSame(first, lease.get(), "no change, no rebuild");
+ }
+
+ // same length as the first config: the token must not depend on size
+ parent.saveComponent(new ExtensionConfig("f",
TrackingFetcherManager.TYPE, "{ }"));
+ try (Lease<Fetcher> lease = fork.acquire("f")) {
+ assertNotSame(first, lease.get());
+ assertEquals("{ }", lease.get().getExtensionConfig().json());
+ }
+ assertEquals(1, ((TrackedFetcher) first).getCloses());
+
+ parent.deleteComponent("f");
+ assertThrows(FetcherNotFoundException.class, () -> fork.acquire("f"));
+ assertEquals(1, fork.getBuilt().get(1).getCloses());
+ }
+
+ @Test
+ public void testHeldInstanceSurvivesARemoteUpdate() throws Exception {
+ TrackingFetcherManager parent = manager();
+ TrackingFetcherManager fork = manager();
+ parent.saveComponent(TrackingFetcherManager.config("f"));
+
+ Lease<Fetcher> inFlight = fork.acquire("f");
+ parent.saveComponent(new ExtensionConfig("f",
TrackingFetcherManager.TYPE, "{ }"));
+ try (Lease<Fetcher> next = fork.acquire("f")) {
+ assertNotSame(inFlight.get(), next.get(), "the next task gets the
update");
+ }
+ assertEquals(0, ((TrackedFetcher) inFlight.get()).getCloses(),
+ "closed under the task that still holds it");
+ inFlight.close();
+ assertEquals(1, ((TrackedFetcher) inFlight.get()).getCloses());
+ }
+
+ @Test
+ public void testConfigFileComponentNotRetired() throws Exception {
+ TrackingFetcherManager parent = manager("configured");
+ TrackingFetcherManager fork = manager("configured");
+ Fetcher configured;
+ try (Lease<Fetcher> lease = fork.acquire("configured")) {
+ configured = lease.get();
+ }
+ parent.saveComponent(TrackingFetcherManager.config("runtime"));
+ try (Lease<Fetcher> lease = fork.acquire("configured")) {
+ assertSame(configured, lease.get());
+ }
+ assertEquals(0, ((TrackedFetcher) configured).getCloses());
+ }
+
+ @Test
+ public void testUnreachableStoreFailsOnlyRuntimeLookups() throws Exception
{
+ boolean[] down = {false};
+ InMemoryConfigStore store = new InMemoryConfigStore() {
+ @Override
+ public Object changeToken() {
+ if (down[0]) {
+ throw new IllegalStateException("store unreachable");
+ }
+ return "t";
+ }
+ };
+ TrackingFetcherManager manager = new TrackingFetcherManager(store,
"configured");
+ manager.saveComponent(TrackingFetcherManager.config("runtime"));
+ down[0] = true;
+ // checked, so FetchHandler reports a fetcher error instead of
crashing the fork
+ assertThrows(IOException.class, () -> manager.acquire("runtime"));
+ manager.acquire("configured").close();
+ }
+
+ @Test
+ public void testDeletedStoreFileEmptiesTheStore() throws Exception {
+ TrackingFetcherManager parent = manager();
+ TrackingFetcherManager fork = manager();
+ parent.saveComponent(TrackingFetcherManager.config("f"));
+ fork.acquire("f").close();
+
+ Files.delete(tmp.resolve("store.json"));
+ assertThrows(FetcherNotFoundException.class, () -> fork.acquire("f"));
+ assertEquals(1, fork.getBuilt().get(0).getCloses());
+ }
+}
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
index fdf40054e1..a8a4dc6a38 100644
---
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
@@ -32,6 +32,7 @@ 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.ConfigStore;
import org.apache.tika.pipes.core.config.InMemoryConfigStore;
import org.apache.tika.plugins.AbstractTikaExtension;
import org.apache.tika.plugins.ExtensionConfig;
@@ -63,7 +64,11 @@ public class TrackingFetcherManager extends
AbstractComponentManager<Fetcher, Fe
};
public TrackingFetcherManager(String... configuredIds) {
- super(null, configs(configuredIds), true, new InMemoryConfigStore());
+ this(new InMemoryConfigStore(), configuredIds);
+ }
+
+ public TrackingFetcherManager(ConfigStore store, String... configuredIds) {
+ super(null, configs(configuredIds), true, store);
}
public static ExtensionConfig config(String id) {
