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 84dd8827a2 TIKA-4804: complete the call in
fetchAndParseServerSideStreaming (#3008)
84dd8827a2 is described below
commit 84dd8827a234554cf9e503bdb583468c7379a6c9
Author: Davide Polato <[email protected]>
AuthorDate: Wed Aug 12 18:31:36 2026 +0200
TIKA-4804: complete the call in fetchAndParseServerSideStreaming (#3008)
The server-streaming handler delivered its reply and never called
onCompleted(), so clients waited for a terminal signal that never came.
One line, mirroring the unary handler, plus a regression test that
counts the observer events.
Signed-off-by: Davide Polato <[email protected]>
---
CHANGES.txt | 4 ++
.../apache/tika/pipes/grpc/TikaGrpcServerImpl.java | 1 +
.../apache/tika/pipes/grpc/TikaGrpcServerTest.java | 71 ++++++++++++++++++++++
3 files changed, 76 insertions(+)
diff --git a/CHANGES.txt b/CHANGES.txt
index e46059d1a1..0fd4471d4c 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -11,6 +11,10 @@ Release 4.0.0 - ???
* MagicDetector now compiles its regular expression once, in the
constructor, instead of recompiling it on every match (TIKA-4796).
+ * tika-grpc's fetchAndParseServerSideStreaming now completes the call
+ after delivering its reply, instead of leaving the client waiting
+ for a terminal signal that never came (TIKA-4804).
+
Release 4.0.0-beta-1 - 6/29/2026
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 7213217edb..8f63ee4767 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
@@ -245,6 +245,7 @@ class TikaGrpcServerImpl extends TikaGrpc.TikaImplBase {
return;
}
fetchAndParseImpl(request, responseObserver);
+ responseObserver.onCompleted();
}
@Override
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 2380735a65..2709f6f070 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
@@ -38,6 +38,7 @@ import java.util.Locale;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
import com.asarkar.grpc.test.GrpcCleanupExtension;
import com.asarkar.grpc.test.Resources;
@@ -377,6 +378,76 @@ public class TikaGrpcServerTest {
return TikaGrpc.newBlockingStub(channel);
}
+ /**
+ * TIKA-4804: the server-streaming variant must close the call. With the
in-process
+ * transport and a direct executor the whole handler runs inside the stub
call, so
+ * the observer counts are final when it returns and nothing here needs to
wait.
+ */
+ @Test
+ public void testServerSideStreamingSendsTerminalSignal(Resources
resources) throws Exception {
+ String serverName = InProcessServerBuilder.generateName();
+ Server server = InProcessServerBuilder
+ .forName(serverName)
+ .directExecutor()
+ .addService(new
TikaGrpcServerImpl(tikaConfigUnlocked.toAbsolutePath().toString()))
+ .build()
+ .start();
+ resources.register(server, Duration.ofSeconds(10));
+
+ ManagedChannel channel = InProcessChannelBuilder
+ .forName(serverName)
+ .directExecutor()
+ .build();
+ resources.register(channel, Duration.ofSeconds(10));
+ TikaGrpc.TikaStub tikaStub = TikaGrpc.newStub(channel);
+
+ // The fetcher must come from the config file: one saved at runtime
through
+ // saveFetcher is not visible to the forked worker, and the fetch
would fail.
+ String fetcherId = createFetcherId(1);
+ String fetchKey = "tika4804-" + UUID.randomUUID() + ".html";
+ File testFile = new File("target", fetchKey);
+ FileUtils.writeStringToFile(testFile,
+ "<html><body>terminal signal</body></html>",
StandardCharsets.UTF_8);
+
+ List<FetchAndParseReply> replies = Collections.synchronizedList(new
ArrayList<>());
+ AtomicInteger errors = new AtomicInteger();
+ AtomicInteger completions = new AtomicInteger();
+ StreamObserver<FetchAndParseReply> observer = new StreamObserver<>() {
+ @Override
+ public void onNext(FetchAndParseReply reply) {
+ replies.add(reply);
+ }
+
+ @Override
+ public void onError(Throwable throwable) {
+ errors.incrementAndGet();
+ }
+
+ @Override
+ public void onCompleted() {
+ completions.incrementAndGet();
+ }
+ };
+
+ try {
+ tikaStub.fetchAndParseServerSideStreaming(FetchAndParseRequest
+ .newBuilder()
+ .setFetcherId(fetcherId)
+ .setFetchKey(fetchKey)
+ .build(), observer);
+
+ assertEquals(1, replies.size(), "one reply for one fetch key");
+ assertEquals(PipesResult.RESULT_STATUS.PARSE_SUCCESS.name(),
+ replies.get(0).getStatus(),
+ "the fixture must actually parse, or this test proves
nothing");
+ assertEquals(0, errors.get(), "no error on the happy path");
+ assertEquals(1, completions.get(),
+ "server streaming must send a terminal signal");
+ } finally {
+ FileUtils.deleteQuietly(testFile);
+ }
+ }
+
@Test
public void testBiStream(Resources resources) throws Exception {
String serverName = InProcessServerBuilder.generateName();
