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 90adae12bd TIKA-5010: tika-grpc reads parse-context timeout-limits and
closes interrupted calls with UNAVAILABLE (#3448)
90adae12bd is described below
commit 90adae12bd34fb8296b9ca0bdee7c399d0bdebde
Author: Tim Allison <[email protected]>
AuthorDate: Thu Oct 29 13:53:37 2026 -0400
TIKA-5010: tika-grpc reads parse-context timeout-limits and closes
interrupted calls with UNAVAILABLE (#3448)
---
CHANGES.txt | 7 ++
.../apache/tika/pipes/grpc/TikaGrpcServerImpl.java | 47 ++++++++----
.../apache/tika/pipes/grpc/TikaGrpcServerTest.java | 84 ++++++++++++++++++++++
3 files changed, 125 insertions(+), 13 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 2500a89594..69bc2cf552 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -22,6 +22,13 @@ Release 4.3.0 - ???
for EKS IAM roles for service accounts (IRSA); the plugin now bundles the
AWS STS
module it needs (TIKA-5006).
+ * tika-grpc applies parse-context.timeout-limits and exception-reporting in
the server
+ process too, so its backstop for a hung worker no longer falls back to
the built-in
+ timeouts. A FetchAndParse call interrupted before the parse finished
(server shutdown)
+ closes with UNAVAILABLE; a unary call used to fail with INTERNAL "no
value received",
+ a server-streaming call completed with no reply, and a bidirectional
stream went on
+ to throw on the closed call. Thanks to Kristian Rickert and Davide Polato
(TIKA-5010).
+
* The GCS fetcher, emitter and pipes iterator no longer require projectId.
When it is
unset, the client resolves a project from the environment, as it does for
its
credentials (TIKA-5009).
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 cc9f731460..9735759b06 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
@@ -118,11 +118,9 @@ class TikaGrpcServerImpl extends TikaGrpc.TikaImplBase {
TikaJsonConfig tikaJsonConfig = TikaJsonConfig.load(configPath);
- // Load PipesConfig directly from root level (not from "parse-context")
- pipesConfig = tikaJsonConfig.deserialize("pipes", PipesConfig.class);
- if (pipesConfig == null) {
- pipesConfig = new PipesConfig();
- }
+ // also reads parse-context.timeout-limits/exception-reporting for the
client's backstop
+ // and the results it fabricates itself (timeouts, crashes)
+ pipesConfig = PipesConfig.load(tikaJsonConfig);
pipesConfig.refuseTrustedRequests("tika-grpc");
// Security-sensitive grpc features (per-request config, runtime
component
@@ -285,25 +283,34 @@ class TikaGrpcServerImpl extends TikaGrpc.TikaImplBase {
if (parseContext == null) {
return;
}
- fetchAndParseImpl(request, parseContext, responseObserver);
- responseObserver.onCompleted();
+ if (fetchAndParseImpl(request, parseContext, responseObserver)) {
+ responseObserver.onCompleted();
+ }
}
@Override
public StreamObserver<FetchAndParseRequest>
fetchAndParseBiDirectionalStreaming(
StreamObserver<FetchAndParseReply> responseObserver) {
return new StreamObserver<>() {
+ // set once the call is closed with an error; later requests have
nowhere to go
+ private boolean closed;
+
@Override
public void onNext(FetchAndParseRequest fetchAndParseRequest) {
+ if (closed) {
+ return;
+ }
if (denyPerRequestConfig(fetchAndParseRequest,
responseObserver)) {
+ closed = true;
return;
}
ParseContext parseContext =
buildRequestParseContext(fetchAndParseRequest,
responseObserver);
if (parseContext == null) {
+ closed = true;
return;
}
- fetchAndParseImpl(fetchAndParseRequest, parseContext,
responseObserver);
+ closed = !fetchAndParseImpl(fetchAndParseRequest,
parseContext, responseObserver);
}
@Override
@@ -313,7 +320,9 @@ class TikaGrpcServerImpl extends TikaGrpc.TikaImplBase {
@Override
public void onCompleted() {
- responseObserver.onCompleted();
+ if (!closed) {
+ responseObserver.onCompleted();
+ }
}
};
}
@@ -328,8 +337,9 @@ class TikaGrpcServerImpl extends TikaGrpc.TikaImplBase {
if (parseContext == null) {
return;
}
- fetchAndParseImpl(request, parseContext, responseObserver);
- responseObserver.onCompleted();
+ if (fetchAndParseImpl(request, parseContext, responseObserver)) {
+ responseObserver.onCompleted();
+ }
}
/**
@@ -382,8 +392,12 @@ class TikaGrpcServerImpl extends TikaGrpc.TikaImplBase {
return parseContext;
}
- private void fetchAndParseImpl(FetchAndParseRequest request, ParseContext
parseContext,
- StreamObserver<FetchAndParseReply>
responseObserver) {
+ /**
+ * Sends the reply for one request. Returns {@code false} when the call
was closed with an
+ * error instead, so the caller must not complete it.
+ */
+ private boolean fetchAndParseImpl(FetchAndParseRequest request,
ParseContext parseContext,
+ StreamObserver<FetchAndParseReply>
responseObserver) {
String fetcherId;
try (Lease<Fetcher> lease =
fetcherManager.getFetcher(request.getFetcherId())) {
fetcherId = lease.get().getExtensionConfig().id();
@@ -413,10 +427,17 @@ class TikaGrpcServerImpl extends TikaGrpc.TikaImplBase {
}
}
responseObserver.onNext(fetchReplyBuilder.build());
+ return true;
} catch (IOException | PipesException e) {
throw new RuntimeException(e);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
+ // no outcome to reply with; completing a unary call without a
reply surfaces as INTERNAL
+ responseObserver.onError(io.grpc.Status.UNAVAILABLE
+ .withDescription("interrupted before the fetch and parse
of '"
+ + request.getFetchKey() + "' finished")
+ .asRuntimeException());
+ return false;
}
}
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 7dcb2d04d5..e4e28e9c55 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
@@ -74,6 +74,8 @@ import org.slf4j.LoggerFactory;
import org.apache.tika.metadata.TikaCoreProperties;
import org.apache.tika.pipes.api.PipesResult;
+import org.apache.tika.pipes.api.fetcher.Fetcher;
+import org.apache.tika.pipes.core.Lease;
import org.apache.tika.pipes.grpc.proto.DeleteFetcherReply;
import org.apache.tika.pipes.grpc.proto.DeleteFetcherRequest;
import org.apache.tika.pipes.grpc.proto.FetchAndParseReply;
@@ -795,4 +797,86 @@ public class TikaGrpcServerTest {
FileUtils.deleteDirectory(testDocumentFolder);
}
}
+
+ @Test
+ public void testParentReadsOperatorTimeoutLimits() throws Exception {
+ // the test config sets
parse-context.timeout-limits.progressTimeoutMillis
+ assertEquals(60000,
newService(tikaConfig).pipesConfig.getDefaultTimeoutLimits()
+ .getProgressTimeoutMillis());
+ }
+
+ @Test
+ public void testInterruptedFetchAndParseClosesWithUnavailable(Resources
resources)
+ throws Exception {
+ String serverName = InProcessServerBuilder.generateName();
+ TikaGrpcServerImpl serviceImpl = newService(tikaConfig);
+ Server server =
InProcessServerBuilder.forName(serverName).directExecutor()
+ .addService(serviceImpl).build().start();
+ resources.register(server, Duration.ofSeconds(10));
+ ManagedChannel channel =
InProcessChannelBuilder.forName(serverName).directExecutor().build();
+ resources.register(channel, Duration.ofSeconds(10));
+
+ String fetcherId = createFetcherId(1);
+ // build the fetcher up front, so only the wait for a pipes client
sees the interrupt
+ try (Lease<Fetcher> lease =
serviceImpl.fetcherManager.getFetcher(fetcherId)) {
+ assertNotNull(lease.get());
+ }
+ FetchAndParseRequest request = FetchAndParseRequest.newBuilder()
+ .setFetcherId(fetcherId)
+ .setFetchKey("interrupted.html")
+ .build();
+ TikaGrpc.TikaStub stub = TikaGrpc.newStub(channel);
+
+ RecordingObserver unary = new RecordingObserver();
+ RecordingObserver serverStreaming = new RecordingObserver();
+ RecordingObserver biDi = new RecordingObserver();
+ try {
+ Thread.currentThread().interrupt();
+ stub.fetchAndParse(request, unary);
+ Thread.currentThread().interrupt();
+ stub.fetchAndParseServerSideStreaming(request, serverStreaming);
+ // direct, not through a stub: a gRPC client drops its own sends
once the call is
+ // closed, but requests already queued at the server still arrive
+ Thread.currentThread().interrupt();
+ StreamObserver<FetchAndParseRequest> requests =
+ serviceImpl.fetchAndParseBiDirectionalStreaming(biDi);
+ requests.onNext(request);
+ Thread.interrupted();
+ requests.onNext(request);
+ requests.onCompleted();
+ } finally {
+ // the server restores the interrupt it caught; clear it before
cleanup waits
+ Thread.interrupted();
+ }
+
+ for (RecordingObserver observer : List.of(unary, serverStreaming,
biDi)) {
+ assertEquals(0, observer.replies.size(), "there is no outcome to
reply with");
+ assertEquals(0, observer.completions.get(), "a call closed with an
error must not complete");
+ assertEquals(1, observer.errors.size(), "the call must be closed
with one error");
+ assertEquals(Status.Code.UNAVAILABLE,
+ Status.fromThrowable(observer.errors.get(0)).getCode(),
+ String.valueOf(observer.errors.get(0)));
+ }
+ }
+
+ private static final class RecordingObserver implements
StreamObserver<FetchAndParseReply> {
+ private final List<FetchAndParseReply> replies =
Collections.synchronizedList(new ArrayList<>());
+ private final List<Throwable> errors =
Collections.synchronizedList(new ArrayList<>());
+ private final AtomicInteger completions = new AtomicInteger();
+
+ @Override
+ public void onNext(FetchAndParseReply reply) {
+ replies.add(reply);
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ errors.add(t);
+ }
+
+ @Override
+ public void onCompleted() {
+ completions.incrementAndGet();
+ }
+ }
}
