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 73bc3d5e [python] Remove PyFlink gateway fallback from
AgentConfigOptions (#867)
73bc3d5e is described below
commit 73bc3d5e50b7ca84773b52121305b9b5d77edfc1
Author: Dagdraumur <[email protected]>
AuthorDate: Wed Jul 8 10:22:22 2026 +0800
[python] Remove PyFlink gateway fallback from AgentConfigOptions (#867)
---
.../test-scripts/test_java_config_in_python.sh | 8 +-
python/flink_agents/api/core_options.py | 186 ++++++++++++++-------
python/flink_agents/api/tests/test_core_options.py | 114 +++++++++++++
.../check_java_python_config_options_parity.py | 183 ++++++++++++++++++++
.../create_python_option_from_java_option.py | 79 ---------
5 files changed, 427 insertions(+), 143 deletions(-)
diff --git a/e2e-test/test-scripts/test_java_config_in_python.sh
b/e2e-test/test-scripts/test_java_config_in_python.sh
index c0ebff92..9706056b 100644
--- a/e2e-test/test-scripts/test_java_config_in_python.sh
+++ b/e2e-test/test-scripts/test_java_config_in_python.sh
@@ -28,13 +28,13 @@ echo $root_dir
python_script_path=$root_dir/python/flink_agents/plan/tests/compatibility
-function test_create_python_option_from_java_option {
- python $python_script_path/create_python_option_from_java_option.py
+function test_java_python_config_options_parity {
+ python $python_script_path/check_java_python_config_options_parity.py
ret=$?
if [ "$ret" != "0" ]
then
- echo "There is failure when create python option from java option, please
check the log for details."
+ echo "Java/Python ConfigOption parity check failed, please check the log
for details."
rm -f $json_path
exit $ret
fi
@@ -42,4 +42,4 @@ function test_create_python_option_from_java_option {
rm -f $json_path
}
-test_create_python_option_from_java_option
+test_java_python_config_options_parity
diff --git a/python/flink_agents/api/core_options.py
b/python/flink_agents/api/core_options.py
index f5247616..d196777c 100644
--- a/python/flink_agents/api/core_options.py
+++ b/python/flink_agents/api/core_options.py
@@ -17,60 +17,10 @@
#################################################################################
import os
from enum import Enum
-from typing import Any
-
-from pyflink.java_gateway import get_gateway
from flink_agents.api.configuration import ConfigOption
-def covert_j_option_to_python_option(j_option: Any) -> ConfigOption:
- """Convert a Java config option to a Python config option."""
- key = j_option.getKey()
- default = j_option.getDefaultValue()
- type_name = j_option.getTypeName()
-
- if type_name == "java.lang.String":
- config_type = str
- elif type_name == "java.lang.Integer":
- config_type = int
- elif type_name == "java.lang.Long":
- config_type = int
- elif type_name == "java.lang.Boolean":
- config_type = bool
- elif type_name == "java.lang.Float":
- config_type = float
- elif type_name == "java.lang.Double":
- config_type = float
- else:
- msg = f"Unsupported type: {type_name}"
- raise TypeError(msg)
-
- return ConfigOption(key, config_type, default)
-
-
-class AgentConfigOptionsMeta(type):
- """Metaclass for FlinkAgentsCoreOptions."""
-
- def __init__(
- cls, name: str, bases: tuple[type, ...], attrs: dict[str, Any]
- ) -> None:
- """Initialize the metaclass for FlinkAgentsCoreOptions."""
- super().__init__(name, bases, attrs)
-
- jvm = get_gateway().jvm
- cls.jvm = jvm
-
- def __getattr__(cls, item: str) -> ConfigOption:
- j_option = getattr(
-
cls.jvm.org.apache.flink.agents.api.configuration.AgentConfigOptions,
- item,
- )
-
- python_option = covert_j_option_to_python_option(j_option)
- return python_option
-
-
class ErrorHandlingStrategy(Enum):
"""Error handling strategy for Agent.
@@ -108,14 +58,23 @@ class LoggerType(Enum):
FILE = "file"
-class AgentConfigOptions(metaclass=AgentConfigOptionsMeta):
- """CoreOptions to manage core configuration parameters for Flink Agents."""
+class EventLogLevel(Enum):
+ """Log level for event logging.
- JOB_IDENTIFIER = ConfigOption(
- key="job-identifier",
- config_type=str,
- default=None,
- )
+ Mirrors the Java ``EventLogLevel`` enum.
+ """
+
+ OFF = "OFF"
+ STANDARD = "STANDARD"
+ VERBOSE = "VERBOSE"
+
+
+class AgentConfigOptions:
+ """CoreOptions to manage core configuration parameters for Flink Agents.
+
+ Options are declared explicitly in Python and must stay aligned with the
+ Java ``AgentConfigOptions`` class.
+ """
EVENT_LOGGER_TYPE = ConfigOption(
key="eventLoggerType",
@@ -123,11 +82,112 @@ class
AgentConfigOptions(metaclass=AgentConfigOptionsMeta):
default=LoggerType.SLF4J,
)
- # Event log level config options
+ BASE_LOG_DIR = ConfigOption(
+ key="baseLogDir",
+ config_type=str,
+ default=None,
+ )
+
+ PRETTY_PRINT = ConfigOption(
+ key="prettyPrint",
+ config_type=bool,
+ default=False,
+ )
+
+ ACTION_STATE_STORE_BACKEND = ConfigOption(
+ key="actionStateStoreBackend",
+ config_type=str,
+ default=None,
+ )
+
+ KAFKA_BOOTSTRAP_SERVERS = ConfigOption(
+ key="kafkaBootstrapServers",
+ config_type=str,
+ default="localhost:9092",
+ )
+
+ KAFKA_ACTION_STATE_TOPIC = ConfigOption(
+ key="kafkaActionStateTopic",
+ config_type=str,
+ default=None,
+ )
+
+ KAFKA_ACTION_STATE_TOPIC_NUM_PARTITIONS = ConfigOption(
+ key="kafkaActionStateTopicNumPartitions",
+ config_type=int,
+ default=64,
+ )
+
+ KAFKA_ACTION_STATE_TOPIC_REPLICATION_FACTOR = ConfigOption(
+ key="kafkaActionStateTopicReplicationFactor",
+ config_type=int,
+ default=1,
+ )
+
+ FLUSS_BOOTSTRAP_SERVERS = ConfigOption(
+ key="flussBootstrapServers",
+ config_type=str,
+ default="localhost:9123",
+ )
+
+ FLUSS_ACTION_STATE_DATABASE = ConfigOption(
+ key="flussActionStateDatabase",
+ config_type=str,
+ default="flink_agents",
+ )
+
+ FLUSS_ACTION_STATE_TABLE = ConfigOption(
+ key="flussActionStateTable",
+ config_type=str,
+ default=None,
+ )
+
+ FLUSS_ACTION_STATE_TABLE_BUCKETS = ConfigOption(
+ key="flussActionStateTableBuckets",
+ config_type=int,
+ default=64,
+ )
+
+ FLUSS_SECURITY_PROTOCOL = ConfigOption(
+ key="flussSecurityProtocol",
+ config_type=str,
+ default="PLAINTEXT",
+ )
+
+ FLUSS_SASL_MECHANISM = ConfigOption(
+ key="flussSaslMechanism",
+ config_type=str,
+ default="PLAIN",
+ )
+
+ FLUSS_SASL_JAAS_CONFIG = ConfigOption(
+ key="flussSaslJaasConfig",
+ config_type=str,
+ default=None,
+ )
+
+ FLUSS_SASL_USERNAME = ConfigOption(
+ key="flussSaslUsername",
+ config_type=str,
+ default=None,
+ )
+
+ FLUSS_SASL_PASSWORD = ConfigOption(
+ key="flussSaslPassword",
+ config_type=str,
+ default=None,
+ )
+
+ JOB_IDENTIFIER = ConfigOption(
+ key="job-identifier",
+ config_type=str,
+ default=None,
+ )
+
EVENT_LOG_LEVEL = ConfigOption(
key="event-log.level",
- config_type=str,
- default="STANDARD",
+ config_type=EventLogLevel,
+ default=EventLogLevel.STANDARD,
)
EVENT_LOG_MAX_STRING_LENGTH = ConfigOption(
@@ -148,6 +208,12 @@ class AgentConfigOptions(metaclass=AgentConfigOptionsMeta):
default=5,
)
+ EVENT_LISTENERS = ConfigOption(
+ key="event-listeners",
+ config_type=list,
+ default=None,
+ )
+
class AgentExecutionOptions:
"""Execution options for Flink Agents."""
diff --git a/python/flink_agents/api/tests/test_core_options.py
b/python/flink_agents/api/tests/test_core_options.py
new file mode 100644
index 00000000..f1e773a8
--- /dev/null
+++ b/python/flink_agents/api/tests/test_core_options.py
@@ -0,0 +1,114 @@
+################################################################################
+# 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.
+#################################################################################
+import ast
+import importlib
+import sys
+import types
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from flink_agents.api.configuration import ConfigOption
+
+
+def _clear_modules(prefixes: tuple[str, ...]) -> None:
+ for name in list(sys.modules):
+ if any(name == prefix or name.startswith(f"{prefix}.") for prefix in
prefixes):
+ sys.modules.pop(name, None)
+
+
+def test_core_options_module_does_not_import_pyflink() -> None:
+ path = Path(__file__).resolve().parents[1] / "core_options.py"
+ tree = ast.parse(path.read_text(encoding="utf-8"))
+
+ imported_modules: list[str] = []
+ for node in ast.walk(tree):
+ if isinstance(node, ast.Import):
+ imported_modules.extend(alias.name for alias in node.names)
+ elif isinstance(node, ast.ImportFrom) and node.module is not None:
+ imported_modules.append(node.module)
+
+ assert "pyflink" not in imported_modules
+ assert "pyflink.java_gateway" not in imported_modules
+
+
+def test_import_core_options_does_not_call_get_gateway() -> None:
+ _clear_modules(("flink_agents.api.core_options",))
+
+ fake_gateway_module = types.ModuleType("pyflink.java_gateway")
+ fake_gateway_module.get_gateway = MagicMock()
+
+ fake_pyflink_module = types.ModuleType("pyflink")
+ fake_pyflink_module.java_gateway = fake_gateway_module
+
+ with pytest.MonkeyPatch.context() as monkeypatch:
+ monkeypatch.setitem(sys.modules, "pyflink", fake_pyflink_module)
+ monkeypatch.setitem(
+ sys.modules, "pyflink.java_gateway", fake_gateway_module
+ )
+
+ importlib.import_module("flink_agents.api.core_options")
+
+ fake_gateway_module.get_gateway.assert_not_called()
+
+
+def test_runtime_import_chain_does_not_call_get_gateway() -> None:
+ """Guard the Pemja import path that previously failed in TaskManager
workers."""
+ pytest.importorskip("cloudpickle")
+ pytest.importorskip("pyflink")
+
+ # Only evict the modules under test. Clearing entire package trees (e.g.
+ # ``flink_agents.plan``) removes already-collected test modules from
+ # ``sys.modules`` and breaks ``inspect.getmodule()`` for functions defined
+ # in those tests.
+ _clear_modules(
+ (
+ "flink_agents.api.core_options",
+ "flink_agents.runtime.flink_runner_context",
+ )
+ )
+
+ with patch("pyflink.java_gateway.get_gateway") as get_gateway:
+ importlib.import_module("flink_agents.runtime.flink_runner_context")
+ get_gateway.assert_not_called()
+
+
+def _collect_config_options(options_class: type) -> dict[str, ConfigOption]:
+ return {
+ name: value
+ for name, value in vars(options_class).items()
+ if not name.startswith("_") and isinstance(value, ConfigOption)
+ }
+
+
+def test_agent_config_options_are_explicitly_declared() -> None:
+ from flink_agents.api.core_options import AgentConfigOptions, EventLogLevel
+
+ options = _collect_config_options(AgentConfigOptions)
+ assert len(options) == 23
+ assert options["BASE_LOG_DIR"].get_key() == "baseLogDir"
+ assert options["KAFKA_BOOTSTRAP_SERVERS"].get_default_value() ==
"localhost:9092"
+ assert options["EVENT_LOG_LEVEL"].get_default_value() is
EventLogLevel.STANDARD
+
+
+def test_unknown_agent_config_option_raises_attribute_error() -> None:
+ from flink_agents.api.core_options import AgentConfigOptions
+
+ with pytest.raises(AttributeError):
+ _ = AgentConfigOptions.UNKNOWN_OPTION
diff --git
a/python/flink_agents/plan/tests/compatibility/check_java_python_config_options_parity.py
b/python/flink_agents/plan/tests/compatibility/check_java_python_config_options_parity.py
new file mode 100644
index 00000000..0e2762d7
--- /dev/null
+++
b/python/flink_agents/plan/tests/compatibility/check_java_python_config_options_parity.py
@@ -0,0 +1,183 @@
+################################################################################
+# 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.
+#################################################################################
+"""Client-side Java/Python ConfigOption parity check.
+
+Loads the Java API JAR into a PyFlink gateway process and compares each
+explicitly declared Python ``ConfigOption`` against the Java definition.
+Invoked by ``test_java_config_in_python.sh`` (not inside the Pemja worker), so
+using ``get_gateway()`` here is intentional and separate from runtime import
+safety.
+"""
+
+from enum import Enum
+from pathlib import Path
+from typing import Any
+
+from pyflink.java_gateway import get_gateway
+from pyflink.util.java_utils import add_jars_to_context_class_loader
+
+from flink_agents.api.configuration import ConfigOption
+from flink_agents.api.core_options import AgentConfigOptions,
AgentExecutionOptions
+
+_JAVA_PRIMITIVE_TYPE_TO_PYTHON: dict[str, type] = {
+ "java.lang.String": str,
+ "java.lang.Integer": int,
+ "java.lang.Long": int,
+ "java.lang.Boolean": bool,
+ "java.lang.Float": float,
+ "java.lang.Double": float,
+ "java.util.List": list,
+}
+
+
+def collect_python_config_options(python_options_class: type) -> dict[str,
ConfigOption]:
+ """Return ``{FIELD_NAME: ConfigOption}`` declared on a Python options
class."""
+ return {
+ name: value
+ for name, value in vars(python_options_class).items()
+ if not name.startswith("_") and isinstance(value, ConfigOption)
+ }
+
+
+def collect_java_config_options(java_options_class: Any, jvm: Any) ->
dict[str, Any]:
+ """Return ``{FIELD_NAME: ConfigOption}`` from Java static fields via
reflection."""
+ modifier = jvm.java.lang.reflect.Modifier
+ class_loader = jvm.java.lang.Thread.currentThread().getContextClassLoader()
+ java_config_option_class = class_loader.loadClass(
+ "org.apache.flink.agents.api.configuration.ConfigOption"
+ )
+ options: dict[str, Any] = {}
+ for field in java_options_class.getDeclaredFields():
+ if not modifier.isStatic(field.getModifiers()):
+ continue
+ if not java_config_option_class.isAssignableFrom(field.getType()):
+ continue
+ field.setAccessible(True)
+ options[field.getName()] = field.get(None)
+ return options
+
+
+def _java_type_matches_python(java_type_name: str, python_config_type: type)
-> bool:
+ expected_primitive = _JAVA_PRIMITIVE_TYPE_TO_PYTHON.get(java_type_name)
+ java_simple_name = java_type_name.rsplit(".", maxsplit=1)[-1]
+ if "$" in java_simple_name:
+ java_simple_name = java_simple_name.split("$", maxsplit=1)[-1]
+
+ if expected_primitive is not None:
+ return python_config_type is expected_primitive
+
+ return python_config_type.__name__ == java_simple_name
+
+
+def normalize_java_default(java_default: Any, python_config_type: type) -> Any:
+ """Convert a Java default value into a Python-comparable form."""
+ if java_default is None:
+ return None
+
+ if hasattr(java_default, "name") and callable(java_default.name):
+ enum_name = java_default.name()
+ if isinstance(python_config_type, type) and
issubclass(python_config_type, Enum):
+ return python_config_type[enum_name]
+ return enum_name
+
+ if python_config_type is int and isinstance(java_default, int | float):
+ return int(java_default)
+
+ return java_default
+
+
+def assert_python_option_matches_java(
+ option_name: str,
+ python_option: ConfigOption,
+ java_option: Any,
+) -> None:
+ """Assert key, type, and default parity for one option pair."""
+ python_config_type = python_option.get_type()
+ java_type_name = java_option.getTypeName()
+
+ assert python_option.get_key() == java_option.getKey(), (
+ f"{option_name}: key mismatch "
+ f"(python={python_option.get_key()!r}, java={java_option.getKey()!r})"
+ )
+ assert _java_type_matches_python(java_type_name, python_config_type), (
+ f"{option_name}: type mismatch "
+ f"(python={python_config_type!r}, java={java_type_name!r})"
+ )
+
+ python_default = python_option.get_default_value()
+ java_default = normalize_java_default(
+ java_option.getDefaultValue(),
+ python_config_type,
+ )
+ assert python_default == java_default, (
+ f"{option_name}: default mismatch "
+ f"(python={python_default!r}, java={java_default!r})"
+ )
+
+
+def assert_options_class_matches_java(
+ python_options_class: type,
+ java_options_class: Any,
+ jvm: Any,
+) -> None:
+ """Compare Python and Java ConfigOption declarations in both directions."""
+ python_options = collect_python_config_options(python_options_class)
+ java_options = collect_java_config_options(java_options_class, jvm)
+
+ python_names = set(python_options)
+ java_names = set(java_options)
+ assert python_names == java_names, (
+ f"{python_options_class.__name__}: option field name mismatch "
+ f"(python-only={sorted(python_names - java_names)!r}, "
+ f"java-only={sorted(java_names - python_names)!r})"
+ )
+
+ for option_name in sorted(python_names):
+ assert_python_option_matches_java(
+ option_name,
+ python_options[option_name],
+ java_options[option_name],
+ )
+
+
+def main() -> None:
+ current_dir = Path(__file__).parent
+
+ jars =
Path(current_dir).glob("../../../../../api/target/flink-agents-api-*.jar")
+ jar_urls = [f"file:///{jar.resolve()}" for jar in jars]
+ add_jars_to_context_class_loader(jar_urls)
+
+ jvm = get_gateway().jvm
+ class_loader = jvm.java.lang.Thread.currentThread().getContextClassLoader()
+ java_agent_config_options = class_loader.loadClass(
+ "org.apache.flink.agents.api.configuration.AgentConfigOptions"
+ )
+ java_agent_execution_options = class_loader.loadClass(
+ "org.apache.flink.agents.api.agents.AgentExecutionOptions"
+ )
+
+ assert_options_class_matches_java(
+ AgentConfigOptions, java_agent_config_options, jvm
+ )
+ assert_options_class_matches_java(
+ AgentExecutionOptions, java_agent_execution_options, jvm
+ )
+
+
+if __name__ == "__main__":
+ main()
diff --git
a/python/flink_agents/plan/tests/compatibility/create_python_option_from_java_option.py
b/python/flink_agents/plan/tests/compatibility/create_python_option_from_java_option.py
deleted file mode 100644
index b7251ee1..00000000
---
a/python/flink_agents/plan/tests/compatibility/create_python_option_from_java_option.py
+++ /dev/null
@@ -1,79 +0,0 @@
-################################################################################
-# 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.
-#################################################################################
-from pathlib import Path
-
-from pyflink.util.java_utils import add_jars_to_context_class_loader
-
-from flink_agents.api.core_options import (
- AgentConfigOptions,
- AgentExecutionOptions,
- ShortTermMemoryTtlUpdate,
- ShortTermMemoryTtlVisibility,
-)
-
-# This script is used to verify that Java-defined configuration options
-# (e.g., AgentConfigOptions) are correctly exposed and accessible in the
-# Python environment via the JAR file. It loads a Java JAR into the Python
-# context and performs basic assertions on the configuration keys, types,
-# and default values to ensure compatibility between Java and Python layers.
-#
-# The JAR file path is relative to this script and should be updated if
-# the build structure changes.
-if __name__ == "__main__":
- current_dir = Path(__file__).parent
-
- jars =
Path(current_dir).glob("../../../../../api/target/flink-agents-api-*.jar")
- jars = [f"file:///{jar}" for jar in jars]
- add_jars_to_context_class_loader(jars)
-
- assert AgentConfigOptions.BASE_LOG_DIR.get_key() == "baseLogDir"
- assert AgentConfigOptions.BASE_LOG_DIR.get_type() is str
- assert AgentConfigOptions.BASE_LOG_DIR.get_default_value() is None
-
- assert (
- AgentExecutionOptions.SHORT_TERM_MEMORY_STATE_TTL_MS.get_key()
- == "short-term-memory.state-ttl.ms"
- )
- assert AgentExecutionOptions.SHORT_TERM_MEMORY_STATE_TTL_MS.get_type() is
int
- assert
AgentExecutionOptions.SHORT_TERM_MEMORY_STATE_TTL_MS.get_default_value() == 0
-
- assert (
- AgentExecutionOptions.SHORT_TERM_MEMORY_STATE_TTL_UPDATE_TYPE.get_key()
- == "short-term-memory.state-ttl.update-type"
- )
- assert (
-
AgentExecutionOptions.SHORT_TERM_MEMORY_STATE_TTL_UPDATE_TYPE.get_type()
- is ShortTermMemoryTtlUpdate
- )
- assert (
-
AgentExecutionOptions.SHORT_TERM_MEMORY_STATE_TTL_UPDATE_TYPE.get_default_value()
- is ShortTermMemoryTtlUpdate.ON_READ_AND_WRITE
- )
-
- assert (
- AgentExecutionOptions.SHORT_TERM_MEMORY_STATE_TTL_VISIBILITY.get_key()
- == "short-term-memory.state-ttl.visibility"
- )
- assert (
- AgentExecutionOptions.SHORT_TERM_MEMORY_STATE_TTL_VISIBILITY.get_type()
- is ShortTermMemoryTtlVisibility
- )
- assert (
-
AgentExecutionOptions.SHORT_TERM_MEMORY_STATE_TTL_VISIBILITY.get_default_value()
- is ShortTermMemoryTtlVisibility.NEVER_RETURN_EXPIRED
- )
