diff --git a/code_review.md b/code_review.md index 757b3d3ad..ad211cba4 100644 --- a/code_review.md +++ b/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) | diff --git a/docs/content/docs/development/chat_models.md b/docs/content/docs/development/chat_models.md index 2fb2d19b6..63b1862e0 100644 --- a/docs/content/docs/development/chat_models.md +++ b/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 | 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 | None | Default headers for API requests | | `model` | String | None | Default model to use if not specified in setup | 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 31b2c5ab8..9f9267f11 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 @@ -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}. * diff --git a/integrations/chat-models/openai/src/main/java/org/apache/flink/agents/integrations/chatmodels/openai/OpenAIChatCompletionsUtils.java b/integrations/chat-models/openai/src/main/java/org/apache/flink/agents/integrations/chatmodels/openai/OpenAIChatCompletionsUtils.java index c9d8c8d9b..35eddaf1c 100644 --- a/integrations/chat-models/openai/src/main/java/org/apache/flink/agents/integrations/chatmodels/openai/OpenAIChatCompletionsUtils.java +++ b/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,8 +47,9 @@ /** * 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. * *

Used by both {@code OpenAICompletionsConnection} (OpenAI / OpenAI-compatible providers) and * {@code AzureOpenAIChatModelConnection} (Azure OpenAI). Both rely on the same openai-java SDK @@ -50,8 +57,105 @@ */ 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_TYPE = new TypeReference<>() {}; 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 92683b9ce..f39c6ed3e 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 @@ -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 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 // 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 0fcd484b4..85d260032 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 @@ -58,8 +58,10 @@ *

