This is an automated email from the ASF dual-hosted git repository.
wenjin272 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-agents.git
The following commit(s) were added to refs/heads/main by this push:
new a146caab [integrations][openai] Let provider exceptions reach the
caller unwrapped (#989)
a146caab is described below
commit a146caab4b3cff783fb6ef3c6aad255d4e12bdf6
Author: Weiqing Yang <[email protected]>
AuthorDate: Mon Aug 17 00:53:54 2026 -0700
[integrations][openai] Let provider exceptions reach the caller unwrapped
(#989)
Generated-by: Claude Code 2.1.226
---
.../openai/AzureOpenAIChatModelConnection.java | 12 +---
.../openai/OpenAICompletionsConnection.java | 46 ++++++-------
.../openai/OpenAIResponsesModelConnection.java | 36 +++++-----
.../openai/AzureOpenAIChatModelConnectionTest.java | 34 ++++++++++
.../chatmodels/openai/FakeOpenAIErrorEndpoint.java | 76 ++++++++++++++++++++++
.../openai/OpenAICompletionsConnectionTest.java | 48 ++++++++++++++
.../openai/OpenAIResponsesModelConnectionTest.java | 75 ++++++++++++++++++++-
7 files changed, 271 insertions(+), 56 deletions(-)
diff --git
a/integrations/chat-models/openai/src/main/java/org/apache/flink/agents/integrations/chatmodels/openai/AzureOpenAIChatModelConnection.java
b/integrations/chat-models/openai/src/main/java/org/apache/flink/agents/integrations/chatmodels/openai/AzureOpenAIChatModelConnection.java
index 9f9267f1..db3270db 100644
---
a/integrations/chat-models/openai/src/main/java/org/apache/flink/agents/integrations/chatmodels/openai/AzureOpenAIChatModelConnection.java
+++
b/integrations/chat-models/openai/src/main/java/org/apache/flink/agents/integrations/chatmodels/openai/AzureOpenAIChatModelConnection.java
@@ -289,15 +289,9 @@ public class AzureOpenAIChatModelConnection extends
BaseChatModelConnection {
List<Tool> tools,
Map<String, Object> modelParams,
Object outputSchema) {
- try {
- ChatCompletionCreateParams params =
- buildRequest(messages, tools, modelParams, outputSchema);
- return toResponse(client.chat().completions().create(params),
modelParams);
- } catch (IllegalArgumentException e) {
- throw e;
- } catch (Exception e) {
- throw new RuntimeException("Failed to call Azure OpenAI chat
completions API.", e);
- }
+ ChatCompletionCreateParams params =
+ buildRequest(messages, tools, modelParams, outputSchema);
+ return toResponse(client.chat().completions().create(params),
modelParams);
}
// Package-private so response handling can be asserted against a
constructed completion without
diff --git
a/integrations/chat-models/openai/src/main/java/org/apache/flink/agents/integrations/chatmodels/openai/OpenAICompletionsConnection.java
b/integrations/chat-models/openai/src/main/java/org/apache/flink/agents/integrations/chatmodels/openai/OpenAICompletionsConnection.java
index f39c6ed3..3af78b72 100644
---
a/integrations/chat-models/openai/src/main/java/org/apache/flink/agents/integrations/chatmodels/openai/OpenAICompletionsConnection.java
+++
b/integrations/chat-models/openai/src/main/java/org/apache/flink/agents/integrations/chatmodels/openai/OpenAICompletionsConnection.java
@@ -191,33 +191,29 @@ public class OpenAICompletionsConnection extends
BaseChatModelConnection {
List<Tool> tools,
Map<String, Object> modelParams,
Object outputSchema) {
- try {
- ChatCompletionCreateParams params =
- buildRequest(messages, tools, modelParams, outputSchema);
- ChatCompletion completion =
client.chat().completions().create(params);
- ChatMessage response =
- OpenAIChatCompletionsUtils.convertFromOpenAIMessage(
- completion.choices().get(0).message());
-
- // Stash token usage
- if (completion.usage().isPresent()) {
- String modelName = modelParams != null ? (String)
modelParams.get("model") : null;
- if (modelName == null || modelName.isBlank()) {
- modelName = this.defaultModel;
- }
- if (modelName != null && !modelName.isBlank()) {
- response.getExtraArgs().put("model_name", modelName);
- response.getExtraArgs()
- .put("promptTokens",
completion.usage().get().promptTokens());
- response.getExtraArgs()
- .put("completionTokens",
completion.usage().get().completionTokens());
- }
+ ChatCompletionCreateParams params =
+ buildRequest(messages, tools, modelParams, outputSchema);
+ ChatCompletion completion = client.chat().completions().create(params);
+ ChatMessage response =
+ OpenAIChatCompletionsUtils.convertFromOpenAIMessage(
+ completion.choices().get(0).message());
+
+ // Stash token usage
+ if (completion.usage().isPresent()) {
+ String modelName = modelParams != null ? (String)
modelParams.get("model") : null;
+ if (modelName == null || modelName.isBlank()) {
+ modelName = this.defaultModel;
+ }
+ if (modelName != null && !modelName.isBlank()) {
+ response.getExtraArgs().put("model_name", modelName);
+ response.getExtraArgs()
+ .put("promptTokens",
completion.usage().get().promptTokens());
+ response.getExtraArgs()
+ .put("completionTokens",
completion.usage().get().completionTokens());
}
-
- return response;
- } catch (Exception e) {
- throw new RuntimeException("Failed to call OpenAI chat completions
API.", e);
}
+
+ return response;
}
// Package-private so the request body (including the native
response_format) can be asserted
diff --git
a/integrations/chat-models/openai/src/main/java/org/apache/flink/agents/integrations/chatmodels/openai/OpenAIResponsesModelConnection.java
b/integrations/chat-models/openai/src/main/java/org/apache/flink/agents/integrations/chatmodels/openai/OpenAIResponsesModelConnection.java
index 85d26003..5dca0b6a 100644
---
a/integrations/chat-models/openai/src/main/java/org/apache/flink/agents/integrations/chatmodels/openai/OpenAIResponsesModelConnection.java
+++
b/integrations/chat-models/openai/src/main/java/org/apache/flink/agents/integrations/chatmodels/openai/OpenAIResponsesModelConnection.java
@@ -129,28 +129,24 @@ public class OpenAIResponsesModelConnection extends
BaseChatModelConnection {
List<ChatMessage> messages,
List<org.apache.flink.agents.api.tools.Tool> tools,
Map<String, Object> modelParams) {
- try {
- ResponseCreateParams params = buildRequest(messages, tools,
modelParams);
- Response response = client.responses().create(params);
- ChatMessage result = convertResponse(response);
-
- if (response.usage().isPresent()) {
- String modelName = modelParams != null ? (String)
modelParams.get("model") : null;
- if (modelName == null || modelName.isBlank()) {
- modelName = this.defaultModel;
- }
- if (modelName != null && !modelName.isBlank()) {
- result.getExtraArgs().put("model_name", modelName);
- result.getExtraArgs().put("promptTokens",
response.usage().get().inputTokens());
- result.getExtraArgs()
- .put("completionTokens",
response.usage().get().outputTokens());
- }
+ ResponseCreateParams params = buildRequest(messages, tools,
modelParams);
+ Response response = client.responses().create(params);
+ ChatMessage result = convertResponse(response);
+
+ if (response.usage().isPresent()) {
+ String modelName = modelParams != null ? (String)
modelParams.get("model") : null;
+ if (modelName == null || modelName.isBlank()) {
+ modelName = this.defaultModel;
+ }
+ if (modelName != null && !modelName.isBlank()) {
+ result.getExtraArgs().put("model_name", modelName);
+ result.getExtraArgs().put("promptTokens",
response.usage().get().inputTokens());
+ result.getExtraArgs()
+ .put("completionTokens",
response.usage().get().outputTokens());
}
-
- return result;
- } catch (Exception e) {
- throw new RuntimeException("Failed to call OpenAI Responses API.",
e);
}
+
+ return result;
}
private ResponseCreateParams buildRequest(
diff --git
a/integrations/chat-models/openai/src/test/java/org/apache/flink/agents/integrations/chatmodels/openai/AzureOpenAIChatModelConnectionTest.java
b/integrations/chat-models/openai/src/test/java/org/apache/flink/agents/integrations/chatmodels/openai/AzureOpenAIChatModelConnectionTest.java
index 2bda0db2..6b7b2d3b 100644
---
a/integrations/chat-models/openai/src/test/java/org/apache/flink/agents/integrations/chatmodels/openai/AzureOpenAIChatModelConnectionTest.java
+++
b/integrations/chat-models/openai/src/test/java/org/apache/flink/agents/integrations/chatmodels/openai/AzureOpenAIChatModelConnectionTest.java
@@ -19,6 +19,7 @@
package org.apache.flink.agents.integrations.chatmodels.openai;
import com.fasterxml.jackson.core.type.TypeReference;
+import com.openai.errors.BadRequestException;
import com.openai.models.ChatModel;
import com.openai.models.ResponseFormatJsonSchema;
import com.openai.models.chat.completions.ChatCompletion;
@@ -43,6 +44,7 @@ import org.junit.jupiter.params.provider.MethodSource;
import org.junit.jupiter.params.provider.NullAndEmptySource;
import org.junit.jupiter.params.provider.ValueSource;
+import java.io.IOException;
import java.time.Duration;
import java.util.HashMap;
import java.util.List;
@@ -96,6 +98,18 @@ class AzureOpenAIChatModelConnectionTest {
return connection(CAPABLE_API_VERSION);
}
+ private static AzureOpenAIChatModelConnection connection(
+ String apiVersion, String azureEndpoint, String azureUrlPathMode) {
+ ResourceDescriptor desc =
+ connectionDescriptor()
+ .addInitialArgument("api_key", "test-key")
+ .addInitialArgument("api_version", apiVersion)
+ .addInitialArgument("azure_endpoint", azureEndpoint)
+ .addInitialArgument("azure_url_path_mode",
azureUrlPathMode)
+ .build();
+ return new AzureOpenAIChatModelConnection(desc, NOOP);
+ }
+
@Test
void testConnectionArgumentDefaultsAndZeroTimeout() {
AzureOpenAIChatModelConnection connection =
@@ -222,6 +236,26 @@ class AzureOpenAIChatModelConnectionTest {
.hasMessageContaining("temperature");
}
+ @Test
+ @DisplayName("A provider error reaches the caller as the SDK exception
carrying its payload")
+ void testProviderErrorPropagatesUnwrapped() throws IOException {
+ try (FakeOpenAIErrorEndpoint endpoint =
FakeOpenAIErrorEndpoint.rejectingWith400()) {
+ // A loopback endpoint is a custom gateway rather than an
*.openai.azure.com resource,
+ // so LEGACY is what builds the deployment-scoped Azure request
path against it.
+ AzureOpenAIChatModelConnection connection =
+ connection(CAPABLE_API_VERSION, endpoint.baseUrl(),
"LEGACY");
+
+ assertThatThrownBy(() -> connection.chat(userMessage(), List.of(),
params(null), null))
+ .isInstanceOfSatisfying(
+ BadRequestException.class,
+ e -> {
+ assertThat(e.statusCode()).isEqualTo(400);
+
assertThat(e.code()).contains(FakeOpenAIErrorEndpoint.ERROR_CODE);
+ })
+
.hasMessageContaining(FakeOpenAIErrorEndpoint.ERROR_MESSAGE);
+ }
+ }
+
@Test
@DisplayName("Native response_format json_schema strict applied for a POJO
on a capable model")
void testNativeAppliedForCapableDeploymentModel() {
diff --git
a/integrations/chat-models/openai/src/test/java/org/apache/flink/agents/integrations/chatmodels/openai/FakeOpenAIErrorEndpoint.java
b/integrations/chat-models/openai/src/test/java/org/apache/flink/agents/integrations/chatmodels/openai/FakeOpenAIErrorEndpoint.java
new file mode 100644
index 00000000..78608f49
--- /dev/null
+++
b/integrations/chat-models/openai/src/test/java/org/apache/flink/agents/integrations/chatmodels/openai/FakeOpenAIErrorEndpoint.java
@@ -0,0 +1,76 @@
+/*
+ * 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.flink.agents.integrations.chatmodels.openai;
+
+import com.sun.net.httpserver.HttpServer;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+
+/**
+ * A loopback HTTP endpoint that answers every request with the provider error
envelope OpenAI
+ * returns for a rejected request, so a connection's error path can be
exercised without a live API
+ * call. A 400 is chosen because the SDK does not retry it, which keeps the
exchange to a single
+ * request.
+ */
+final class FakeOpenAIErrorEndpoint implements AutoCloseable {
+
+ static final String ERROR_MESSAGE = "The requested model does not exist.";
+ static final String ERROR_CODE = "model_not_found";
+
+ private static final String ERROR_BODY =
+ "{\"error\":{\"message\":\""
+ + ERROR_MESSAGE
+ + "\",\"type\":\"invalid_request_error\","
+ + "\"param\":\"model\",\"code\":\""
+ + ERROR_CODE
+ + "\"}}";
+
+ private final HttpServer server;
+
+ private FakeOpenAIErrorEndpoint(HttpServer server) {
+ this.server = server;
+ }
+
+ static FakeOpenAIErrorEndpoint rejectingWith400() throws IOException {
+ HttpServer server = HttpServer.create(new
InetSocketAddress("127.0.0.1", 0), 0);
+ byte[] body = ERROR_BODY.getBytes(StandardCharsets.UTF_8);
+ server.createContext(
+ "/",
+ exchange -> {
+ exchange.getResponseHeaders().add("Content-Type",
"application/json");
+ exchange.sendResponseHeaders(400, body.length);
+ exchange.getResponseBody().write(body);
+ exchange.close();
+ });
+ server.setExecutor(null);
+ server.start();
+ return new FakeOpenAIErrorEndpoint(server);
+ }
+
+ String baseUrl() {
+ return "http://127.0.0.1:"; + server.getAddress().getPort();
+ }
+
+ @Override
+ public void close() {
+ server.stop(0);
+ }
+}
diff --git
a/integrations/chat-models/openai/src/test/java/org/apache/flink/agents/integrations/chatmodels/openai/OpenAICompletionsConnectionTest.java
b/integrations/chat-models/openai/src/test/java/org/apache/flink/agents/integrations/chatmodels/openai/OpenAICompletionsConnectionTest.java
index 1c5a4d15..269dee29 100644
---
a/integrations/chat-models/openai/src/test/java/org/apache/flink/agents/integrations/chatmodels/openai/OpenAICompletionsConnectionTest.java
+++
b/integrations/chat-models/openai/src/test/java/org/apache/flink/agents/integrations/chatmodels/openai/OpenAICompletionsConnectionTest.java
@@ -18,6 +18,7 @@
package org.apache.flink.agents.integrations.chatmodels.openai;
+import com.openai.errors.BadRequestException;
import com.openai.models.ResponseFormatJsonSchema;
import com.openai.models.chat.completions.ChatCompletionCreateParams;
import org.apache.flink.agents.api.chat.messages.ChatMessage;
@@ -33,6 +34,7 @@ import org.apache.flink.agents.api.tools.ToolType;
import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.Test;
+import java.io.IOException;
import java.time.Duration;
import java.util.HashMap;
import java.util.List;
@@ -65,6 +67,16 @@ class OpenAICompletionsConnectionTest {
return new OpenAICompletionsConnection(desc, NOOP);
}
+ private static OpenAICompletionsConnection connection(String apiBaseUrl) {
+ ResourceDescriptor desc =
+
ResourceDescriptor.Builder.newBuilder(OpenAICompletionsConnection.class.getName())
+ .addInitialArgument("api_key", "test-key")
+ .addInitialArgument("api_base_url", apiBaseUrl)
+ .addInitialArgument("model", "gpt-4o")
+ .build();
+ return new OpenAICompletionsConnection(desc, NOOP);
+ }
+
private static Map<String, Object> params(String model) {
Map<String, Object> params = new HashMap<>();
params.put("model", model);
@@ -99,6 +111,42 @@ class OpenAICompletionsConnectionTest {
OpenAIClientTestUtils.assertNoTimeoutConfigured(connection);
}
+ @Test
+ @DisplayName("A request-building failure reaches the caller as its own
type, not a wrapper")
+ void testRequestBuildingFailurePropagatesUnwrapped() {
+ List<ChatMessage> toolMessageWithoutExternalId =
+ List.of(new ChatMessage(MessageRole.TOOL, "result", Map.of()));
+
+ assertThatThrownBy(
+ () ->
+ connection()
+ .chat(
+ toolMessageWithoutExternalId,
+ List.of(),
+ params("gpt-4o"),
+ null))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("externalId");
+ }
+
+ @Test
+ @DisplayName("A provider error reaches the caller as the SDK exception
carrying its payload")
+ void testProviderErrorPropagatesUnwrapped() throws IOException {
+ try (FakeOpenAIErrorEndpoint endpoint =
FakeOpenAIErrorEndpoint.rejectingWith400()) {
+ OpenAICompletionsConnection connection =
connection(endpoint.baseUrl());
+
+ assertThatThrownBy(
+ () -> connection.chat(userMessage(), List.of(),
params("gpt-4o"), null))
+ .isInstanceOfSatisfying(
+ BadRequestException.class,
+ e -> {
+ assertThat(e.statusCode()).isEqualTo(400);
+
assertThat(e.code()).contains(FakeOpenAIErrorEndpoint.ERROR_CODE);
+ })
+
.hasMessageContaining(FakeOpenAIErrorEndpoint.ERROR_MESSAGE);
+ }
+ }
+
@Test
@DisplayName("Native response_format json_schema strict applied for a POJO
on a capable model")
void testNativeAppliedForPojoCapableModel() {
diff --git
a/integrations/chat-models/openai/src/test/java/org/apache/flink/agents/integrations/chatmodels/openai/OpenAIResponsesModelConnectionTest.java
b/integrations/chat-models/openai/src/test/java/org/apache/flink/agents/integrations/chatmodels/openai/OpenAIResponsesModelConnectionTest.java
index d5f0641c..dfb8d625 100644
---
a/integrations/chat-models/openai/src/test/java/org/apache/flink/agents/integrations/chatmodels/openai/OpenAIResponsesModelConnectionTest.java
+++
b/integrations/chat-models/openai/src/test/java/org/apache/flink/agents/integrations/chatmodels/openai/OpenAIResponsesModelConnectionTest.java
@@ -18,20 +18,27 @@
package org.apache.flink.agents.integrations.chatmodels.openai;
+import com.openai.errors.BadRequestException;
+import org.apache.flink.agents.api.chat.messages.ChatMessage;
+import org.apache.flink.agents.api.chat.messages.MessageRole;
import org.apache.flink.agents.api.chat.model.BaseChatModelConnection;
import org.apache.flink.agents.api.resource.ResourceContext;
import org.apache.flink.agents.api.resource.ResourceDescriptor;
import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.Test;
+import java.io.IOException;
import java.time.Duration;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
/**
- * Unit tests for {@link OpenAIResponsesModelConnection} — constructor
validation and default
- * resolution only, no network access.
+ * Unit tests for {@link OpenAIResponsesModelConnection} — constructor
validation, default
+ * resolution and error propagation, none of which need a live API call.
*/
class OpenAIResponsesModelConnectionTest {
@@ -42,6 +49,36 @@ class OpenAIResponsesModelConnectionTest {
OpenAIResponsesModelConnection.class.getName());
}
+ private static OpenAIResponsesModelConnection connection() {
+ ResourceDescriptor desc =
+ connectionDescriptor()
+ .addInitialArgument("api_key", "test-key")
+ .addInitialArgument("model", "gpt-4o")
+ .build();
+ return new OpenAIResponsesModelConnection(desc, NOOP);
+ }
+
+ private static OpenAIResponsesModelConnection connection(String
apiBaseUrl) {
+ ResourceDescriptor desc =
+ ResourceDescriptor.Builder.newBuilder(
+ OpenAIResponsesModelConnection.class.getName())
+ .addInitialArgument("api_key", "test-key")
+ .addInitialArgument("api_base_url", apiBaseUrl)
+ .addInitialArgument("model", "gpt-4o")
+ .build();
+ return new OpenAIResponsesModelConnection(desc, NOOP);
+ }
+
+ private static Map<String, Object> params(String model) {
+ Map<String, Object> params = new HashMap<>();
+ params.put("model", model);
+ return params;
+ }
+
+ private static List<ChatMessage> userMessage() {
+ return List.of(new ChatMessage(MessageRole.USER, "hi"));
+ }
+
@Test
@DisplayName("Constructor throws when api_key is missing")
void testConstructorMissingApiKey() {
@@ -201,4 +238,38 @@ class OpenAIResponsesModelConnectionTest {
OpenAIResponsesModelConnection conn = new
OpenAIResponsesModelConnection(desc, NOOP);
assertThat(conn.getMaxRetries()).isEqualTo(0);
}
+
+ @Test
+ @DisplayName("A request-building failure reaches the caller as its own
type, not a wrapper")
+ void testRequestBuildingFailurePropagatesUnwrapped() {
+ List<ChatMessage> toolMessageWithoutExternalId =
+ List.of(new ChatMessage(MessageRole.TOOL, "result", Map.of()));
+
+ assertThatThrownBy(
+ () ->
+ connection()
+ .chat(
+ toolMessageWithoutExternalId,
+ List.of(),
+ params("gpt-4o")))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("externalId");
+ }
+
+ @Test
+ @DisplayName("A provider error reaches the caller as the SDK exception
carrying its payload")
+ void testProviderErrorPropagatesUnwrapped() throws IOException {
+ try (FakeOpenAIErrorEndpoint endpoint =
FakeOpenAIErrorEndpoint.rejectingWith400()) {
+ OpenAIResponsesModelConnection connection =
connection(endpoint.baseUrl());
+
+ assertThatThrownBy(() -> connection.chat(userMessage(), List.of(),
params("gpt-4o")))
+ .isInstanceOfSatisfying(
+ BadRequestException.class,
+ e -> {
+ assertThat(e.statusCode()).isEqualTo(400);
+
assertThat(e.code()).contains(FakeOpenAIErrorEndpoint.ERROR_CODE);
+ })
+
.hasMessageContaining(FakeOpenAIErrorEndpoint.ERROR_MESSAGE);
+ }
+ }
}
