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 3f81aeec5f TIKA-4912: retry a 429/502/503/504 answer from a hosted
inference engine (#3204)
3f81aeec5f is described below
commit 3f81aeec5fb911da664c24677a1514aad8f83804
Author: Tim Allison <[email protected]>
AuthorDate: Fri Sep 18 16:20:53 2026 -0400
TIKA-4912: retry a 429/502/503/504 answer from a hosted inference engine
(#3204)
---
CHANGES.txt | 3 +
.../ROOT/pages/configuration/inference.adoc | 24 +++-
.../pages/configuration/parsers/vlm-parsers.adoc | 8 +-
.../java/org/apache/tika/http/TikaHttpClient.java | 138 +++++++++++++++++++--
.../org/apache/tika/http/TikaHttpClientTest.java | 102 +++++++++++++++
.../org/apache/tika/http/TikaTestHttpServer.java | 22 +++-
.../tika/inference/OpenAIEmbeddingEngine.java | 17 ++-
.../apache/tika/parser/vlm/AbstractVLMParser.java | 7 +-
.../org/apache/tika/parser/vlm/VLMOCRConfig.java | 19 ++-
9 files changed, 318 insertions(+), 22 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 7ed0b8a415..8bd6c9aed9 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,5 +1,8 @@
Release 4.1.0 - unreleased
+ * Calls to a hosted inference engine retry a 429, 502, 503 or 504 answer
+ with a jittered backoff (TIKA-4912).
+
* Handle inference exceptions more robustly in PDFParser (TIKA-4911).
* Add ffmpeg to tika-server's "full" Docker image (TIKA-4910).
diff --git a/docs/modules/ROOT/pages/configuration/inference.adoc
b/docs/modules/ROOT/pages/configuration/inference.adoc
index 8bc792c5d4..9947d2d834 100644
--- a/docs/modules/ROOT/pages/configuration/inference.adoc
+++ b/docs/modules/ROOT/pages/configuration/inference.adoc
@@ -50,8 +50,21 @@ different models. Secrets and paths live here and only here.
`openai-embedding-engine` posts to `/v1/embeddings` in the OpenAI shape and
sends up to
`maxBatchSize` images per request. Options: `baseUrl`, `model`, `apiKey`,
`timeoutMillis`
-(120000), `maxBatchSize` (32), `embeddingsPath`, `apiKeyHeaderName`,
`apiKeyPrefix`,
-`requestParameters`, `imageInput`, `mediaInput` (see <<media>>).
+(120000), `maxBatchSize` (32), `maxRetries` (4), `embeddingsPath`,
`apiKeyHeaderName`,
+`apiKeyPrefix`, `requestParameters`, `imageInput`, `mediaInput` (see
<<media>>).
+
+[#concurrency]
+=== Hosted engines and concurrency
+
+A hosted engine caps the requests in flight per key and answers the excess
with 429 at once,
+usually with no `Retry-After`; a saturated one answers 502, 503 or 504. Those
four answers are
+retried, `maxRetries` times (4 unless set, 0 to fail at once), with a short
jittered backoff
+from 250 ms doubling to 4 s, or the `Retry-After` the server sends, and never
past the budget
+the parse granted the call. Any other answer, a 401 or a 400, fails at once.
The retries cover
+collisions between a few parse workers; they do not turn more workers than the
key allows
+into throughput. Keep `numClients` (tika-server, Pipes) at or below the key's
concurrency,
+counting every engine the workers call, and raise the key's tier before
raising the workers.
+The VLM parsers take the same `maxRetries`.
=== Vendor keys: `requestParameters`
@@ -419,6 +432,13 @@ MEDIA bindings.
|Items per request across modalities; the segment cuts are 100-500 KB each as
base64, so
keep the request well under Jina's body limit (start at 8).
+|`maxRetries`
+|Jina's limit is concurrency per key, not requests per minute: 2 requests in
flight on a free
+key (1 on the OCR endpoint), 50 on a paid key, 500 on premium; the excess gets
a 429 at once
+with no `Retry-After`, and the limit clears as soon as an in-flight request
returns. The
+default 4 retries absorb the collisions of two or three workers; more workers
than the key's
+concurrency need a bigger key, not more retries (see <<concurrency>>).
+
|`requestParameters`
|`task`: `retrieval.passage` for indexing (default nothing is sent; set it),
`retrieval.query`
for queries, `text-matching`, `classification`, `separation`. `dimensions`: a
Matryoshka
diff --git a/docs/modules/ROOT/pages/configuration/parsers/vlm-parsers.adoc
b/docs/modules/ROOT/pages/configuration/parsers/vlm-parsers.adoc
index b9c394a510..a9e5e4e721 100644
--- a/docs/modules/ROOT/pages/configuration/parsers/vlm-parsers.adoc
+++ b/docs/modules/ROOT/pages/configuration/parsers/vlm-parsers.adoc
@@ -195,6 +195,12 @@ All three parsers share the same configuration POJO
(`VLMOCRConfig`):
|`120000`
|HTTP read timeout in milliseconds.
+|`maxRetries`
+|`4`
+|Retries of a 429, 502, 503 or 504 answer, with a jittered backoff inside the
parse budget;
+`0` fails at once. A hosted service caps requests in flight per key, so see
+xref:configuration/inference.adoc#concurrency[Hosted engines and concurrency].
+
|`apiKey`
|`""` (empty)
|API key. Format depends on the parser (Bearer header, x-api-key header,
@@ -290,7 +296,7 @@ rejects cost- and security-sensitive changes with an
exception:
|May be lowered, never raised above the initialization-time value — that
ceiling is what bounds
cost on a paid endpoint.
-|`maxImagePixels`, `allowRuntimePrompt`
+|`maxImagePixels`, `maxRetries`, `allowRuntimePrompt`
|Always blocked.
|`prompt`, `textRecognizer`
diff --git
a/tika-parsers/tika-http-jdk/src/main/java/org/apache/tika/http/TikaHttpClient.java
b/tika-parsers/tika-http-jdk/src/main/java/org/apache/tika/http/TikaHttpClient.java
index af431ac896..f15da1652d 100644
---
a/tika-parsers/tika-http-jdk/src/main/java/org/apache/tika/http/TikaHttpClient.java
+++
b/tika-parsers/tika-http-jdk/src/main/java/org/apache/tika/http/TikaHttpClient.java
@@ -25,11 +25,16 @@ import java.net.http.HttpResponse;
import java.net.http.HttpTimeoutException;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
+import java.time.ZonedDateTime;
+import java.time.format.DateTimeFormatter;
+import java.time.format.DateTimeParseException;
import java.util.Map;
+import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import org.apache.tika.config.ParseTimeout;
@@ -49,12 +54,28 @@ import org.apache.tika.parser.ParseContext;
* Obtain an instance via {@link #build(int)} and close it when done to release
* the underlying executor.
*
+ * <p>
+ * A hosted inference engine caps the requests in flight per key and answers
the excess with
+ * 429 at once, usually without a Retry-After; a saturated one answers 502,
503 or 504. Those
+ * four statuses are retried up to {@code maxRetries} times with a short
jittered backoff
+ * (250 ms doubling to 4 s), or the Retry-After the server sends, never
sleeping past the
+ * budget the parse granted the call. Any other status fails at once. The
retries cover
+ * collisions between a few parse workers, not a sustained excess of workers
over the key's
+ * concurrency.
* @since Apache Tika 4.0
*/
public class TikaHttpClient implements Closeable {
private static final String JSON_CONTENT_TYPE = "application/json;
charset=utf-8";
+ /** Retries of a 429/502/503/504 answer unless {@link #build(int, int)}
says otherwise. */
+ public static final int DEFAULT_MAX_RETRIES = 4;
+ private static final Set<Integer> RETRYABLE = Set.of(429, 502, 503, 504);
+ private static final long INITIAL_BACKOFF_MILLIS = 250;
+ private static final long MAX_BACKOFF_MILLIS = 4000;
+ /** A Retry-After longer than this is treated as "not now": the call fails
instead. */
+ private static final long MAX_RETRY_AFTER_MILLIS = 30_000;
+
// How often a bounded HTTP wait checkpoints the task's ParseTimeout -- see
// org.apache.tika.utils.ProcessUtils.HEARTBEAT_INTERVAL_MILLIS for the
same rationale.
private static final long HEARTBEAT_INTERVAL_MILLIS = 1000;
@@ -62,12 +83,14 @@ public class TikaHttpClient implements Closeable {
private final HttpClient httpClient;
private final ExecutorService executor;
private final int defaultTimeoutSeconds;
+ private final int maxRetries;
private TikaHttpClient(HttpClient httpClient, ExecutorService executor,
- int defaultTimeoutSeconds) {
+ int defaultTimeoutSeconds, int maxRetries) {
this.httpClient = httpClient;
this.executor = executor;
this.defaultTimeoutSeconds = defaultTimeoutSeconds;
+ this.maxRetries = maxRetries;
}
/**
@@ -76,6 +99,20 @@ public class TikaHttpClient implements Closeable {
* @param connectTimeoutSeconds TCP connection timeout in seconds
*/
public static TikaHttpClient build(int connectTimeoutSeconds) {
+ return build(connectTimeoutSeconds, DEFAULT_MAX_RETRIES);
+ }
+
+ /**
+ * Create a new {@code TikaHttpClient} with a daemon-thread executor.
+ *
+ * @param connectTimeoutSeconds TCP connection timeout in seconds
+ * @param maxRetries how many times a 429, 502, 503 or 504
answer is retried;
+ * 0 fails on the first one
+ */
+ public static TikaHttpClient build(int connectTimeoutSeconds, int
maxRetries) {
+ if (maxRetries < 0) {
+ throw new IllegalArgumentException("maxRetries must be >= 0, not "
+ maxRetries);
+ }
ExecutorService executor = Executors.newCachedThreadPool(r -> {
Thread t = new Thread(r, "tika-http-jdk");
t.setDaemon(true);
@@ -87,7 +124,7 @@ public class TikaHttpClient implements Closeable {
.followRedirects(HttpClient.Redirect.NORMAL)
.version(HttpClient.Version.HTTP_1_1)
.build();
- return new TikaHttpClient(client, executor, connectTimeoutSeconds);
+ return new TikaHttpClient(client, executor, connectTimeoutSeconds,
maxRetries);
}
/**
@@ -223,18 +260,49 @@ public class TikaHttpClient implements Closeable {
}
}
+ /**
+ * One budget for every attempt: each retry is sent with what is left of
+ * {@code grantedMillis}, and a backoff that would run past it fails the
call with the
+ * last answer instead.
+ */
private String send(HttpRequest request, ParseContext context, long
requestedMillis, long grantedMillis)
throws IOException, TikaException {
+ long startNanos = System.nanoTime();
+ for (int attempt = 0; ; attempt++) {
+ long remaining = grantedMillis - (System.nanoTime() - startNanos)
/ 1_000_000L;
+ if (remaining <= 0) {
+ throw new TikaTimeoutException("HTTP request to " +
request.uri() + " timed out",
+ requestedMillis, grantedMillis);
+ }
+ HttpRequest attemptRequest = attempt == 0 ? request
+ : HttpRequest.newBuilder(request, (name, value) -> true)
+ .timeout(Duration.ofMillis(remaining)).build();
+ HttpResponse<String> response = sendOnce(attemptRequest, context,
requestedMillis, remaining);
+ int status = response.statusCode();
+ if (status >= 200 && status < 300) {
+ return response.body();
+ }
+ TikaException failure = new TikaException("HTTP " + status
+ + " from " + request.uri() + ": " + response.body());
+ if (!RETRYABLE.contains(status) || attempt >= maxRetries) {
+ throw failure;
+ }
+ long delay = retryDelayMillis(response, attempt);
+ long left = grantedMillis - (System.nanoTime() - startNanos) /
1_000_000L;
+ if (delay < 0 || delay >= left) {
+ throw failure;
+ }
+ sleepWithHeartbeat(delay, context);
+ }
+ }
+
+ private HttpResponse<String> sendOnce(HttpRequest request, ParseContext
context,
+ long requestedMillis, long
grantedMillis)
+ throws IOException, TikaException {
CompletableFuture<HttpResponse<String>> future = httpClient.sendAsync(
request,
HttpResponse.BodyHandlers.ofString(StandardCharsets.UTF_8));
try {
- HttpResponse<String> response = waitWithHeartbeat(future, context,
grantedMillis,
- request.uri(), requestedMillis);
- if (response.statusCode() < 200 || response.statusCode() >= 300) {
- throw new TikaException("HTTP " + response.statusCode()
- + " from " + request.uri() + ": " + response.body());
- }
- return response.body();
+ return waitWithHeartbeat(future, context, grantedMillis,
request.uri(), requestedMillis);
} catch (ExecutionException e) {
Throwable cause = e.getCause();
if (cause instanceof HttpTimeoutException) {
@@ -254,6 +322,58 @@ public class TikaHttpClient implements Closeable {
}
}
+ /** The server's Retry-After (delta seconds or an HTTP date) when usable,
else backoff. */
+ static long retryDelayMillis(HttpResponse<String> response, int attempt) {
+ String retryAfter =
response.headers().firstValue("Retry-After").orElse(null);
+ if (retryAfter != null) {
+ long millis = parseRetryAfterMillis(retryAfter.trim());
+ if (millis >= 0) {
+ return millis > MAX_RETRY_AFTER_MILLIS ? -1 : millis;
+ }
+ }
+ return backoffMillis(attempt);
+ }
+
+ /** {@code 250 ms * 2^attempt}, capped at 4 s, with 25% jitter so workers
do not retry in step. */
+ static long backoffMillis(int attempt) {
+ long base = Math.min(MAX_BACKOFF_MILLIS, INITIAL_BACKOFF_MILLIS <<
Math.min(attempt, 20));
+ long jitter = base / 4;
+ return base - jitter + ThreadLocalRandom.current().nextLong(2 * jitter
+ 1);
+ }
+
+ /** Millis to wait, or -1 when the header is neither a delta nor an HTTP
date. */
+ static long parseRetryAfterMillis(String value) {
+ try {
+ return Long.parseLong(value) * 1000L;
+ } catch (NumberFormatException e) {
+ // an HTTP date, then
+ }
+ try {
+ ZonedDateTime at = ZonedDateTime.parse(value,
DateTimeFormatter.RFC_1123_DATE_TIME);
+ return Math.max(0, at.toInstant().toEpochMilli() -
System.currentTimeMillis());
+ } catch (DateTimeParseException e) {
+ return -1;
+ }
+ }
+
+ /** Sleeps in heartbeat slices so the stall detector sees the wait as
progress. */
+ private static void sleepWithHeartbeat(long millis, ParseContext context)
throws IOException {
+ long deadline = System.nanoTime() + millis * 1_000_000L;
+ try {
+ while (true) {
+ long left = (deadline - System.nanoTime()) / 1_000_000L;
+ if (left <= 0) {
+ return;
+ }
+ Thread.sleep(Math.min(left, HEARTBEAT_INTERVAL_MILLIS));
+ ParseTimeout.checkpoint(context);
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IOException("HTTP retry interrupted", e);
+ }
+ }
+
/**
* Waits for the future to complete, polling in up to {@value
#HEARTBEAT_INTERVAL_MILLIS}
* ms increments and checkpointing {@code context}'s {@link ParseTimeout}
on each
diff --git
a/tika-parsers/tika-http-jdk/src/test/java/org/apache/tika/http/TikaHttpClientTest.java
b/tika-parsers/tika-http-jdk/src/test/java/org/apache/tika/http/TikaHttpClientTest.java
index df0d34af08..a2a296fb6c 100644
---
a/tika-parsers/tika-http-jdk/src/test/java/org/apache/tika/http/TikaHttpClientTest.java
+++
b/tika-parsers/tika-http-jdk/src/test/java/org/apache/tika/http/TikaHttpClientTest.java
@@ -18,6 +18,7 @@ package org.apache.tika.http;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import java.util.Map;
@@ -26,6 +27,7 @@ import org.junit.jupiter.api.Test;
import org.apache.tika.config.ParseTimeout;
import org.apache.tika.config.TimeoutLimits;
+import org.apache.tika.exception.TikaException;
import org.apache.tika.exception.TikaTimeoutException;
import org.apache.tika.http.TikaTestHttpServer.MockResponse;
import org.apache.tika.parser.ParseContext;
@@ -182,4 +184,104 @@ public class TikaHttpClientTest {
"no HTTP request should have been attempted at all with a
0ms granted budget");
}
}
+
+ /** A concurrency-capped engine answers 429 at once with no Retry-After:
the retry succeeds. */
+ @Test
+ public void testRetriesA429ThenSucceeds() throws Exception {
+ try (TikaTestHttpServer server = new TikaTestHttpServer();
+ TikaHttpClient client = TikaHttpClient.build(30, 4)) {
+ server.enqueue(new MockResponse(429, "{\"detail\":\"Concurrency
limit exceeded\"}"));
+ server.enqueue(new MockResponse(429, "{\"detail\":\"Concurrency
limit exceeded\"}"));
+ server.enqueue(new MockResponse(200, "{\"ok\":true}"));
+ long start = System.currentTimeMillis();
+ String body = client.postJson(server.url(), "{}", Map.of(),
10_000, new ParseContext());
+ long elapsed = System.currentTimeMillis() - start;
+ assertEquals("{\"ok\":true}", body);
+ assertEquals(3, server.getRequestCount());
+ assertTrue(elapsed >= 187 + 375, "two backoffs, 250 and 500 ms
less jitter: " + elapsed);
+ }
+ }
+
+ @Test
+ public void testNonRetryableStatusFailsAtOnce() throws Exception {
+ try (TikaTestHttpServer server = new TikaTestHttpServer();
+ TikaHttpClient client = TikaHttpClient.build(30, 4)) {
+ server.enqueue(new MockResponse(401, "{\"detail\":\"Invalid API
key\"}"));
+ server.enqueue(new MockResponse(200, "{\"ok\":true}"));
+ TikaException e = assertThrows(TikaException.class,
+ () -> client.postJson(server.url(), "{}", Map.of(),
10_000, new ParseContext()));
+ assertTrue(e.getMessage().contains("HTTP 401"), e.getMessage());
+ assertEquals(1, server.getRequestCount());
+ }
+ }
+
+ @Test
+ public void testGivesUpAfterMaxRetriesWithTheLastAnswer() throws Exception
{
+ try (TikaTestHttpServer server = new TikaTestHttpServer();
+ TikaHttpClient client = TikaHttpClient.build(30, 2)) {
+ for (int i = 0; i < 4; i++) {
+ server.enqueue(new MockResponse(503, "{\"detail\":\"busy " + i
+ "\"}"));
+ }
+ TikaException e = assertThrows(TikaException.class,
+ () -> client.postJson(server.url(), "{}", Map.of(),
10_000, new ParseContext()));
+ assertTrue(e.getMessage().contains("busy 2"), "the third answer is
the one reported: " + e.getMessage());
+ assertEquals(3, server.getRequestCount(), "one attempt plus two
retries");
+ }
+ }
+
+ @Test
+ public void testZeroRetriesFailsOnTheFirst429() throws Exception {
+ try (TikaTestHttpServer server = new TikaTestHttpServer();
+ TikaHttpClient client = TikaHttpClient.build(30, 0)) {
+ server.enqueue(new MockResponse(429, "{}"));
+ server.enqueue(new MockResponse(200, "{}"));
+ assertThrows(TikaException.class,
+ () -> client.postJson(server.url(), "{}", Map.of(),
10_000, new ParseContext()));
+ assertEquals(1, server.getRequestCount());
+ }
+ }
+
+ @Test
+ public void testRetryAfterHeaderSetsTheDelay() throws Exception {
+ try (TikaTestHttpServer server = new TikaTestHttpServer();
+ TikaHttpClient client = TikaHttpClient.build(30, 4)) {
+ server.enqueue(new MockResponse(429, "{}", Map.of("Retry-After",
"1")));
+ server.enqueue(new MockResponse(200, "{\"ok\":true}"));
+ long start = System.currentTimeMillis();
+ client.postJson(server.url(), "{}", Map.of(), 10_000, new
ParseContext());
+ long elapsed = System.currentTimeMillis() - start;
+ assertTrue(elapsed >= 1000, "waited the second the server asked
for: " + elapsed);
+ assertEquals(2, server.getRequestCount());
+ }
+ }
+
+ /** A backoff that would run past the parse budget is not taken: the last
answer fails the call. */
+ @Test
+ public void testBackoffNeverSleepsPastTheBudget() throws Exception {
+ try (TikaTestHttpServer server = new TikaTestHttpServer();
+ TikaHttpClient client = TikaHttpClient.build(30, 4)) {
+ server.enqueue(new MockResponse(429, "{}", Map.of("Retry-After",
"5")));
+ server.enqueue(new MockResponse(200, "{}"));
+ long start = System.currentTimeMillis();
+ TikaException e = assertThrows(TikaException.class,
+ () -> client.postJson(server.url(), "{}", Map.of(), 800,
new ParseContext()));
+ long elapsed = System.currentTimeMillis() - start;
+ assertTrue(e.getMessage().contains("HTTP 429"), e.getMessage());
+ assertTrue(elapsed < 800, "failed without waiting out the budget:
" + elapsed);
+ assertEquals(1, server.getRequestCount());
+ }
+ }
+
+ @Test
+ public void testRetryAfterParsing() {
+ assertEquals(7000, TikaHttpClient.parseRetryAfterMillis("7"));
+ assertEquals(-1, TikaHttpClient.parseRetryAfterMillis("soon"));
+ assertEquals(0, TikaHttpClient.parseRetryAfterMillis("Wed, 21 Oct 2015
07:28:00 GMT"),
+ "a date in the past is now");
+ for (int attempt = 0; attempt < 6; attempt++) {
+ long b = TikaHttpClient.backoffMillis(attempt);
+ long base = Math.min(4000, 250L << attempt);
+ assertTrue(b >= base - base / 4 && b <= base + base / 4, attempt +
": " + b);
+ }
+ }
}
diff --git
a/tika-parsers/tika-http-jdk/src/test/java/org/apache/tika/http/TikaTestHttpServer.java
b/tika-parsers/tika-http-jdk/src/test/java/org/apache/tika/http/TikaTestHttpServer.java
index 9749c74b2c..f042bf21cd 100644
---
a/tika-parsers/tika-http-jdk/src/test/java/org/apache/tika/http/TikaTestHttpServer.java
+++
b/tika-parsers/tika-http-jdk/src/test/java/org/apache/tika/http/TikaTestHttpServer.java
@@ -62,13 +62,20 @@ public class TikaTestHttpServer implements Closeable {
* after headers have already arrived,
distinct
* from an overall-slow response
*/
- public record MockResponse(int status, String body, long delayMillis, long
bodyDelayAfterHeadersMillis) {
+ public record MockResponse(int status, String body, long delayMillis,
+ long bodyDelayAfterHeadersMillis, Map<String,
String> headers) {
public MockResponse(int status, String body) {
- this(status, body, 0, 0);
+ this(status, body, 0, 0, Map.of());
}
-
public MockResponse(int status, String body, long delayMillis) {
- this(status, body, delayMillis, 0);
+ this(status, body, delayMillis, 0, Map.of());
+ }
+ public MockResponse(int status, String body, long delayMillis, long
bodyDelayAfterHeadersMillis) {
+ this(status, body, delayMillis, bodyDelayAfterHeadersMillis,
Map.of());
+ }
+ /** Extra response headers, e.g. {@code Retry-After}. */
+ public MockResponse(int status, String body, Map<String, String>
headers) {
+ this(status, body, 0, 0, headers);
}
}
@@ -198,8 +205,11 @@ public class TikaTestHttpServer implements Closeable {
"HTTP/1.1 " + resp.status() + " " + statusText + "\r\n"
+ "Content-Type: application/json\r\n"
+ "Content-Length: " + responseBytes.length + "\r\n"
- + "Connection: close\r\n"
- + "\r\n";
+ + "Connection: close\r\n";
+ for (Map.Entry<String, String> h : resp.headers().entrySet()) {
+ responseHeaders += h.getKey() + ": " + h.getValue() + "\r\n";
+ }
+ responseHeaders += "\r\n";
out.write(responseHeaders.getBytes(StandardCharsets.US_ASCII));
out.flush();
if (resp.bodyDelayAfterHeadersMillis() > 0) {
diff --git
a/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/OpenAIEmbeddingEngine.java
b/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/OpenAIEmbeddingEngine.java
index 31451b9118..e4433ed98b 100644
---
a/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/OpenAIEmbeddingEngine.java
+++
b/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/OpenAIEmbeddingEngine.java
@@ -67,6 +67,7 @@ public class OpenAIEmbeddingEngine implements
EmbeddingEngine, Initializable {
private Map<String, Object> requestParameters = new LinkedHashMap<>();
private String imageInput = IMAGE_INPUT_OBJECT;
private String mediaInput = MEDIA_INPUT_NONE;
+ private int maxRetries = TikaHttpClient.DEFAULT_MAX_RETRIES;
public static final String MEDIA_INPUT_NONE = "none";
public static final String MEDIA_INPUT_OBJECT = "object";
@@ -85,6 +86,9 @@ public class OpenAIEmbeddingEngine implements
EmbeddingEngine, Initializable {
if (StringUtils.isBlank(baseUrl)) {
throw new TikaConfigException("openai-embedding-engine needs a
baseUrl");
}
+ if (maxRetries < 0) {
+ throw new TikaConfigException("maxRetries must be at least 0");
+ }
if (maxBatchSize < 1) {
throw new TikaConfigException("maxBatchSize must be at least 1");
}
@@ -107,7 +111,7 @@ public class OpenAIEmbeddingEngine implements
EmbeddingEngine, Initializable {
+ "must stay the default");
}
}
- httpClient = TikaHttpClient.build(30);
+ httpClient = TikaHttpClient.build(30, maxRetries);
}
@Override
@@ -175,7 +179,7 @@ public class OpenAIEmbeddingEngine implements
EmbeddingEngine, Initializable {
private List<float[]> post(ObjectNode root, int expected, ParseContext
context)
throws IOException, TikaException {
if (httpClient == null) {
- httpClient = TikaHttpClient.build(30);
+ httpClient = TikaHttpClient.build(30, maxRetries);
}
Map<String, String> headers = new HashMap<>();
if (!StringUtils.isBlank(apiKey)) {
@@ -267,6 +271,15 @@ public class OpenAIEmbeddingEngine implements
EmbeddingEngine, Initializable {
this.timeoutMillis = timeoutMillis;
}
+ public int getMaxRetries() {
+ return maxRetries;
+ }
+
+ /** Retries of a 429/502/503/504 answer, with backoff inside the parse
budget; 0 fails at once. */
+ public void setMaxRetries(int maxRetries) {
+ this.maxRetries = maxRetries;
+ }
+
@Override
public int getMaxBatchSize() {
return maxBatchSize;
diff --git
a/tika-parsers/tika-parsers-ml/tika-vlm/src/main/java/org/apache/tika/parser/vlm/AbstractVLMParser.java
b/tika-parsers/tika-parsers-ml/tika-vlm/src/main/java/org/apache/tika/parser/vlm/AbstractVLMParser.java
index e901ece43d..95690f4e02 100644
---
a/tika-parsers/tika-parsers-ml/tika-vlm/src/main/java/org/apache/tika/parser/vlm/AbstractVLMParser.java
+++
b/tika-parsers/tika-parsers-ml/tika-vlm/src/main/java/org/apache/tika/parser/vlm/AbstractVLMParser.java
@@ -380,7 +380,7 @@ public abstract class AbstractVLMParser implements Parser,
Initializable, Closea
}
private TikaHttpClient buildHttpClient() {
- return TikaHttpClient.build(30);
+ return TikaHttpClient.build(30, defaultConfig.getMaxRetries());
}
// ---- delegating config getters/setters --------------------------------
@@ -429,6 +429,11 @@ public abstract class AbstractVLMParser implements Parser,
Initializable, Closea
defaultConfig.setTimeoutMillis(timeoutMillis);
}
+ /** Retries of a 429/502/503/504 answer from the service; 0 fails at once.
*/
+ public void setMaxRetries(int maxRetries) {
+ defaultConfig.setMaxRetries(maxRetries);
+ }
+
public String getApiKey() {
return defaultConfig.getApiKey();
}
diff --git
a/tika-parsers/tika-parsers-ml/tika-vlm/src/main/java/org/apache/tika/parser/vlm/VLMOCRConfig.java
b/tika-parsers/tika-parsers-ml/tika-vlm/src/main/java/org/apache/tika/parser/vlm/VLMOCRConfig.java
index 2951411043..22b716be88 100644
---
a/tika-parsers/tika-parsers-ml/tika-vlm/src/main/java/org/apache/tika/parser/vlm/VLMOCRConfig.java
+++
b/tika-parsers/tika-parsers-ml/tika-vlm/src/main/java/org/apache/tika/parser/vlm/VLMOCRConfig.java
@@ -67,6 +67,9 @@ public class VLMOCRConfig implements Serializable {
*/
private long timeoutMillis = 120_000;
+ /** Retries of a 429/502/503/504 answer from the service; 0 fails at once.
*/
+ private int maxRetries = 4;
+
/** Optional API key for authenticated endpoints. Empty means no auth. */
private String apiKey = "";
@@ -163,6 +166,14 @@ public class VLMOCRConfig implements Serializable {
this.timeoutMillis = timeoutMillis;
}
+ public int getMaxRetries() {
+ return maxRetries;
+ }
+
+ public void setMaxRetries(int maxRetries) {
+ this.maxRetries = maxRetries;
+ }
+
public String getApiKey() {
return apiKey;
}
@@ -241,7 +252,7 @@ public class VLMOCRConfig implements Serializable {
* and cost-sensitive fields at parse time.
* <p>
* <b>Always blocked:</b> {@code baseUrl}, {@code apiKey}, {@code model},
- * {@code maxTokens}, {@code allowRuntimePrompt}.
+ * {@code maxTokens}, {@code maxRetries}, {@code allowRuntimePrompt}.
* <p>
* <b>Blocked by default (opt-in):</b> {@code prompt} — set
* {@code allowRuntimePrompt=true} at initialization time to permit
@@ -299,6 +310,12 @@ public class VLMOCRConfig implements Serializable {
+ "separate parser instance.");
}
+ @Override
+ public void setMaxRetries(int maxRetries) {
+ throw new IllegalStateException("Cannot modify maxRetries at
runtime. "
+ + "Retries are configured at initialization time.");
+ }
+
@Override
public void setPrompt(String prompt) {
if (!isAllowRuntimePrompt()) {
