address comments

Author: prwhelan

x-pack/plugin/inference/src/main/java/org/elasticsearch/xpack/inference/external/openai/OpenAiUnifiedChatCompletionResponseHandler.java

@@ -68,11 +68,10 @@ protected Exception buildError(String message, Request request, HttpResult resul
     }
 
     private static Exception buildMidStreamError(Request request, String message, Exception e) {
-        var restStatus = RestStatus.OK;
         var errorResponse = OpenAiErrorResponse.fromString(message);
         if (errorResponse instanceof OpenAiErrorResponse oer) {
             return new UnifiedChatCompletionException(
-                restStatus,
+                RestStatus.INTERNAL_SERVER_ERROR,
                 format(
                     "%s for request from inference entity id [%s]. Error message: [%s]",
                     SERVER_ERROR_OBJECT,
@@ -87,7 +86,7 @@ private static Exception buildMidStreamError(Request request, String message, Ex
             return UnifiedChatCompletionException.fromThrowable(e);
         } else {
             return new UnifiedChatCompletionException(
-                restStatus,
+                RestStatus.INTERNAL_SERVER_ERROR,
                 format("%s for request from inference entity id [%s]", SERVER_ERROR_OBJECT, request.getInferenceEntityId()),
                 errorResponse != null ? errorResponse.getClass().getSimpleName() : "unknown",
                 "stream_error"

x-pack/plugin/inference/src/main/java/org/elasticsearch/xpack/inference/external/openai/OpenAiUnifiedStreamingProcessor.java

@@ -61,7 +61,7 @@ public class OpenAiUnifiedStreamingProcessor extends DelegatingProcessor<Deque<S
 
     private final BiFunction<String, Exception, Exception> errorParser;
     private final Deque<StreamingUnifiedChatCompletionResults.ChatCompletionChunk> buffer = new LinkedBlockingDeque<>();
-    private volatile boolean previousEventWasNotError = true;
+    private volatile boolean previousEventWasError = false;
 
     public OpenAiUnifiedStreamingProcessor(BiFunction<String, Exception, Exception> errorParser) {
         this.errorParser = errorParser;
@@ -83,19 +83,19 @@ protected void next(Deque<ServerSentEvent> item) throws Exception {
         var results = new ArrayDeque<StreamingUnifiedChatCompletionResults.ChatCompletionChunk>(item.size());
         for (var event : item) {
             if (ServerSentEventField.EVENT == event.name() && "error".equals(event.value())) {
-                previousEventWasNotError = false;
+                previousEventWasError = true;
             } else if (ServerSentEventField.DATA == event.name() && event.hasValue()) {
-                if (previousEventWasNotError) {
-                    try {
-                        var delta = parse(parserConfig, event);
-                        delta.forEachRemaining(results::offer);
-                    } catch (Exception e) {
-                        logger.warn("Failed to parse event from inference provider: {}", event);
-                        throw errorParser.apply(event.value(), e);
-                    }
-                } else {
+                if (previousEventWasError) {
                     throw errorParser.apply(event.value(), null);
                 }
+
+                try {
+                    var delta = parse(parserConfig, event);
+                    delta.forEachRemaining(results::offer);
+                } catch (Exception e) {
+                    logger.warn("Failed to parse event from inference provider: {}", event);
+                    throw errorParser.apply(event.value(), e);
+                }
             }
         }