[pull] main from apache:main

code_review.md

@@ -41,7 +41,7 @@ Guides load on demand, so the general passes here stay short.
 |---|---|---|
 | `runtime/` state and recovery | serde and replay type fidelity, real failure-path tests | [review-guides/runtime-state-recovery.md](review-guides/runtime-state-recovery.md) |
 | Python-Java bridge | cross-language parity, type mapping across Pemja | [review-guides/python-java-bridge.md](review-guides/python-java-bridge.md) |
-| `api/` contract | API shape, compatibility policy, deprecation | planned |
+| `api/` contract | API shape, compatibility policy, deprecation | [review-guides/api-contract.md](review-guides/api-contract.md) |
 | `dist` and dependency | shading, LICENSE and NOTICE, dist registration | planned |
 | docs-only | facts match their source of truth | [review-guides/docs-only.md](review-guides/docs-only.md) |
 

docs/content/docs/development/chat_models.md

@@ -508,7 +508,7 @@ Azure OpenAI provides access to OpenAI models (GPT-4, GPT-4o, etc.) through Azur
 | `api_key` | str | Required | Azure OpenAI API key for authentication |
 | `api_version` | str | Required | Azure OpenAI REST API version (e.g., "2024-10-21"). See [API versions](https://learn.microsoft.com/en-us/azure/ai-services/openai/reference#rest-api-versioning) |
 | `azure_endpoint` | str | Required | Azure OpenAI endpoint URL (e.g., `https://{resource-name}.openai.azure.com`) |
-| `timeout` | float | `60.0` | API request timeout in seconds |
+| `timeout` | float | `60.0` | API request timeout in seconds; `0` disables timeouts |
 | `max_retries` | int | `3` | Maximum number of API retry attempts |
 
 {{< /tab >}}
@@ -520,8 +520,8 @@ Azure OpenAI provides access to OpenAI models (GPT-4, GPT-4o, etc.) through Azur
 | `api_key` | String | Required | Azure OpenAI API key for authentication |
 | `api_version` | String | Required | Azure OpenAI REST API version (e.g., "2024-10-21"). See [API versions](https://learn.microsoft.com/en-us/azure/ai-services/openai/reference#rest-api-versioning) |
 | `azure_endpoint` | String | Required | Azure OpenAI endpoint URL (e.g., `https://{resource-name}.openai.azure.com`) — either a direct Azure resource or a proxy/gateway URL that fronts an Azure OpenAI service |
-| `timeout` | int | None | Timeout in seconds for API requests; must be greater than 0, otherwise ignored (SDK default applies) |
-| `max_retries` | int | None | Maximum number of API retry attempts; must be non-negative, otherwise ignored (SDK default applies) |
+| `timeout` | float | `60` | Timeout in seconds for API requests; `0` disables timeouts; must be 0–2,147,483.647 |
+| `max_retries` | int | `3` | Maximum number of API retry attempts; must be non-negative |
 | `azure_url_path_mode` | String | `"AUTO"` | Controls how the SDK constructs Azure OpenAI request URLs. One of `"AUTO"`, `"LEGACY"`, or `"UNIFIED"`. Custom gateways that proxy Azure OpenAI typically need `"LEGACY"` to force the `/openai/deployments/{model}` path |
 
 {{< /tab >}}
@@ -911,7 +911,7 @@ OpenAI provides cloud-based chat models with state-of-the-art performance for a
 | `api_key` | str | Required | OpenAI API key for authentication |
 | `api_base_url` | str | `"https://api.openai.com/v1"` | Base URL for OpenAI API |
 | `max_retries` | int | `3` | Maximum number of API retry attempts |
-| `timeout` | float | `60.0` | API request timeout in seconds |
+| `timeout` | float | `60.0` | API request timeout in seconds; `0` disables timeouts |
 | `default_headers` | dict | None | Default headers for API requests |
 | `reuse_client` | bool | `True` | Whether to reuse the OpenAI client between requests |
 
@@ -923,8 +923,8 @@ OpenAI provides cloud-based chat models with state-of-the-art performance for a
 |-----------|------|---------|-------------|
 | `api_key` | String | Required | OpenAI API key for authentication |
 | `api_base_url` | String | `"https://api.openai.com/v1"` | Base URL for OpenAI API |
-| `max_retries` | int | `2` | Maximum number of API retry attempts |
-| `timeout` | int | None | Timeout in seconds for API requests |
+| `max_retries` | int | `3` | Maximum number of API retry attempts; must be non-negative |
+| `timeout` | float | `60` | Timeout in seconds for API requests; `0` disables timeouts; must be 0–2,147,483.647 |
 | `default_headers` | Map<String, String> | None | Default headers for API requests |
 | `model` | String | None | Default model to use if not specified in setup |
 
@@ -1054,8 +1054,8 @@ Responses API is only supported in Java currently. To use OpenAI Responses API f
 |-----------|------|---------|-------------|
 | `api_key` | String | Required | OpenAI API key for authentication |
 | `api_base_url` | String | None | Base URL for OpenAI API (useful for proxies) |
-| `max_retries` | int | `2` | Maximum number of API retry attempts |
-| `timeout` | int | None | Timeout in seconds for API requests |
+| `max_retries` | int | `3` | Maximum number of API retry attempts; must be non-negative |
+| `timeout` | float | `60` | Timeout in seconds for API requests; `0` disables timeouts; must be 0–2,147,483.647 |
 | `default_headers` | Map<String, String> | None | Default headers for API requests |
 | `model` | String | None | Default model to use if not specified in setup |
 

integrations/chat-models/openai/src/main/java/org/apache/flink/agents/integrations/chatmodels/openai/AzureOpenAIChatModelConnection.java

@@ -141,6 +141,8 @@ public class AzureOpenAIChatModelConnection extends BaseChatModelConnection {
     private static final Pattern API_VERSION_DATE_PREFIX = Pattern.compile("^\\d{4}-\\d{2}-\\d{2}");
 
     private final OpenAIClient client;
+    private final Duration timeout;
+    private final int maxRetries;
 
     private final String apiVersion;
 
@@ -170,15 +172,11 @@ public AzureOpenAIChatModelConnection(
                         .credential(AzureApiKeyCredential.create(apiKey))
                         .azureServiceVersion(AzureOpenAIServiceVersion.fromString(apiVersion));
 
-        Integer timeoutSeconds = descriptor.getArgument("timeout");
-        if (timeoutSeconds != null && timeoutSeconds > 0) {
-            clientBuilder.timeout(Duration.ofSeconds(timeoutSeconds));
-        }
+        this.timeout = OpenAIChatCompletionsUtils.parseTimeout(descriptor);
+        clientBuilder.timeout(OpenAIChatCompletionsUtils.toSdkTimeout(this.timeout));
 
-        Integer maxRetries = descriptor.getArgument("max_retries");
-        if (maxRetries != null && maxRetries >= 0) {
-            clientBuilder.maxRetries(maxRetries);
-        }
+        this.maxRetries = OpenAIChatCompletionsUtils.parseMaxRetries(descriptor);
+        clientBuilder.maxRetries(this.maxRetries);
 
         String azureUrlPathMode = descriptor.getArgument("azure_url_path_mode");
         if (azureUrlPathMode != null && !azureUrlPathMode.isBlank()) {
@@ -196,6 +194,16 @@ public AzureOpenAIChatModelConnection(
         this.client = clientBuilder.build();
     }
 
+    // visible for testing
+    Duration getTimeout() {
+        return timeout;
+    }
+
+    // visible for testing
+    int getMaxRetries() {
+        return maxRetries;
+    }
+
     /**
      * Whether Azure documents json_schema strict support for {@code effectiveModel}.
      *

integrations/chat-models/openai/src/main/java/org/apache/flink/agents/integrations/chatmodels/openai/OpenAIChatCompletionsUtils.java

@@ -21,6 +21,7 @@
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.openai.core.JsonValue;
+import com.openai.core.Timeout;
 import com.openai.models.chat.completions.ChatCompletionAssistantMessageParam;
 import com.openai.models.chat.completions.ChatCompletionMessage;
 import com.openai.models.chat.completions.ChatCompletionMessageFunctionToolCall;
@@ -31,7 +32,12 @@
 import com.openai.models.chat.completions.ChatCompletionUserMessageParam;
 import org.apache.flink.agents.api.chat.messages.ChatMessage;
 import org.apache.flink.agents.api.chat.messages.MessageRole;
+import org.apache.flink.agents.api.resource.ResourceDescriptor;
 
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.math.RoundingMode;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.LinkedHashMap;
 import java.util.List;
@@ -41,17 +47,115 @@
 
 /**
  * Static helpers for converting between Flink Agents {@link ChatMessage} and OpenAI Chat
- * Completions API message types. Restricted to message conversion (no tool-definition conversion —
- * that stays per-connection).
+ * Completions API message types, plus shared parsing/validation of common connection arguments
+ * ({@code timeout}, {@code max_retries}). No tool-definition conversion — that stays
+ * per-connection.
  *
  * <p>Used by both {@code OpenAICompletionsConnection} (OpenAI / OpenAI-compatible providers) and
  * {@code AzureOpenAIChatModelConnection} (Azure OpenAI). Both rely on the same openai-java SDK
  * message types.
  */
 final class OpenAIChatCompletionsUtils {
 
+    private static final BigDecimal MAX_TIMEOUT_SECONDS =
+            BigDecimal.valueOf(Integer.MAX_VALUE).movePointLeft(3);
+
+    /** Default timeout in seconds for OpenAI API requests (aligned with Python SDK). */
+    static final int DEFAULT_TIMEOUT_SECONDS = 60;
+
+    /** Default max retries for OpenAI API requests (aligned with Python SDK). */
+    static final int DEFAULT_MAX_RETRIES = 3;
+
     private OpenAIChatCompletionsUtils() {}
 
+    /**
+     * Resolve and validate the {@code timeout} argument (in seconds). The raw value is validated
+     * before any numeric conversion so that e.g. {@code -0.5} cannot truncate to {@code 0} and
+     * bypass the non-negative check. Fractional values are rounded up to the SDK's millisecond
+     * precision so that a positive value can never become an unlimited timeout.
+     */
+    static Duration parseTimeout(ResourceDescriptor descriptor) {
+        Number raw = descriptor.getArgument("timeout");
+        if (raw == null) {
+            return Duration.ofSeconds(DEFAULT_TIMEOUT_SECONDS);
+        }
+        BigDecimal seconds = toBigDecimal(raw, "timeout");
+        if (seconds.signum() < 0) {
+            throw new IllegalArgumentException("timeout must be >= 0, got: " + raw);
+        }
+        if (seconds.compareTo(MAX_TIMEOUT_SECONDS) > 0) {
+            throw new IllegalArgumentException(
+                    "timeout exceeds the SDK maximum of "
+                            + MAX_TIMEOUT_SECONDS.toPlainString()
+                            + " seconds, got: "
+                            + raw);
+        }
+        try {
+            // The SDK's OkHttp transport accepts millisecond precision. Round positive values up
+            // so a valid nonzero timeout cannot become Duration.ZERO, which disables timeouts.
+            BigInteger milliseconds =
+                    seconds.multiply(BigDecimal.valueOf(1_000L))
+                            .setScale(0, RoundingMode.CEILING)
+                            .toBigIntegerExact();
+            return Duration.ofMillis(milliseconds.longValueExact());
+        } catch (ArithmeticException e) {
+            throw new IllegalArgumentException(
+                    "timeout is outside the supported range, got: " + raw, e);
+        }
+    }
+
+    /**
+     * Configure every SDK timeout component from the connection timeout. A zero duration means no
+     * timeout in openai-java, so all components must be set explicitly; setting only the request
+     * timeout leaves the SDK's default connection timeout in effect.
+     */
+    static Timeout toSdkTimeout(Duration timeout) {
+        return Timeout.builder()
+                .connect(timeout)
+                .read(timeout)
+                .write(timeout)
+                .request(timeout)
+                .build();
+    }
+
+    /**
+     * Resolve and validate the {@code max_retries} argument. Requires an exact non-negative integer
+     * within int range, matching Python-side validation (pydantic rejects fractional values for int
+     * fields).
+     */
+    static int parseMaxRetries(ResourceDescriptor descriptor) {
+        Number raw = descriptor.getArgument("max_retries");
+        if (raw == null) {
+            return DEFAULT_MAX_RETRIES;
+        }
+        BigDecimal value = toBigDecimal(raw, "max_retries");
+        try {
+            BigInteger retries = value.toBigIntegerExact();
+            if (retries.signum() < 0
+                    || retries.compareTo(BigInteger.valueOf(Integer.MAX_VALUE)) > 0) {
+                throw new IllegalArgumentException(
+                        "max_retries must be a non-negative integer, got: " + raw);
+            }
+            return retries.intValueExact();
+        } catch (ArithmeticException e) {
+            throw new IllegalArgumentException(
+                    "max_retries must be a non-negative integer, got: " + raw, e);
+        }
+    }
+
+    private static BigDecimal toBigDecimal(Number raw, String argumentName) {
+        if ((raw instanceof Double || raw instanceof Float)
+                && !Double.isFinite(raw.doubleValue())) {
+            throw new IllegalArgumentException(argumentName + " must be finite, got: " + raw);
+        }
+        try {
+            return new BigDecimal(raw.toString());
+        } catch (NumberFormatException e) {
+            throw new IllegalArgumentException(
+                    argumentName + " must be a finite number, got: " + raw, e);
+        }
+    }
+
     private static final ObjectMapper mapper = new ObjectMapper();
     private static final TypeReference<Map<String, Object>> MAP_TYPE = new TypeReference<>() {};
 

integrations/chat-models/openai/src/main/java/org/apache/flink/agents/integrations/chatmodels/openai/OpenAICompletionsConnection.java

@@ -84,6 +84,8 @@ public class OpenAICompletionsConnection extends BaseChatModelConnection {
     private static final ObjectMapper mapper = new ObjectMapper();
     private final OpenAIClient client;
     private final String defaultModel;
+    private final Duration timeout;
+    private final int maxRetries;
 
     public OpenAICompletionsConnection(
             ResourceDescriptor descriptor, ResourceContext resourceContext) {
@@ -101,15 +103,11 @@ public OpenAICompletionsConnection(
             builder.baseUrl(apiBaseUrl);
         }
 
-        Integer timeoutSeconds = descriptor.getArgument("timeout");
-        if (timeoutSeconds != null && timeoutSeconds > 0) {
-            builder.timeout(Duration.ofSeconds(timeoutSeconds));
-        }
+        this.timeout = OpenAIChatCompletionsUtils.parseTimeout(descriptor);
+        builder.timeout(OpenAIChatCompletionsUtils.toSdkTimeout(this.timeout));
 
-        Integer maxRetries = descriptor.getArgument("max_retries");
-        if (maxRetries != null && maxRetries >= 0) {
-            builder.maxRetries(maxRetries);
-        }
+        this.maxRetries = OpenAIChatCompletionsUtils.parseMaxRetries(descriptor);
+        builder.maxRetries(this.maxRetries);
 
         Map<String, String> defaultHeaders = descriptor.getArgument("default_headers");
         if (defaultHeaders != null && !defaultHeaders.isEmpty()) {
@@ -122,6 +120,16 @@ public OpenAICompletionsConnection(
         this.client = builder.build();
     }
 
+    // visible for testing
+    Duration getTimeout() {
+        return timeout;
+    }
+
+    // visible for testing
+    int getMaxRetries() {
+        return maxRetries;
+    }
+
     // Models for which OpenAI documents json_schema strict Structured Outputs support.
     // Source of truth: https://platform.openai.com/docs/guides/structured-outputs
     //

integrations/chat-models/openai/src/main/java/org/apache/flink/agents/integrations/chatmodels/openai/OpenAIResponsesModelConnection.java

@@ -58,8 +58,10 @@
  * <ul>
  *   <li><b>api_key</b> (required): OpenAI API key
  *   <li><b>api_base_url</b> (optional): Base URL for OpenAI API (useful for proxies)
- *   <li><b>timeout</b> (optional): Timeout in seconds for API requests
- *   <li><b>max_retries</b> (optional): Maximum number of retry attempts (default: 2)
+ *   <li><b>timeout</b> (optional): Timeout in seconds for API requests; must be non-negative
+ *       (default: 60)
+ *   <li><b>max_retries</b> (optional): Maximum number of retry attempts; must be non-negative
+ *       (default: 3)
  *   <li><b>default_headers</b> (optional): Map of default headers to include in all requests
  *   <li><b>model</b> (optional): Default model to use if not specified in setup
  * </ul>
@@ -86,6 +88,8 @@ public class OpenAIResponsesModelConnection extends BaseChatModelConnection {
 
     private final OpenAIClient client;
     private final String defaultModel;
+    private final Duration timeout;
+    private final int maxRetries;
 
     public OpenAIResponsesModelConnection(
             ResourceDescriptor descriptor, ResourceContext resourceContext) {
@@ -103,15 +107,11 @@ public OpenAIResponsesModelConnection(
             builder.baseUrl(apiBaseUrl);
         }
 
-        Integer timeoutSeconds = descriptor.getArgument("timeout");
-        if (timeoutSeconds != null && timeoutSeconds > 0) {
-            builder.timeout(Duration.ofSeconds(timeoutSeconds));
-        }
+        this.timeout = OpenAIChatCompletionsUtils.parseTimeout(descriptor);
+        builder.timeout(OpenAIChatCompletionsUtils.toSdkTimeout(this.timeout));
 
-        Integer maxRetries = descriptor.getArgument("max_retries");
-        if (maxRetries != null && maxRetries >= 0) {
-            builder.maxRetries(maxRetries);
-        }
+        this.maxRetries = OpenAIChatCompletionsUtils.parseMaxRetries(descriptor);
+        builder.maxRetries(this.maxRetries);
 
         Map<String, String> defaultHeaders = descriptor.getArgument("default_headers");
         if (defaultHeaders != null && !defaultHeaders.isEmpty()) {
@@ -453,6 +453,14 @@ private Map<String, Object> toMap(Object value) {
         return mapper.convertValue(value, MAP_TYPE);
     }
 
+    Duration getTimeout() {
+        return timeout;
+    }
+
+    int getMaxRetries() {
+        return maxRetries;
+    }
+
     @Override
     public void close() throws Exception {
         this.client.close();

integrations/chat-models/openai/src/test/java/org/apache/flink/agents/integrations/chatmodels/openai/AzureOpenAIChatModelConnectionTest.java

@@ -43,6 +43,7 @@
 import org.junit.jupiter.params.provider.NullAndEmptySource;
 import org.junit.jupiter.params.provider.ValueSource;
 
+import java.time.Duration;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -95,6 +96,25 @@ private static AzureOpenAIChatModelConnection connection() {
         return connection(CAPABLE_API_VERSION);
     }
 
+    @Test
+    void testConnectionArgumentDefaultsAndZeroTimeout() {
+        AzureOpenAIChatModelConnection connection =
+                new AzureOpenAIChatModelConnection(
+                        connectionDescriptor()
+                                .addInitialArgument("api_key", "test-key")
+                                .addInitialArgument("api_version", "2024-02-01")
+                                .addInitialArgument(
+                                        "azure_endpoint", "https://example.openai.azure.com")
+                                .addInitialArgument("timeout", 0)
+                                .addInitialArgument("max_retries", 0)
+                                .build(),
+                        NOOP);
+        assertThat(connection).isInstanceOf(BaseChatModelConnection.class);
+        assertThat(connection.getTimeout()).isEqualTo(Duration.ZERO);
+        assertThat(connection.getMaxRetries()).isZero();
+        OpenAIClientTestUtils.assertNoTimeoutConfigured(connection);
+    }
+
     /**
      * Model params addressing {@link #DEPLOYMENT}. A null {@code modelOfAzureDeployment} omits the
      * key entirely, which is how the setup emits an unset backing model.