gnodet-bot commented on code in PR #26601: URL: https://github.com/apache/camel/pull/26601#discussion_r4048104047
########## 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; + String outputFileId; + String errorFileId; + int total; + int completed; + int failed; Review Comment: ⚠️ **Data race on `StoredBatch` mutable fields (raised in previous review, not yet addressed)** `StoredBatch` exposes nine plain, non-`volatile` instance fields (`statusIndex`, `cancelling`, `cancelled`, `resultsBuilt`, `outputFileId`, `errorFileId`, `total`, `completed`, `failed`). These are mutated by `BatchRequestHandler` methods (`retrieveBatch`, `cancelBatch`, `maybeBuildResults`) which are dispatched by `com.sun.net.httpserver.HttpServer` — a thread-pool executor. A simultaneous retrieve + cancel from a polled test will data-race on `cancelling`/`cancelled`/`statusIndex` with no synchronization beyond the `ConcurrentHashMap` holding the batch reference (which only guarantees safe publication, not ongoing field visibility). In practice, tests run sequentially and the race window is narrow, but the mock is designed to be shared infrastructure. The fix is straightforward — either make all mutable fields `volatile`, or synchronize the `BatchRequestHandler` methods on the `StoredBatch` instance: ```suggestion volatile int statusIndex; volatile boolean cancelling; volatile boolean cancelled; volatile boolean resultsBuilt; volatile String outputFileId; volatile String errorFileId; volatile int total; volatile int completed; volatile int failed; ``` ########## 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; + String outputFileId; + String errorFileId; + int total; + int completed; + int failed; + Review Comment: ⚠️ **Data race on `StoredBatch` mutable fields** `BatchStore` wraps batch objects in a `ConcurrentHashMap`, but that only protects map-level operations. Once a `StoredBatch` reference is retrieved, concurrent `retrieveBatch()`, `cancelBatch()`, and `maybeBuildResults()` calls in `BatchRequestHandler` race on the same object's plain `int`/`boolean` fields. `com.sun.net.httpserver` uses a thread pool, so two parallel HTTP requests against the same batch can produce corrupted state. The most dangerous race is in `maybeBuildResults`: two concurrent retrieves that both see a batch reaching its final status will both read `resultsBuilt == false`, both set it to `true`, and both build output/error files — producing duplicate stored files and wrong counts. **Fix:** Either declare the mutable fields `volatile` and use an `AtomicBoolean` for `resultsBuilt`, or add a `synchronized (batch)` guard in the methods that mutate it. ```suggestion public static class StoredBatch { final String id; final String inputFileId; final String endpoint; final String metadata; volatile int statusIndex; volatile boolean cancelling; volatile boolean cancelled; volatile boolean resultsBuilt; volatile String outputFileId; volatile String errorFileId; volatile int total; volatile int completed; volatile int failed; ``` ########## 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` breaks on Windows-style line endings** An input file with `\r\n` line endings — which a `WrappedFile` body from a Windows filesystem, or a `String` body produced with `System.lineSeparator()` on Windows, will have — produces request lines with a trailing `\r`. `objectMapper.readTree(requestLine)` then returns a `custom_id` value like `"ticket-1\r"`, which matches no expectation, and every line lands in the error file instead of the output file. The multipart parser above already guards against this (it strips `\r\n` from part values); the JSONL reader should too. ```suggestion for (String requestLine : new String(input.content(), StandardCharsets.UTF_8).split("\\r?\\n")) { ``` -- 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]