@@ -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 defaultHeaders = descriptor.getArgument("default_headers"); if (defaultHeaders != null && !defaultHeaders.isEmpty()) { @@ -453,6 +453,14 @@ private Map 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(); 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 97b9a2cb6..2bda0db2b 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 @@ -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. diff --git a/integrations/chat-models/openai/src/test/java/org/apache/flink/agents/integrations/chatmodels/openai/OpenAIChatCompletionsUtilsTest.java b/integrations/chat-models/openai/src/test/java/org/apache/flink/agents/integrations/chatmodels/openai/OpenAIChatCompletionsUtilsTest.java index 9f260366d..da93951ea 100644 --- a/integrations/chat-models/openai/src/test/java/org/apache/flink/agents/integrations/chatmodels/openai/OpenAIChatCompletionsUtilsTest.java +++ b/integrations/chat-models/openai/src/test/java/org/apache/flink/agents/integrations/chatmodels/openai/OpenAIChatCompletionsUtilsTest.java @@ -20,12 +20,16 @@ import com.openai.models.chat.completions.ChatCompletionMessage; import org.apache.flink.agents.api.chat.messages.ChatMessage; +import org.apache.flink.agents.api.resource.ResourceDescriptor; import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Test; +import java.math.BigDecimal; +import java.time.Duration; import java.util.Optional; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; /** * Unit tests for how {@link OpenAIChatCompletionsUtils} carries a provider refusal from a Chat @@ -60,4 +64,29 @@ void testNoRefusalKeyWhenAbsent() { assertThat(result.getExtraArgs()).doesNotContainKey("refusal"); } + + @Test + void testMaxRetriesRejectsFractionalBigDecimal() { + assertThatThrownBy( + () -> + OpenAIChatCompletionsUtils.parseMaxRetries( + descriptor( + "max_retries", + new BigDecimal("2.0000000000000000000000001")))) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + void testPositiveTimeoutBelowMillisecondRoundsUpToSdkPrecision() { + assertThat( + OpenAIChatCompletionsUtils.parseTimeout( + descriptor("timeout", new BigDecimal("0.0000000001")))) + .isEqualTo(Duration.ofMillis(1)); + } + + private static ResourceDescriptor descriptor(String argumentName, Number value) { + return ResourceDescriptor.Builder.newBuilder(OpenAICompletionsConnection.class.getName()) + .addInitialArgument(argumentName, value) + .build(); + } } diff --git a/integrations/chat-models/openai/src/test/java/org/apache/flink/agents/integrations/chatmodels/openai/OpenAIClientTestUtils.java b/integrations/chat-models/openai/src/test/java/org/apache/flink/agents/integrations/chatmodels/openai/OpenAIClientTestUtils.java new file mode 100644 index 000000000..6347d63f5 --- /dev/null +++ b/integrations/chat-models/openai/src/test/java/org/apache/flink/agents/integrations/chatmodels/openai/OpenAIClientTestUtils.java @@ -0,0 +1,53 @@ +/* + * 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.openai.client.OpenAIClient; +import com.openai.core.ClientOptions; + +import java.lang.reflect.Field; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Test helpers for inspecting the timeout configured in openai-java. */ +final class OpenAIClientTestUtils { + + private OpenAIClientTestUtils() {} + + static void assertNoTimeoutConfigured(Object connection) { + OpenAIClient client = readField(connection, "client", OpenAIClient.class); + ClientOptions options = readField(client, "clientOptions", ClientOptions.class); + + // A zero duration maps to an unlimited timeout in each OkHttp timeout component. + assertThat(options.timeout().connect()).isZero(); + assertThat(options.timeout().read()).isZero(); + assertThat(options.timeout().write()).isZero(); + assertThat(options.timeout().request()).isZero(); + } + + private static T readField(Object target, String name, Class type) { + try { + Field field = target.getClass().getDeclaredField(name); + field.setAccessible(true); + return type.cast(field.get(target)); + } catch (ReflectiveOperationException e) { + throw new AssertionError("Unable to inspect openai-java client configuration", e); + } + } +} 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 be7a966e9..1c5a4d15b 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 @@ -22,6 +22,7 @@ import com.openai.models.chat.completions.ChatCompletionCreateParams; 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.apache.flink.agents.api.tools.Tool; @@ -32,11 +33,13 @@ import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Test; +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 OpenAICompletionsConnection}'s native structured-output behavior. These @@ -72,6 +75,30 @@ private static List userMessage() { return List.of(new ChatMessage(MessageRole.USER, "hi")); } + @Test + void testConnectionArgumentValidation() { + ResourceDescriptor missingKey = + ResourceDescriptor.Builder.newBuilder(OpenAICompletionsConnection.class.getName()) + .build(); + assertThatThrownBy(() -> new OpenAICompletionsConnection(missingKey, NOOP)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("api_key"); + + OpenAICompletionsConnection connection = + new OpenAICompletionsConnection( + ResourceDescriptor.Builder.newBuilder( + OpenAICompletionsConnection.class.getName()) + .addInitialArgument("api_key", "test-key") + .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); + } + @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 new file mode 100644 index 000000000..d5f0641c8 --- /dev/null +++ b/integrations/chat-models/openai/src/test/java/org/apache/flink/agents/integrations/chatmodels/openai/OpenAIResponsesModelConnectionTest.java @@ -0,0 +1,204 @@ +/* + * 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 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.time.Duration; + +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. + */ +class OpenAIResponsesModelConnectionTest { + + private static final ResourceContext NOOP = ResourceContext.fromGetResource((a, b) -> null); + + private static ResourceDescriptor.Builder connectionDescriptor() { + return ResourceDescriptor.Builder.newBuilder( + OpenAIResponsesModelConnection.class.getName()); + } + + @Test + @DisplayName("Constructor throws when api_key is missing") + void testConstructorMissingApiKey() { + ResourceDescriptor desc = connectionDescriptor().build(); + assertThatThrownBy(() -> new OpenAIResponsesModelConnection(desc, NOOP)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("api_key"); + } + + @Test + @DisplayName("Constructor succeeds with api_key only (no network call)") + void testConstructorMinimal() { + ResourceDescriptor desc = + connectionDescriptor().addInitialArgument("api_key", "test-key").build(); + OpenAIResponsesModelConnection conn = new OpenAIResponsesModelConnection(desc, NOOP); + assertThat(conn).isInstanceOf(BaseChatModelConnection.class); + } + + @Test + @DisplayName("Defaults resolve to timeout=60 and max_retries=3 when not specified") + void testDefaultTimeoutAndMaxRetries() { + ResourceDescriptor desc = + connectionDescriptor().addInitialArgument("api_key", "test-key").build(); + OpenAIResponsesModelConnection conn = new OpenAIResponsesModelConnection(desc, NOOP); + + assertThat(conn.getTimeout()) + .isEqualTo(Duration.ofSeconds(OpenAIChatCompletionsUtils.DEFAULT_TIMEOUT_SECONDS)); + assertThat(conn.getMaxRetries()).isEqualTo(OpenAIChatCompletionsUtils.DEFAULT_MAX_RETRIES); + } + + @Test + @DisplayName("Explicit timeout and max_retries override the defaults") + void testExplicitOverrides() { + ResourceDescriptor desc = + connectionDescriptor() + .addInitialArgument("api_key", "test-key") + .addInitialArgument("timeout", 120) + .addInitialArgument("max_retries", 5) + .build(); + OpenAIResponsesModelConnection conn = new OpenAIResponsesModelConnection(desc, NOOP); + + assertThat(conn.getTimeout()).isEqualTo(Duration.ofSeconds(120)); + assertThat(conn.getMaxRetries()).isEqualTo(5); + } + + @Test + @DisplayName("Negative timeout throws IllegalArgumentException") + void testNegativeTimeoutThrows() { + ResourceDescriptor desc = + connectionDescriptor() + .addInitialArgument("api_key", "test-key") + .addInitialArgument("timeout", -5) + .build(); + assertThatThrownBy(() -> new OpenAIResponsesModelConnection(desc, NOOP)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("timeout"); + } + + @Test + @DisplayName("Negative max_retries throws IllegalArgumentException") + void testNegativeMaxRetriesThrows() { + ResourceDescriptor desc = + connectionDescriptor() + .addInitialArgument("api_key", "test-key") + .addInitialArgument("max_retries", -1) + .build(); + assertThatThrownBy(() -> new OpenAIResponsesModelConnection(desc, NOOP)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("max_retries"); + } + + @Test + @DisplayName("Negative fractional timeout throws instead of truncating to zero") + void testNegativeFractionalTimeoutThrows() { + ResourceDescriptor desc = + connectionDescriptor() + .addInitialArgument("api_key", "test-key") + .addInitialArgument("timeout", -0.5) + .build(); + assertThatThrownBy(() -> new OpenAIResponsesModelConnection(desc, NOOP)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("timeout"); + } + + @Test + @DisplayName("Fractional max_retries throws instead of truncating") + void testFractionalMaxRetriesThrows() { + ResourceDescriptor desc = + connectionDescriptor() + .addInitialArgument("api_key", "test-key") + .addInitialArgument("max_retries", 2.5) + .build(); + assertThatThrownBy(() -> new OpenAIResponsesModelConnection(desc, NOOP)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("max_retries"); + } + + @Test + @DisplayName("Negative fractional max_retries throws instead of truncating to zero") + void testNegativeFractionalMaxRetriesThrows() { + ResourceDescriptor desc = + connectionDescriptor() + .addInitialArgument("api_key", "test-key") + .addInitialArgument("max_retries", -0.5) + .build(); + assertThatThrownBy(() -> new OpenAIResponsesModelConnection(desc, NOOP)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("max_retries"); + } + + @Test + @DisplayName("max_retries beyond int range throws instead of overflowing") + void testOverflowMaxRetriesThrows() { + ResourceDescriptor desc = + connectionDescriptor() + .addInitialArgument("api_key", "test-key") + .addInitialArgument("max_retries", 4294967296L) + .build(); + assertThatThrownBy(() -> new OpenAIResponsesModelConnection(desc, NOOP)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("max_retries"); + } + + @Test + @DisplayName("Zero timeout disables the effective SDK timeout") + void testZeroTimeoutDisablesSdkTimeout() { + ResourceDescriptor desc = + connectionDescriptor() + .addInitialArgument("api_key", "test-key") + .addInitialArgument("timeout", 0) + .build(); + OpenAIResponsesModelConnection conn = new OpenAIResponsesModelConnection(desc, NOOP); + assertThat(conn.getTimeout()).isEqualTo(Duration.ZERO); + OpenAIClientTestUtils.assertNoTimeoutConfigured(conn); + } + + @Test + @DisplayName("Sub-millisecond timeout rounds up to the SDK precision") + void testSubMillisecondTimeoutRoundsUpToSdkPrecision() { + ResourceDescriptor desc = + connectionDescriptor() + .addInitialArgument("api_key", "test-key") + .addInitialArgument("timeout", 0.0001) + .build(); + OpenAIResponsesModelConnection conn = new OpenAIResponsesModelConnection(desc, NOOP); + assertThat(conn.getTimeout()).isEqualTo(Duration.ofMillis(1)); + } + + @Test + @DisplayName("Zero max_retries is accepted as valid") + void testZeroMaxRetriesAccepted() { + ResourceDescriptor desc = + connectionDescriptor() + .addInitialArgument("api_key", "test-key") + .addInitialArgument("max_retries", 0) + .build(); + OpenAIResponsesModelConnection conn = new OpenAIResponsesModelConnection(desc, NOOP); + assertThat(conn.getMaxRetries()).isEqualTo(0); + } +} diff --git a/python/flink_agents/integrations/chat_models/azure/azure_openai_chat_model.py b/python/flink_agents/integrations/chat_models/azure/azure_openai_chat_model.py index 422d9d09e..50884bf68 100644 --- a/python/flink_agents/integrations/chat_models/azure/azure_openai_chat_model.py +++ b/python/flink_agents/integrations/chat_models/azure/azure_openai_chat_model.py @@ -43,6 +43,8 @@ ) logger = logging.getLogger(__name__) +MAX_OPENAI_TIMEOUT_SECONDS = 2_147_483.647 +MAX_OPENAI_RETRIES = 2_147_483_647 _RESERVED_KWARG_KEYS = frozenset( {"model", "model_of_azure_deployment", "temperature", "max_tokens", "logprobs"} @@ -144,13 +146,16 @@ class AzureOpenAIChatModelConnection(BaseChatModelConnection): ) timeout: float = Field( default=60.0, - description="The number of seconds to wait for an API call before it times out.", + description="The number of seconds to wait for an API call before it times out. Set to 0 to disable timeouts.", ge=0, + le=MAX_OPENAI_TIMEOUT_SECONDS, + allow_inf_nan=False, ) max_retries: int = Field( default=3, description="The number of times to retry the API call upon failure.", ge=0, + le=MAX_OPENAI_RETRIES, ) def __init__( @@ -183,7 +188,8 @@ def client(self) -> AzureOpenAI: azure_endpoint=self.azure_endpoint, api_key=self.api_key, api_version=self.api_version, - timeout=self.timeout, + # Match Java's Duration.ZERO: None avoids an immediate timeout in httpx. + timeout=None if self.timeout == 0 else self.timeout, max_retries=self.max_retries, ) return self._client diff --git a/python/flink_agents/integrations/chat_models/azure/tests/test_azure_openai_chat_model.py b/python/flink_agents/integrations/chat_models/azure/tests/test_azure_openai_chat_model.py index 983bbb47f..0f1c99fb2 100644 --- a/python/flink_agents/integrations/chat_models/azure/tests/test_azure_openai_chat_model.py +++ b/python/flink_agents/integrations/chat_models/azure/tests/test_azure_openai_chat_model.py @@ -15,15 +15,19 @@ # See the License for the specific language governing permissions and # limitations under the License. ################################################################################# +import math import os from unittest.mock import MagicMock import pytest +from pydantic import ValidationError from flink_agents.api.chat_message import ChatMessage, MessageRole from flink_agents.api.resource import Resource, ResourceType from flink_agents.api.resource_context import ResourceContext from flink_agents.integrations.chat_models.azure.azure_openai_chat_model import ( + MAX_OPENAI_RETRIES, + MAX_OPENAI_TIMEOUT_SECONDS, AzureOpenAIChatModelConnection, AzureOpenAIChatModelSetup, ) @@ -131,6 +135,55 @@ def test_model_field_roundtrip() -> None: assert restored.model == "test-deployment" +def test_zero_timeout_disables_client_timeout() -> None: + """Keep zero-timeout semantics aligned with the Java OpenAI SDK.""" + conn = AzureOpenAIChatModelConnection( + api_key="fake-key", + azure_endpoint="https://example.openai.azure.com", + api_version="2024-02-01", + timeout=0, + ) + + assert conn.client.timeout is None + # openai>=3 wraps timeouts in its own Timeout class, so compare + # components instead of instances. + transport_timeout = conn.client._client.timeout + assert transport_timeout.connect is None + assert transport_timeout.read is None + assert transport_timeout.write is None + assert transport_timeout.pool is None + + +@pytest.mark.parametrize("timeout", [math.nan, math.inf]) +def test_connection_rejects_non_finite_timeout(timeout: float) -> None: + with pytest.raises(ValidationError, match="finite"): + AzureOpenAIChatModelConnection( + api_key="fake-key", + azure_endpoint="https://example.openai.azure.com", + api_version="2024-02-01", + timeout=timeout, + ) + + +@pytest.mark.parametrize( + ("argument", "value"), + [ + ("timeout", MAX_OPENAI_TIMEOUT_SECONDS + 0.001), + ("max_retries", MAX_OPENAI_RETRIES + 1), + ], +) +def test_connection_rejects_values_beyond_java_sdk_limits( + argument: str, value: float | int +) -> None: + with pytest.raises(ValidationError, match="less than or equal"): + AzureOpenAIChatModelConnection( + api_key="fake-key", + azure_endpoint="https://example.openai.azure.com", + api_version="2024-02-01", + **{argument: value}, + ) + + def test_model_kwargs_nests_additional_kwargs() -> None: """`additional_kwargs` is nested under its own key, not flattened. diff --git a/python/flink_agents/integrations/chat_models/openai/openai_chat_model.py b/python/flink_agents/integrations/chat_models/openai/openai_chat_model.py index 381098b00..e810613f7 100644 --- a/python/flink_agents/integrations/chat_models/openai/openai_chat_model.py +++ b/python/flink_agents/integrations/chat_models/openai/openai_chat_model.py @@ -43,6 +43,8 @@ ) DEFAULT_OPENAI_MODEL = "gpt-4o-mini" +MAX_OPENAI_TIMEOUT_SECONDS = 2_147_483.647 +MAX_OPENAI_RETRIES = 2_147_483_647 # Models with documented json_schema strict Structured Outputs support. Source of # truth: https://platform.openai.com/docs/guides/structured-outputs @@ -129,11 +131,14 @@ class OpenAIChatModelConnection(BaseChatModelConnection): default=3, description="The maximum number of API retries.", ge=0, + le=MAX_OPENAI_RETRIES, ) timeout: float = Field( default=60.0, - description="The timeout, in seconds, for API requests.", + description="The timeout, in seconds, for API requests. Set to 0 to disable timeouts.", ge=0, + le=MAX_OPENAI_TIMEOUT_SECONDS, + allow_inf_nan=False, ) default_headers: Dict[str, str] | None = Field( default=None, description="The default headers for API requests." @@ -177,6 +182,8 @@ def __init__( self._http_client = http_client self._async_http_client = async_http_client + if self.timeout == 0 and self._http_client is not None: + self._http_client.timeout = httpx.Timeout(None) @property def client(self) -> OpenAI: @@ -195,7 +202,7 @@ def __get_client_kwargs(self) -> Dict[str, Any]: "api_key": self.api_key, "base_url": self.api_base_url, "max_retries": self.max_retries, - "timeout": self.timeout, + "timeout": None if self.timeout == 0 else self.timeout, "default_headers": self.default_headers, "http_client": self._http_client, } diff --git a/python/flink_agents/integrations/chat_models/openai/tests/test_openai_chat_model.py b/python/flink_agents/integrations/chat_models/openai/tests/test_openai_chat_model.py index ff4d2bb39..6eae8273a 100644 --- a/python/flink_agents/integrations/chat_models/openai/tests/test_openai_chat_model.py +++ b/python/flink_agents/integrations/chat_models/openai/tests/test_openai_chat_model.py @@ -15,16 +15,21 @@ # See the License for the specific language governing permissions and # limitations under the License. ################################################################################# +import math import os from unittest.mock import MagicMock +import httpx import pytest +from pydantic import ValidationError from flink_agents.api.chat_message import ChatMessage, MessageRole from flink_agents.api.resource import Resource, ResourceType from flink_agents.api.resource_context import ResourceContext from flink_agents.integrations.chat_models.openai.openai_chat_model import ( DEFAULT_OPENAI_MODEL, + MAX_OPENAI_RETRIES, + MAX_OPENAI_TIMEOUT_SECONDS, OpenAIChatModelConnection, OpenAIChatModelSetup, ) @@ -121,3 +126,74 @@ def test_default_model_when_omitted() -> None: """Verify per-integration default applies when `model` is omitted from __init__.""" setup = OpenAIChatModelSetup(connection="conn") assert setup.model == DEFAULT_OPENAI_MODEL + + +def test_connection_default_timeout_and_max_retries() -> None: + """Pin canonical connection defaults to prevent silent drift.""" + conn = OpenAIChatModelConnection( + name="test", api_key="fake", api_base_url="http://localhost" + ) + assert conn.timeout == 60.0 + assert conn.max_retries == 3 + + +def test_zero_timeout_disables_client_timeout() -> None: + """Keep zero-timeout semantics aligned with the Java OpenAI SDK.""" + conn = OpenAIChatModelConnection( + name="test", api_key="fake", api_base_url="http://localhost", timeout=0 + ) + + assert conn.client.timeout is None + # openai>=3 wraps timeouts in its own Timeout class, so compare + # components instead of instances. + transport_timeout = conn.client._client.timeout + assert transport_timeout.connect is None + assert transport_timeout.read is None + assert transport_timeout.write is None + assert transport_timeout.pool is None + + +def test_zero_timeout_disables_custom_http_client_timeout() -> None: + http_client = httpx.Client(timeout=10.0) + conn = OpenAIChatModelConnection( + name="test", + api_key="fake", + api_base_url="http://localhost", + timeout=0, + http_client=http_client, + ) + + assert conn.client.timeout is None + assert http_client.timeout == httpx.Timeout(None) + + http_client.close() + + +@pytest.mark.parametrize("timeout", [math.nan, math.inf]) +def test_connection_rejects_non_finite_timeout(timeout: float) -> None: + with pytest.raises(ValidationError, match="finite"): + OpenAIChatModelConnection( + name="test", + api_key="fake", + api_base_url="http://localhost", + timeout=timeout, + ) + + +@pytest.mark.parametrize( + ("argument", "value"), + [ + ("timeout", MAX_OPENAI_TIMEOUT_SECONDS + 0.001), + ("max_retries", MAX_OPENAI_RETRIES + 1), + ], +) +def test_connection_rejects_values_beyond_java_sdk_limits( + argument: str, value: float | int +) -> None: + with pytest.raises(ValidationError, match="less than or equal"): + OpenAIChatModelConnection( + name="test", + api_key="fake", + api_base_url="http://localhost", + **{argument: value}, + ) diff --git a/review-guides/api-contract.md b/review-guides/api-contract.md new file mode 100644 index 000000000..c8d953a6a --- /dev/null +++ b/review-guides/api-contract.md @@ -0,0 +1,80 @@ +# Review Guide: api/ Contract + +Load this guide when a PR changes a public API surface: a signature or type in +`api/`, a new resource implementation, a config option, a YAML-visible name, or +anything a user's agent code calls. It narrows the full passes in +`code_review.md` to the ones that matter most for this area; the general passes +still apply. + +## Focused checklist + +- When a PR adds a public resource implementation, check that its short YAML + alias landed in both alias tables and the doc table. Nothing fails when all + three are skipped: each loader passes an unrecognized name through unchanged, + so the class stays reachable by fully-qualified name and no test notices the + omission. +- Regenerate the cross-language snapshots on both sides in the same change when + a field on a built-in event or on the agent plan is added, renamed, or + retyped. Each language pins its own serialization against its own committed + file, so refreshing one side leaves the other side's stability test failing. A + field added in only one language is caught by nothing, because the payload is + a free-form attribute map on the read side. +- Check that a new public config option landed on both languages' sides. A + Java-only option passes every fast CI job: the bidirectional parity check runs + only in the slow cross-language lane, and the in-tree guard is a hardcoded + count on the Python side. +- Treat the public base classes users extend as source-compatibility boundaries. + A new abstract method breaks every implementation, including ones outside this + repo, while a defaulted overload plus a capability probe does not. Compiling + is not the same as honoring the new argument: a default that forwards to the + older signature drops it in silence, so check that the default rejects what it + cannot honor, and that an override gating it behind a capability probe leaves + a fallback in force or fails, rather than silently doing neither. The Python + guard that catches a mis-declared override only sees connections whose module + is imported by hand at the top of the test. +- Check that a removal is complete rather than asking whether to deprecate. + There is no deprecation mechanism in this repo, so an API is either kept or + deleted outright. Under the beta policy, prefer deleting unless a concrete + compatibility obligation requires keeping it. +- Name the docs the change invalidates. Config keys, YAML aliases, and whole + code samples are restated by hand across the doc site, and nothing in pull + request CI builds or checks them, so a doc that contradicts the code ships + green. Java code under `examples/` is in the Maven reactor and breaks loudly, + but nothing imports or runs the Python examples. + +## Validation + +Run both lanes. A change verified in one language only is untested in the other. + +- Java, from the repo root: `mvn --batch-mode test -pl api`. +- Python, from `python/`: `uv sync --extra test`, then `uv pip install + "apache-flink==$(mvn -q -N --batch-mode -f ../pom.xml help:evaluate + -Dexpression=flink.version -DforceStdout)"`, then `uv run --no-sync pytest + flink_agents/api flink_agents/plan`. PyFlink is imported at module scope by + the event types but is declared in neither the base dependencies nor the + `test` extra, so collection fails without it. Install it after the sync, not + before, because `uv sync` removes it. Reading the version out of the root + `pom.xml` rather than typing it runs this lane on the same Flink the Java + bullet resolves, and survives a version bump. +- Resource-name constants, from `python/`: `uv run --no-sync python + ../e2e-test/test-scripts/check_resource_consistency.py`. This one is a script + rather than a test, so neither Maven nor pytest reaches it. + +Two more when the change reaches further: + +- Agent-plan wire format: `mvn --batch-mode test -pl plan -am + -Dtest=AgentPlanCrossLanguageTest -Dsurefire.failIfNoSpecifiedTests=false`. + The `-am` is what puts your edited `api` classes on the classpath in place of + the last installed jar. +- A new or changed config option: `mvn --batch-mode package -pl api -DskipTests`, + then from `python/`, `uv run --no-sync python + flink_agents/plan/tests/compatibility/check_java_python_config_options_parity.py`. + It loads the Java class out of `api/target/`, so an absent or stale jar is + exactly what it reads. + +## Examples from past reviews + +| Case | Pass it exercises | Review | +|---|---|---| +| A new public reconciler contract gave a recovering call two ways out, both of them throws: a terminal business exception, or a fallback exception that re-runs the call. The review asked what a partially succeeded call is supposed to do, and proposed reducing the contract to returning a result or throwing. | Whether a new public contract's semantics fit its users, not only the implementation behind it. | [#600](https://github.com/apache/flink-agents/pull/600#discussion_r3027614878) | +| Renaming a persisted plan key left a deserializer fallback so plans written under the old name still loaded. The review argued the fallback was not worth keeping, since API compatibility is not guaranteed in the 0.x series and the formal stability commitment starts at 1.0. | Whether a compatibility path is justified under the beta policy. | [#756](https://github.com/apache/flink-agents/pull/756#discussion_r3392902728) | diff --git a/review-guides/python-java-bridge.md b/review-guides/python-java-bridge.md index 685a19888..b55393fda 100644 --- a/review-guides/python-java-bridge.md +++ b/review-guides/python-java-bridge.md @@ -36,11 +36,12 @@ Run both language lanes. A bridge change verified on one side only is untested. - Java: `mvn --batch-mode test -pl runtime -am`. The `-am` matters here because the Java halves of the cross-language snapshot tests live in `api` and `plan`, upstream of the module that owns the bridge implementations. -- Python: from `python/`, run `uv sync --extra test`, install the - `apache-flink` release for the Flink version under test (`tools/ut.sh` names - the supported versions), then `uv run --no-sync pytest flink_agents/runtime - flink_agents/api flink_agents/plan`. PyFlink is not a declared test - dependency and the event types import it, so collection fails without it. +- Python: from `python/`, run `uv sync --extra test`, then `uv pip install + "apache-flink==$(mvn -q -N --batch-mode -f ../pom.xml help:evaluate + -Dexpression=flink.version -DforceStdout)"`, then `uv run --no-sync pytest + flink_agents/runtime flink_agents/api flink_agents/plan`. PyFlink is not a + declared test dependency and the event types import it, so collection fails + without it. Together these run the committed cross-language snapshot tests from both sides. Dispatch through a real interpreter is only covered by the cross-language