enhance input error handling

Signed-off-by: Jiaping Zeng <jpz@amazon.com>

Author: jiapingzeng

plugin/src/main/java/org/opensearch/ml/rest/RestMLExecuteStreamAction.java

@@ -217,17 +217,22 @@ public MLTaskResponse read(StreamInput in) throws IOException {
                         );
 
                     return Mono.fromCompletionStage(future);
-                } catch (IOException e) {
-                    return Mono.error(new OpenSearchStatusException("Failed to parse request", RestStatus.BAD_REQUEST, e));
+                } catch (Exception e) {
+                    log.error("Failed to parse or process request", e);
+                    return Mono.error(e);
                 }
-            }).doOnNext(channel::sendChunk).onErrorComplete(ex -> {
-                // Error handling
+            }).doOnNext(channel::sendChunk).onErrorResume(ex -> {
+                log.error("Error occurred", ex);
                 try {
-                    channel.sendResponse(new BytesRestResponse(channel, (Exception) ex));
-                    return true;
-                } catch (final IOException e) {
-                    throw new UncheckedIOException(e);
+                    String errorMessage = ex instanceof IOException 
+                        ? "Failed to parse request: " + ex.getMessage()
+                        : "Error processing request: " + ex.getMessage();
+                    HttpChunk errorChunk = createHttpChunk("data: {\"error\": \"" + errorMessage.replace("\"", "\\\"") + "\"}\n\n", true);
+                    channel.sendChunk(errorChunk);
+                } catch (Exception e) {
+                    log.error("Failed to send error chunk", e);
                 }
+                return Mono.empty();
             }).subscribe();
         };