Skip to content

Commit 84dd882

Browse files
dpol1dpol1
authored
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 <dpol1@apache.org>
1 parent 787d9ff commit 84dd882

3 files changed

Lines changed: 76 additions & 0 deletions

File tree

‎CHANGES.txt‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,10 @@ Release 4.0.0 - ???
1111
* MagicDetector now compiles its regular expression once, in the
1212
constructor, instead of recompiling it on every match (TIKA-4796).
1313

14+
* tika-grpc's fetchAndParseServerSideStreaming now completes the call
15+
after delivering its reply, instead of leaving the client waiting
16+
for a terminal signal that never came (TIKA-4804).
17+
1418

1519
Release 4.0.0-beta-1 - 6/29/2026
1620

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

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -245,6 +245,7 @@ public void fetchAndParseServerSideStreaming(FetchAndParseRequest request,
245245
return;
246246
}
247247
fetchAndParseImpl(request, responseObserver);
248+
responseObserver.onCompleted();
248249
}
249250

250251
@Override

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

Lines changed: 71 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,7 @@
3838
import java.util.Map;
3939
import java.util.UUID;
4040
import java.util.concurrent.atomic.AtomicBoolean;
41+
import java.util.concurrent.atomic.AtomicInteger;
4142

4243
import com.asarkar.grpc.test.GrpcCleanupExtension;
4344
import com.asarkar.grpc.test.Resources;
@@ -377,6 +378,76 @@ private static TikaGrpc.TikaBlockingStub startServer(Resources resources, Path c
377378
return TikaGrpc.newBlockingStub(channel);
378379
}
379380

381+
/**
382+
* TIKA-4804: the server-streaming variant must close the call. With the in-process
383+
* transport and a direct executor the whole handler runs inside the stub call, so
384+
* the observer counts are final when it returns and nothing here needs to wait.
385+
*/
386+
@Test
387+
public void testServerSideStreamingSendsTerminalSignal(Resources resources) throws Exception {
388+
String serverName = InProcessServerBuilder.generateName();
389+
Server server = InProcessServerBuilder
390+
.forName(serverName)
391+
.directExecutor()
392+
.addService(new TikaGrpcServerImpl(tikaConfigUnlocked.toAbsolutePath().toString()))
393+
.build()
394+
.start();
395+
resources.register(server, Duration.ofSeconds(10));
396+
397+
ManagedChannel channel = InProcessChannelBuilder
398+
.forName(serverName)
399+
.directExecutor()
400+
.build();
401+
resources.register(channel, Duration.ofSeconds(10));
402+
TikaGrpc.TikaStub tikaStub = TikaGrpc.newStub(channel);
403+
404+
// The fetcher must come from the config file: one saved at runtime through
405+
// saveFetcher is not visible to the forked worker, and the fetch would fail.
406+
String fetcherId = createFetcherId(1);
407+
String fetchKey = "tika4804-" + UUID.randomUUID() + ".html";
408+
File testFile = new File("target", fetchKey);
409+
FileUtils.writeStringToFile(testFile,
410+
"<html><body>terminal signal</body></html>", StandardCharsets.UTF_8);
411+
412+
List<FetchAndParseReply> replies = Collections.synchronizedList(new ArrayList<>());
413+
AtomicInteger errors = new AtomicInteger();
414+
AtomicInteger completions = new AtomicInteger();
415+
StreamObserver<FetchAndParseReply> observer = new StreamObserver<>() {
416+
@Override
417+
public void onNext(FetchAndParseReply reply) {
418+
replies.add(reply);
419+
}
420+
421+
@Override
422+
public void onError(Throwable throwable) {
423+
errors.incrementAndGet();
424+
}
425+
426+
@Override
427+
public void onCompleted() {
428+
completions.incrementAndGet();
429+
}
430+
};
431+
432+
try {
433+
tikaStub.fetchAndParseServerSideStreaming(FetchAndParseRequest
434+
.newBuilder()
435+
.setFetcherId(fetcherId)
436+
.setFetchKey(fetchKey)
437+
.build(), observer);
438+
439+
assertEquals(1, replies.size(), "one reply for one fetch key");
440+
assertEquals(PipesResult.RESULT_STATUS.PARSE_SUCCESS.name(),
441+
replies.get(0).getStatus(),
442+
"the fixture must actually parse, or this test proves nothing");
443+
assertEquals(0, errors.get(), "no error on the happy path");
444+
assertEquals(1, completions.get(),
445+
"server streaming must send a terminal signal");
446+
} finally {
447+
FileUtils.deleteQuietly(testFile);
448+
}
449+
}
450+
380451
@Test
381452
public void testBiStream(Resources resources) throws Exception {
382453
String serverName = InProcessServerBuilder.generateName();

0 commit comments

Comments
 (0)