This is an automated email from the ASF dual-hosted git repository. tballison pushed a commit to branch TIKA-4809-stage-9 in repository https://gitbox.apache.org/repos/asf/tika.git
commit ff3916cd69726212c1112879c950d5c724b38a11 Author: tallison <[email protected]> AuthorDate: Mon Aug 10 14:32:48 2026 -0400 TIKA-4809: Correct status mapping for caller errors, and add Retry-After --- .../tika/pipes/core/server/FetchHandler.java | 6 +++- .../apache/tika/server/core/TikaServerProcess.java | 2 +- .../tika/server/core/resource/AsyncResource.java | 26 +++++++++++----- .../server/core/resource/PipesParsingHelper.java | 35 ++++++++++++++++++++-- .../tika/server/core/resource/PipesResource.java | 10 +++++-- .../org/apache/tika/server/core/TikaPipesTest.java | 28 ++++++++++++++++- .../apache/tika/server/standard/TikaPipesTest.java | 2 +- 7 files changed, 92 insertions(+), 17 deletions(-) diff --git a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/FetchHandler.java b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/FetchHandler.java index c14ee24656..cdd28a5a7b 100644 --- a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/FetchHandler.java +++ b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/FetchHandler.java @@ -28,6 +28,7 @@ import org.apache.tika.parser.ParseContext; import org.apache.tika.pipes.api.FetchEmitTuple; import org.apache.tika.pipes.api.PipesResult; import org.apache.tika.pipes.api.fetcher.Fetcher; +import org.apache.tika.pipes.api.fetcher.FetcherNotFoundException; import org.apache.tika.pipes.core.fetcher.FetcherManager; import org.apache.tika.utils.ExceptionUtils; @@ -58,7 +59,10 @@ class FetchHandler { private FetcherOrResult getFetcher(FetchEmitTuple t) { try { return new FetcherOrResult(fetcherManager.getFetcher(t.getFetchKey().getFetcherId()), null); - } catch (IllegalArgumentException e) { + } catch (FetcherNotFoundException e) { + // Was IllegalArgumentException, which FetcherManager never throws -- so an unknown + // fetcher id fell through to the initialization branch below and every caller saw + // FETCHER_INITIALIZATION_EXCEPTION (a server-side 500) instead of FETCHER_NOT_FOUND. String noFetcherMsg = getNoFetcherMsg(t.getFetchKey().getFetcherId()); LOG.warn(noFetcherMsg); return new FetcherOrResult(null, new PipesResult(PipesResult.RESULT_STATUS.FETCHER_NOT_FOUND, noFetcherMsg)); diff --git a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/TikaServerProcess.java b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/TikaServerProcess.java index d51586aeed..eb5f5fe58b 100644 --- a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/TikaServerProcess.java +++ b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/TikaServerProcess.java @@ -431,7 +431,7 @@ public class TikaServerProcess { // not by PipesResource. PipesParsingHelper helper = tikaResource.getPipesParsingHelper(); resourceProviders.add(new SingletonResourceProvider( - new PipesResource(helper.getPipesParser()))); + new PipesResource(helper.getPipesParser(), helper.getPipesConfig()))); } resourceProviders.addAll(loadResourceServices(serverStatus)); return resourceProviders; diff --git a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/AsyncResource.java b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/AsyncResource.java index 4c0b24f353..f013f4eff4 100644 --- a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/AsyncResource.java +++ b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/AsyncResource.java @@ -24,6 +24,7 @@ import java.nio.charset.StandardCharsets; import java.util.HashMap; import java.util.Map; import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.TimeUnit; import jakarta.ws.rs.BadRequestException; import jakarta.ws.rs.POST; @@ -31,6 +32,7 @@ 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.apache.commons.lang3.StringUtils; import org.slf4j.Logger; @@ -96,7 +98,7 @@ public class AsyncResource { */ @POST @Produces("application/json") - public Map<String, Object> post(InputStream is, @Context HttpHeaders httpHeaders, @Context UriInfo info) throws Exception { + public Response post(InputStream is, @Context HttpHeaders httpHeaders, @Context UriInfo info) throws Exception { AsyncRequest request = deserializeASyncRequest(is); @@ -154,26 +156,36 @@ public class AsyncResource { } } - private Map<String, Object> ok(int size) { + private Response ok(int size) { Map<String, Object> map = new HashMap<>(); map.put("status", "ok"); map.put("added", size); - return map; + return Response.ok(map).build(); } - private Map<String, Object> throttle(int requestSize) { + /** + * 429, not 200. The queue was full for the whole {@code maxQueuePauseMs} wait, so nothing + * was accepted -- a 200 made a rejected batch indistinguishable from an accepted one to + * any client that checks status rather than parsing the body, and there are such clients. + */ + private Response throttle(int requestSize) { Map<String, Object> map = new HashMap<>(); map.put("status", "throttled"); map.put("msg", "not able to receive request of size " + requestSize + " at this time"); map.put("capacity", asyncProcessor.getCapacity()); - return map; + return Response + .status(Response.Status.TOO_MANY_REQUESTS) + .header(HttpHeaders.RETRY_AFTER, + Math.max(1, TimeUnit.MILLISECONDS.toSeconds(maxQueuePauseMs))) + .entity(map) + .build(); } - private Map<String, Object> badEmitter(String emitterName) { + private Response badEmitter(String emitterName) { throw new BadRequestException("can't find emitter for " + emitterName); } - private Map<String, Object> badFetcher(FetchKey fetchKey) { + private Response badFetcher(FetchKey fetchKey) { throw new BadRequestException("can't find fetcher for " + fetchKey.getFetcherId()); } diff --git a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/PipesParsingHelper.java b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/PipesParsingHelper.java index fe8c50f2b8..a667548c9a 100644 --- a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/PipesParsingHelper.java +++ b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/PipesParsingHelper.java @@ -23,10 +23,12 @@ import java.nio.file.Path; import java.util.Collections; import java.util.List; import java.util.UUID; +import java.util.concurrent.TimeUnit; import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.node.ObjectNode; import jakarta.ws.rs.WebApplicationException; +import jakarta.ws.rs.core.HttpHeaders; import jakarta.ws.rs.core.MediaType; import jakarta.ws.rs.core.Response; import org.slf4j.Logger; @@ -249,7 +251,7 @@ public class PipesParsingHelper { LOG.warn("Failed to serialize PipesResult error response as JSON; falling back to status-only body", e); json = "{\"status\":\"" + result.status().name() + "\"}"; } - return Response.status(mapStatusToHttpResponse(result.status())) + return responseBuilder(result.status(), pipesConfig.getMaxWaitForClientMillis()) .entity(json) .type(MediaType.APPLICATION_JSON) .build(); @@ -320,15 +322,42 @@ public class PipesParsingHelper { // (scale up numClients) apart from "a worker is actually crashing" (503). case CLIENT_UNAVAILABLE_WITHIN_MS -> Response.Status.TOO_MANY_REQUESTS; + // The caller named a fetcher/emitter this server does not have. Nothing failed + // on our side, and retrying the same request will never succeed -- 500 told + // clients to retry a request that is permanently malformed. + case FETCHER_NOT_FOUND, EMITTER_NOT_FOUND -> + Response.Status.BAD_REQUEST; + case PAYLOAD_LIMIT_EXCEEDED -> + Response.Status.REQUEST_ENTITY_TOO_LARGE; case FETCH_EXCEPTION, EMIT_EXCEPTION, - FETCHER_NOT_FOUND, EMITTER_NOT_FOUND, - PAYLOAD_LIMIT_EXCEEDED, FETCHER_INITIALIZATION_EXCEPTION, EMITTER_INITIALIZATION_EXCEPTION, FAILED_TO_INITIALIZE -> Response.Status.INTERNAL_SERVER_ERROR; }; } + /** A crashed child is replaced promptly; no point holding clients off for a minute. */ + private static final long CRASH_RETRY_AFTER_SECONDS = 5; + + /** + * Response builder for a pipes status, carrying {@code Retry-After} on the two families + * where the server knows the condition is transient. Without it, a client loop's only + * options are to give up or to hammer a pool that is already saturated. + */ + public static Response.ResponseBuilder responseBuilder(PipesResult.RESULT_STATUS status, + long maxWaitForClientMillis) { + Response.Status httpStatus = mapStatusToHttpResponse(status); + Response.ResponseBuilder builder = Response.status(httpStatus); + if (httpStatus == Response.Status.TOO_MANY_REQUESTS) { + // The pool was already full for this long, so a faster retry just re-queues. + builder.header(HttpHeaders.RETRY_AFTER, + Math.max(1, TimeUnit.MILLISECONDS.toSeconds(maxWaitForClientMillis))); + } else if (httpStatus == Response.Status.SERVICE_UNAVAILABLE) { + builder.header(HttpHeaders.RETRY_AFTER, CRASH_RETRY_AFTER_SECONDS); + } + return builder; + } + /** * Gets the PipesParser instance. */ 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 09956bbf5e..efb1c52c57 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 @@ -40,6 +40,7 @@ import org.apache.tika.pipes.api.FetchEmitTuple; import org.apache.tika.pipes.api.PipesResult; import org.apache.tika.pipes.core.EmitStrategy; import org.apache.tika.pipes.core.EmitStrategyConfig; +import org.apache.tika.pipes.core.PipesConfig; import org.apache.tika.pipes.core.PipesException; import org.apache.tika.pipes.core.PipesParser; import org.apache.tika.pipes.core.serialization.JsonFetchEmitTuple; @@ -52,15 +53,17 @@ public class PipesResource { private static final Logger LOG = LoggerFactory.getLogger(PipesResource.class); private final PipesParser pipesParser; + private final PipesConfig pipesConfig; /** * @param pipesParser shared parser, also used by /tika, /rmeta, and /unpack. * Lifecycle (construction, shutdown) is owned by whoever * built it, not by this class. - * vs. just the first line. + * @param pipesConfig the parser's config; read here for the {@code Retry-After} value. */ - public PipesResource(PipesParser pipesParser) { + public PipesResource(PipesParser pipesParser, PipesConfig pipesConfig) { this.pipesParser = pipesParser; + this.pipesConfig = pipesConfig; } @@ -123,7 +126,8 @@ public class PipesResource { // 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())) + return PipesParsingHelper + .responseBuilder(pipesResult.status(), pipesConfig.getMaxWaitForClientMillis()) .entity(body) .build(); } diff --git a/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/TikaPipesTest.java b/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/TikaPipesTest.java index 852916636c..8e1fda53a7 100644 --- a/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/TikaPipesTest.java +++ b/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/TikaPipesTest.java @@ -153,7 +153,7 @@ public class TikaPipesTest extends CXFTestBase { PipesConfig pipesConfig = PipesConfig.load(tikaJsonConfig); pipesConfig.setEmitStrategy(new EmitStrategyConfig(EmitStrategy.EMIT_ALL)); pipesParser = PipesParser.load(tikaJsonConfig, pipesConfig, tikaConfigPath); - pipesResource = new PipesResource(pipesParser); + pipesResource = new PipesResource(pipesParser, pipesConfig); rCoreProviders.add(new SingletonResourceProvider(pipesResource)); } catch (IOException | TikaConfigException e) { throw new RuntimeException(e); @@ -321,4 +321,30 @@ public class TikaPipesTest extends CXFTestBase { .asBoolean()); assertFalse(Files.isRegularFile(tmpNpeOutputFile)); } + /** + * A fetcher or emitter this server does not have is a permanently malformed request: + * retrying it will never succeed, so it must not come back as a 5xx "try again". + */ + @Test + public void testUnknownFetcherAndEmitterAre400() throws Exception { + FetchEmitTuple badFetcher = new FetchEmitTuple("badFetcher", + new FetchKey("no-such-fetcher", "hello_world.xml"), + new EmitKey(EMITTER_JSON_ID, ""), new Metadata()); + assertEquals(400, postTuple(badFetcher).getStatus()); + + FetchEmitTuple badEmitter = new FetchEmitTuple("badEmitter", + new FetchKey(FETCHER_ID, "hello_world.xml"), + new EmitKey("no-such-emitter", ""), new Metadata()); + assertEquals(400, postTuple(badEmitter).getStatus()); + } + + private Response postTuple(FetchEmitTuple t) throws Exception { + StringWriter writer = new StringWriter(); + JsonFetchEmitTuple.toJson(t, writer); + return WebClient + .create(endPoint + PIPES_PATH) + .accept("application/json") + .post(writer.toString()); + } + } diff --git a/tika-server/tika-server-standard/src/test/java/org/apache/tika/server/standard/TikaPipesTest.java b/tika-server/tika-server-standard/src/test/java/org/apache/tika/server/standard/TikaPipesTest.java index f8eb51fe27..d7440adefd 100644 --- a/tika-server/tika-server-standard/src/test/java/org/apache/tika/server/standard/TikaPipesTest.java +++ b/tika-server/tika-server-standard/src/test/java/org/apache/tika/server/standard/TikaPipesTest.java @@ -145,7 +145,7 @@ public class TikaPipesTest extends CXFTestBase { PipesConfig pipesConfig = PipesConfig.load(tikaJsonConfig); pipesConfig.setEmitStrategy(new EmitStrategyConfig(EmitStrategy.EMIT_ALL)); pipesParser = PipesParser.load(tikaJsonConfig, pipesConfig, tikaConfigPath); - pipesResource = new PipesResource(pipesParser); + pipesResource = new PipesResource(pipesParser, pipesConfig); rCoreProviders.add(new SingletonResourceProvider(pipesResource)); } catch (IOException | TikaConfigException e) { throw new RuntimeException(e);
