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 <[email protected]> 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
