gnodet-bot commented on code in PR #26601: URL: https://github.com/apache/camel/pull/26601#discussion_r4048045884
########## test-infra/camel-test-infra-openai-mock/src/main/java/org/apache/camel/test/infra/openai/mock/BatchStore.java: ########## @@ -0,0 +1,128 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.test.infra.openai.mock; + +import java.nio.charset.StandardCharsets; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicInteger; + +/** + * In-memory state behind the Files and Batch APIs of the mock: the uploaded input files, the result files built for a + * finished batch, and the batches themselves. Tests can read the uploaded content back to assert on the JSONL a route + * produced. + */ +public class BatchStore { + + /** + * A file uploaded by a route or generated by the mock for a finished batch. + */ + public record StoredFile(String id, String filename, String purpose, byte[] content) { + } + + /** + * A created batch and the position it reached in the configured status progression. + */ + public static class StoredBatch { + final String id; + final String inputFileId; + final String endpoint; + final String metadata; + int statusIndex; + boolean cancelling; + boolean cancelled; + boolean resultsBuilt; Review Comment: ⚠️ **Data race on `StoredBatch` mutable fields** `StoredBatch` fields (`statusIndex`, `cancelling`, `cancelled`, `resultsBuilt`, `outputFileId`, `errorFileId`, etc.) are plain, non-`volatile` instance fields. `BatchStore` correctly uses a `ConcurrentHashMap` for the batch registry, but that only guards the lookup — once you have the `StoredBatch` reference, concurrent requests from multiple threads can race on its fields without any memory visibility guarantee. `HttpServer.create` without `setExecutor` uses `sun.net.httpserver.ServerImpl`'s default thread pool, which **is not single-threaded**. A concurrent polling route that fires two retrieve calls close together will have two threads writing `batch.statusIndex`, `batch.cancelling`, and `batch.cancelled` simultaneously with no synchronization. The mock is public test-infra API; other teams that use it with parallel routes will hit this. The simplest fix: make `retrieveBatch` and `cancelBatch` in `BatchRequestHandler` `synchronized` on `batch` (or make `StoredBatch` hold a `ReentrantLock`), or mark the mutated fields `volatile`. The minimal safe approach here is to synchronize on the `StoredBatch` instance in both handlers: ```java synchronized (batch) { if (batch.cancelling) { batch.cancelling = false; batch.cancelled = true; } else if (!batch.cancelled && batch.statusIndex < statusList().size() - 1) { batch.statusIndex++; } } ``` ########## components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIBatchResultsProducer.java: ########## @@ -0,0 +1,119 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.openai; + +import java.util.Optional; + +import com.openai.core.http.HttpResponse; +import com.openai.models.batches.Batch; +import org.apache.camel.CamelExchangeException; +import org.apache.camel.Exchange; +import org.apache.camel.component.ai.observability.GenAiErrorSupport; +import org.apache.camel.support.DefaultProducer; +import org.apache.camel.support.SynchronizationAdapter; +import org.apache.camel.util.ObjectHelper; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * OpenAI producer for the {@code batch-results} operation, which downloads the output or error file of a finished + * batch. + * <p> + * The body is the file as an {@link java.io.InputStream}, so a large result can be split line by line with the splitter + * in streaming mode. The HTTP response is closed when the exchange completes. Stream caching reads such a body into + * memory before the next step, so a route handling large results should turn it off. + */ +public class OpenAIBatchResultsProducer extends DefaultProducer { + + private static final Logger LOG = LoggerFactory.getLogger(OpenAIBatchResultsProducer.class); + private static final String ERROR_FILE = "error"; + + public OpenAIBatchResultsProducer(OpenAIEndpoint endpoint) { + super(endpoint); + } + + @Override + public OpenAIEndpoint getEndpoint() { + return (OpenAIEndpoint) super.getEndpoint(); + } + + @Override + public void process(Exchange exchange) throws Exception { + OpenAIConfiguration config = getEndpoint().getConfiguration(); + String batchId = exchange.getIn().getHeader(OpenAIConstants.BATCH_ID, String.class); + if (ObjectHelper.isEmpty(batchId)) { + throw new IllegalArgumentException( + "The " + OpenAIConstants.BATCH_ID + " header must hold the id of the batch to read the results of"); + } + + String resultsFile = exchange.getIn().getHeader(OpenAIConstants.BATCH_RESULTS_FILE, + config.getBatchResultsFile(), String.class); + if (!"output".equals(resultsFile) && !ERROR_FILE.equals(resultsFile)) { + throw new IllegalArgumentException( + "Unsupported batch results file: " + resultsFile + ". Supported: output, error"); + } + + Batch batch; + try { + batch = getEndpoint().getClient().batches().retrieve(batchId); + } catch (RuntimeException e) { + GenAiErrorSupport.apply(exchange, e); + throw e; + } + OpenAIBatchSupport.setBatchHeaders(exchange.getMessage(), batch); + + // the API sets the file ids once it has stopped processing, so their presence, not the status, says whether + // the results can be read; the status only explains why a file is missing + Optional<String> fileId = ERROR_FILE.equals(resultsFile) ? batch.errorFileId() : batch.outputFileId(); + if (fileId.isEmpty()) { + switch (batch.status().value()) { + case FAILED -> + // the input was rejected, so no result file exists: the reason is in the errors of the batch + throw new CamelExchangeException( + "Batch " + batchId + " failed: " + OpenAIBatchSupport.errors(batch), exchange); Review Comment: 🔍 **Misleading error message when `FAILED` batch has no errors populated** ```java case FAILED -> throw new CamelExchangeException( "Batch " + batchId + " failed: " + OpenAIBatchSupport.errors(batch), exchange); ``` `errors(batch)` returns an empty `List` when `batch.errors()` is absent or `data` is empty — the OpenAI API does not guarantee that validation errors are always populated on a `failed` batch (e.g. the batch failed at the file-parsing stage before individual errors are attributed). The message becomes `"Batch X failed: []"`, which tells the user nothing. Consider: ```java case FAILED -> { List<Map<String, Object>> errors = OpenAIBatchSupport.errors(batch); String detail = errors.isEmpty() ? "check the batch in the OpenAI dashboard" : errors.toString(); throw new CamelExchangeException("Batch " + batchId + " failed: " + detail, exchange); } ``` ########## test-infra/camel-test-infra-openai-mock/src/main/java/org/apache/camel/test/infra/openai/mock/BatchRequestHandler.java: ########## @@ -0,0 +1,404 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.test.infra.openai.mock; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.List; +import java.util.regex.Pattern; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.ObjectNode; +import com.sun.net.httpserver.HttpExchange; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Serves the Files and Batch APIs: uploading an input file, creating a batch from it, reporting its status and serving + * the result files. + * <p> + * A batch walks the configured status progression, one step per retrieve, so a route that polls sees the same sequence + * of states as it would against the real API. A cancel moves a running batch to {@code cancelling}, and the next + * retrieve to {@code cancelled}. Once a final status is reached the output and error files are built from the input + * file, matching each request line to a {@link BatchExpectation} by its {@code custom_id}. + */ +public class BatchRequestHandler { + private static final Logger LOG = LoggerFactory.getLogger(BatchRequestHandler.class); + + static final List<String> DEFAULT_STATUSES = List.of("validating", "in_progress", "finalizing", "completed"); + + private static final String DEFAULT_RESPONSE_BODY + = "{\"id\":\"chatcmpl-mock\",\"object\":\"chat.completion\",\"model\":\"openai-mock\"," + + "\"choices\":[{\"index\":0,\"message\":{\"role\":\"assistant\",\"content\":\"mock response\"}," + + "\"finish_reason\":\"stop\"}]}"; + + private final List<BatchExpectation> expectations; + private final List<String> statuses; + private final BatchStore store; + private final ObjectMapper objectMapper; + + public BatchRequestHandler(List<BatchExpectation> expectations, List<String> statuses, BatchStore store, + ObjectMapper objectMapper) { + this.expectations = expectations; + this.statuses = statuses; + this.store = store; + this.objectMapper = objectMapper; + } + + /** + * Whether this handler serves the given path, so the dispatcher can route Files and Batch calls here before the + * handlers of the inference APIs. + */ + public static boolean handles(String path) { + return path.endsWith("/files") || path.contains("/files/") + || path.endsWith("/batches") || path.contains("/batches/"); + } + + public void handleRequest(HttpExchange exchange) throws IOException { + String path = exchange.getRequestURI().getPath(); + String method = exchange.getRequestMethod(); + try { + if (path.endsWith("/files") && "POST".equalsIgnoreCase(method)) { + uploadFile(exchange); + } else if (path.endsWith("/content") && "GET".equalsIgnoreCase(method)) { + fileContent(exchange, idBefore(path, "/content")); + } else if (path.contains("/files/") && "GET".equalsIgnoreCase(method)) { + fileObject(exchange, lastSegment(path)); + } else if (path.contains("/files/") && "DELETE".equalsIgnoreCase(method)) { + deleteFile(exchange, lastSegment(path)); + } else if (path.endsWith("/batches") && "POST".equalsIgnoreCase(method)) { + createBatch(exchange); + } else if (path.endsWith("/cancel") && "POST".equalsIgnoreCase(method)) { + cancelBatch(exchange, idBefore(path, "/cancel")); + } else if (path.contains("/batches/") && "GET".equalsIgnoreCase(method)) { + retrieveBatch(exchange, lastSegment(path)); + } else { + sendError(exchange, 404, "invalid_request_error", "Unsupported request: " + method + " " + path); + } + } catch (Exception e) { + LOG.error("Error processing batch request {} {}", method, path, e); + sendError(exchange, 500, "internal_error", String.valueOf(e.getMessage())); + } + } + + private void uploadFile(HttpExchange exchange) throws IOException { + byte[] body; + try (InputStream is = exchange.getRequestBody()) { + body = is.readAllBytes(); + } + String contentType = exchange.getRequestHeaders().getFirst("Content-Type"); + String boundary = contentType != null && contentType.contains("boundary=") + ? contentType.substring(contentType.indexOf("boundary=") + "boundary=".length()).trim() : null; + if (boundary == null) { + sendError(exchange, 400, "invalid_request_error", "The file upload must be a multipart request"); + return; + } + + // ISO-8859-1 maps every byte to one character, so the file part survives the split unchanged + String raw = new String(body, StandardCharsets.ISO_8859_1); + String filename = "unknown"; + String purpose = "unknown"; + byte[] content = new byte[0]; + // the boundary may hold regex metacharacters, so it is quoted rather than spliced into the pattern + for (String part : raw.split(Pattern.quote("--" + boundary))) { + int headerEnd = part.indexOf("\r\n\r\n"); + if (headerEnd < 0) { + continue; + } + String headers = part.substring(0, headerEnd); + String value = part.substring(headerEnd + 4); + if (value.endsWith("\r\n")) { + value = value.substring(0, value.length() - 2); + } + if (headers.contains("name=\"file\"")) { + filename = between(headers, "filename=\"", "\""); + content = value.getBytes(StandardCharsets.ISO_8859_1); + } else if (headers.contains("name=\"purpose\"")) { + purpose = value.trim(); + } + } + + String id = store.addFile(filename, purpose, content); + LOG.debug("Stored uploaded file {} ({} bytes, purpose {})", id, content.length, purpose); + sendJson(exchange, 200, fileNode(id, filename, purpose, content.length).toString()); + } + + private void fileContent(HttpExchange exchange, String fileId) throws IOException { + byte[] content = store.getFileContent(fileId); + if (content == null) { + sendError(exchange, 404, "invalid_request_error", "No such file: " + fileId); + return; + } + exchange.getResponseHeaders().add("Content-Type", "application/octet-stream"); + exchange.sendResponseHeaders(200, content.length); + try (OutputStream os = exchange.getResponseBody()) { + os.write(content); + } + } + + private void fileObject(HttpExchange exchange, String fileId) throws IOException { + BatchStore.StoredFile file = store.file(fileId); + if (file == null) { + sendError(exchange, 404, "invalid_request_error", "No such file: " + fileId); + return; + } + sendJson(exchange, 200, fileNode(file.id(), file.filename(), file.purpose(), file.content().length).toString()); + } + + private void deleteFile(HttpExchange exchange, String fileId) throws IOException { + if (store.file(fileId) == null) { + sendError(exchange, 404, "invalid_request_error", "No such file: " + fileId); + return; + } + store.removeFile(fileId); + ObjectNode node = objectMapper.createObjectNode(); + node.put("id", fileId); + node.put("object", "file"); + node.put("deleted", true); + sendJson(exchange, 200, node.toString()); + } + + private void createBatch(HttpExchange exchange) throws IOException { + JsonNode request; + try (InputStream is = exchange.getRequestBody()) { + request = objectMapper.readTree(is.readAllBytes()); + } + String inputFileId = request.path("input_file_id").asText(); + if (store.file(inputFileId) == null) { + sendError(exchange, 404, "invalid_request_error", "No such file: " + inputFileId); + return; + } + JsonNode metadata = request.get("metadata"); + BatchStore.StoredBatch batch = store.addBatch(inputFileId, request.path("endpoint").asText(), + metadata != null && !metadata.isNull() ? metadata.toString() : null); + maybeBuildResults(batch); + LOG.debug("Created batch {} for input file {}", batch.id, inputFileId); + sendJson(exchange, 200, batchNode(batch).toString()); + } + + private void retrieveBatch(HttpExchange exchange, String batchId) throws IOException { + BatchStore.StoredBatch batch = store.batch(batchId); + if (batch == null) { + sendError(exchange, 404, "invalid_request_error", "No such batch: " + batchId); + return; + } + if (batch.cancelling) { + // a cancelled batch is final, so the retrieve after the cancel reports it as such, as the API does + batch.cancelling = false; + batch.cancelled = true; + } else if (!batch.cancelled && batch.statusIndex < statusList().size() - 1) { + batch.statusIndex++; + } + maybeBuildResults(batch); + sendJson(exchange, 200, batchNode(batch).toString()); + } + + private void cancelBatch(HttpExchange exchange, String batchId) throws IOException { + BatchStore.StoredBatch batch = store.batch(batchId); + if (batch == null) { + sendError(exchange, 404, "invalid_request_error", "No such batch: " + batchId); + return; + } + if (isFinal(status(batch))) { + sendError(exchange, 400, "invalid_request_error", + "Cannot cancel a batch with status '" + status(batch) + "'."); + return; + } + batch.cancelling = true; + sendJson(exchange, 200, batchNode(batch).toString()); + } + + private List<String> statusList() { + return statuses.isEmpty() ? DEFAULT_STATUSES : statuses; + } + + private String status(BatchStore.StoredBatch batch) { + if (batch.cancelled) { + return "cancelled"; + } + if (batch.cancelling) { + return "cancelling"; + } + return statusList().get(Math.min(batch.statusIndex, statusList().size() - 1)); + } + + private boolean isFinal(String status) { + return "completed".equals(status) || "expired".equals(status) || "cancelled".equals(status); + } + + /** + * Builds the output and error files once the batch reaches a final status, as the API does when it finishes + * processing the input file. + */ + private void maybeBuildResults(BatchStore.StoredBatch batch) throws IOException { + if (batch.resultsBuilt || !isFinal(status(batch))) { + return; + } + batch.resultsBuilt = true; + + BatchStore.StoredFile input = store.file(batch.inputFileId); + List<String> outputLines = new ArrayList<>(); + List<String> errorLines = new ArrayList<>(); + int line = 0; + for (String requestLine : new String(input.content(), StandardCharsets.UTF_8).split("\n")) { Review Comment: ⚠️ **JSONL split on `\n` will break on Windows** ```java for (String requestLine : new String(input.content(), StandardCharsets.UTF_8).split("\n")) { ``` The producer writes the temp file with `BufferedWriter.newLine()`, which uses the **platform line separator** — `\r\n` on Windows. If the mock runs on Windows (or the input was produced there), splitting on `\n` leaves a trailing `\r` in each line. `objectMapper.readTree(requestLine).path("custom_id").asText()` then returns `"ticket-1\r"` instead of `"ticket-1"`, no expectation matches, and all lines fall to the `mock_no_expectation` error path. Fix: ```suggestion for (String requestLine : new String(input.content(), StandardCharsets.UTF_8).split("\r?\n")) { ``` The output-file joins are `String.join("\n", ...)` (always `\n`), which is fine. Only the input split needs the fix. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
