You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tika.apache.org by ta...@apache.org on 2026/08/07 20:07:55 UTC
(tika) 04/09: TIKA-4809: Add backpressure status codes to /pipes
This is an automated email from the ASF dual-hosted git repository.
tballison pushed a commit to branch TIKA-4809-stage-2
in repository https://gitbox.apache.org/repos/asf/tika.git
commit 6c2cb3bc54d61e8554cb2bac3480f23f98c488d2
Author: tallison <ta...@apache.org>
AuthorDate: Fri Aug 7 11:48:08 2026 -0400
TIKA-4809: Add backpressure status codes to /pipes
---
.../tika/server/core/resource/PipesResource.java | 34 ++++++++++-------
.../core/TikaServerPipesIntegrationTest.java | 43 +++++++++++-----------
2 files changed, 43 insertions(+), 34 deletions(-)
diff --git a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/PipesResource.java b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/PipesResource.java
index 960935e910..e09b479be1 100644
--- a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/PipesResource.java
+++ b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/PipesResource.java
@@ -29,11 +29,11 @@ import jakarta.ws.rs.Path;
import jakarta.ws.rs.Produces;
import jakarta.ws.rs.core.Context;
import jakarta.ws.rs.core.HttpHeaders;
+import jakarta.ws.rs.core.Response;
import jakarta.ws.rs.core.UriInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.apache.tika.metadata.Metadata;
import org.apache.tika.metadata.TikaCoreProperties;
import org.apache.tika.parser.ParseContext;
import org.apache.tika.pipes.api.FetchEmitTuple;
@@ -76,12 +76,14 @@ public class PipesResource {
* Must specify a fetcherString and an emitter in the posted json.
*
* @param info uri info
- * @return InputStream that can be deserialized as a list of {@link Metadata} objects
+ * @return a JSON body describing the outcome (status/type, or a parse_exception),
+ * with HTTP status reflecting whether the process succeeded, crashed, or
+ * was unavailable within the configured wait
* @throws Exception
*/
@POST
@Produces("application/json")
- public Map<String, String> postRmeta(InputStream is, @Context HttpHeaders httpHeaders, @Context UriInfo info) throws Exception {
+ public Response postRmeta(InputStream is, @Context HttpHeaders httpHeaders, @Context UriInfo info) throws Exception {
FetchEmitTuple t = null;
try (Reader reader = new InputStreamReader(is, StandardCharsets.UTF_8)) {
t = JsonFetchEmitTuple.fromJson(reader);
@@ -91,7 +93,7 @@ public class PipesResource {
return processTuple(t);
}
- private Map<String, String> processTuple(FetchEmitTuple fetchEmitTuple) throws InterruptedException, PipesException, IOException {
+ private Response processTuple(FetchEmitTuple fetchEmitTuple) throws InterruptedException, PipesException, IOException {
// This parser is shared with /tika+/rmeta+/unpack, whose own default is
// PASSBACK_ALL. /pipes needs the child to emit via the client's configured
// emitter by default -- set EMIT_ALL explicitly per-request rather than
@@ -102,21 +104,27 @@ public class PipesResource {
parseContext.set(EmitStrategyConfig.class, new EmitStrategyConfig(EmitStrategy.EMIT_ALL));
}
PipesResult pipesResult = pipesParser.parse(fetchEmitTuple);
+ Map<String, String> body;
if (pipesResult.isProcessCrash()) {
- return returnProcessCrash(pipesResult.status().toString());
+ body = returnProcessCrash(pipesResult.status().toString());
} else if (!pipesResult.isSuccess()) {
// Handle fatal errors, initialization failures, and task exceptions
- return returnApplicationError(pipesResult
+ body = returnApplicationError(pipesResult
.status()
.toString());
+ } else {
+ body = switch (pipesResult.status()) {
+ case EMIT_SUCCESS_PARSE_EXCEPTION -> parseException(pipesResult.message(), true);
+ case PARSE_EXCEPTION_NO_EMIT -> parseException(pipesResult.message(), false);
+ default -> returnSuccess();
+ };
}
- switch (pipesResult.status()) {
- case EMIT_SUCCESS_PARSE_EXCEPTION:
- return parseException(pipesResult.message(), true);
- case PARSE_EXCEPTION_NO_EMIT:
- return parseException(pipesResult.message(), false);
- }
- return returnSuccess();
+ // Same status mapping /tika+/rmeta+/unpack use (PipesParsingHelper) -- e.g. 429 for
+ // CLIENT_UNAVAILABLE_WITHIN_MS, 503 for TIMEOUT/OOM/UNSPECIFIED_CRASH -- rather than
+ // always 200 with the failure only visible in the body.
+ return Response.status(PipesParsingHelper.mapStatusToHttpResponse(pipesResult.status()))
+ .entity(body)
+ .build();
}
private Map<String, String> parseException(String msg, boolean emitted) {
diff --git a/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/TikaServerPipesIntegrationTest.java b/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/TikaServerPipesIntegrationTest.java
index 01637dd308..675518069d 100644
--- a/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/TikaServerPipesIntegrationTest.java
+++ b/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/TikaServerPipesIntegrationTest.java
@@ -151,7 +151,7 @@ public class TikaServerPipesIntegrationTest extends IntegrationTestBase {
"-config", ProcessUtils.escapeCommandLine(TIKA_CONFIG
.toAbsolutePath()
.toString())});
- JsonNode node = testOne("system_exit.xml", false);
+ JsonNode node = testOne("system_exit.xml", false, FetchEmitTuple.ON_PARSE_EXCEPTION.EMIT, 503);
assertEquals("process_crash", node
.get("status")
.asText());
@@ -168,7 +168,7 @@ public class TikaServerPipesIntegrationTest extends IntegrationTestBase {
"-config", ProcessUtils.escapeCommandLine(TIKA_CONFIG
.toAbsolutePath()
.toString())});
- JsonNode node = testOne("fake_oom.xml", false);
+ JsonNode node = testOne("fake_oom.xml", false, FetchEmitTuple.ON_PARSE_EXCEPTION.EMIT, 503);
assertEquals("process_crash", node
.get("status")
.asText());
@@ -188,7 +188,7 @@ public class TikaServerPipesIntegrationTest extends IntegrationTestBase {
"-config", ProcessUtils.escapeCommandLine(TIKA_CONFIG_TIMEOUT
.toAbsolutePath()
.toString())});
- JsonNode node = testOne("heavy_hang_30000.xml", false);
+ JsonNode node = testOne("heavy_hang_30000.xml", false, FetchEmitTuple.ON_PARSE_EXCEPTION.EMIT, 503);
assertEquals("process_crash", node
.get("status")
.asText());
@@ -206,7 +206,7 @@ public class TikaServerPipesIntegrationTest extends IntegrationTestBase {
"-config", ProcessUtils.escapeCommandLine(TIKA_CONFIG
.toAbsolutePath()
.toString())});
- JsonNode node = testOneWithPerRequestTimeout("heavy_hang_30000.xml", 100);
+ JsonNode node = testOneWithPerRequestTimeout("heavy_hang_30000.xml", 100, 503);
assertEquals("process_crash", node
.get("status")
.asText());
@@ -215,17 +215,15 @@ public class TikaServerPipesIntegrationTest extends IntegrationTestBase {
.asText());
}
- private JsonNode testOneWithPerRequestTimeout(String fileName, long timeoutMillis) throws Exception {
+ private JsonNode testOneWithPerRequestTimeout(String fileName, long timeoutMillis, int expectedStatus) throws Exception {
awaitServerStartup();
Response response = WebClient
.create(endPoint + "/pipes")
.accept("application/json")
.post(getJsonStringWithTimeout(fileName, timeoutMillis));
- if (response.getStatus() == 200) {
- Reader reader = new InputStreamReader((InputStream) response.getEntity(), UTF_8);
- return new ObjectMapper().readTree(reader);
- }
- return null;
+ assertEquals(expectedStatus, response.getStatus());
+ Reader reader = new InputStreamReader((InputStream) response.getEntity(), UTF_8);
+ return new ObjectMapper().readTree(reader);
}
private String getJsonStringWithTimeout(String fileName, long timeoutMillis) throws IOException {
@@ -245,27 +243,30 @@ public class TikaServerPipesIntegrationTest extends IntegrationTestBase {
}
private JsonNode testOne(String fileName, boolean shouldFileExist) throws Exception {
- return testOne(fileName, shouldFileExist, FetchEmitTuple.ON_PARSE_EXCEPTION.EMIT);
+ return testOne(fileName, shouldFileExist, FetchEmitTuple.ON_PARSE_EXCEPTION.EMIT, 200);
}
private JsonNode testOne(String fileName, boolean shouldFileExist, FetchEmitTuple.ON_PARSE_EXCEPTION onParseException) throws Exception {
+ return testOne(fileName, shouldFileExist, onParseException, 200);
+ }
+
+ private JsonNode testOne(String fileName, boolean shouldFileExist,
+ FetchEmitTuple.ON_PARSE_EXCEPTION onParseException, int expectedStatus) throws Exception {
awaitServerStartup();
Response response = WebClient
.create(endPoint + "/pipes")
.accept("application/json")
.post(getJsonString(fileName, onParseException));
- if (response.getStatus() == 200) {
- Path targFile = TEMP_OUTPUT_DIR.resolve(fileName + ".json");
- if (shouldFileExist) {
- assertTrue(Files.size(targFile) > 1);
- } else {
- assertFalse(Files.isRegularFile(targFile));
- }
- Reader reader = new InputStreamReader((InputStream) response.getEntity(), UTF_8);
- return new ObjectMapper().readTree(reader);
+ assertEquals(expectedStatus, response.getStatus());
+ Path targFile = TEMP_OUTPUT_DIR.resolve(fileName + ".json");
+ if (shouldFileExist) {
+ assertTrue(Files.size(targFile) > 1);
+ } else {
+ assertFalse(Files.isRegularFile(targFile));
}
- return null;
+ Reader reader = new InputStreamReader((InputStream) response.getEntity(), UTF_8);
+ return new ObjectMapper().readTree(reader);
}
@Test