This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 99a84bc9ec [Fix][CLI] Fix Bedrock compatibility: temperature rejection
and engine template variables (#11509)
99a84bc9ec is described below
commit 99a84bc9ecf8f43557db941b5054d1a0a1361cbe
Author: SEZ <[email protected]>
AuthorDate: Thu Jul 23 23:17:53 2026 +0800
[Fix][CLI] Fix Bedrock compatibility: temperature rejection and engine
template variables (#11509)
Co-authored-by: SEZ9 <[email protected]>
---
seatunnel-cli/README.md | 19 +++
seatunnel-cli/seatunnel_cli/agents.py | 27 +++-
seatunnel-cli/seatunnel_cli/llm_provider.py | 84 +++++++---
seatunnel-cli/tests/test_llm_provider_bedrock.py | 177 +++++++++++++++++++++
.../tests/test_validate_hocon_placeholders.py | 105 ++++++++++++
5 files changed, 385 insertions(+), 27 deletions(-)
diff --git a/seatunnel-cli/README.md b/seatunnel-cli/README.md
index 17f81c02af..dc285199be 100644
--- a/seatunnel-cli/README.md
+++ b/seatunnel-cli/README.md
@@ -105,6 +105,12 @@ export
ANTHROPIC_SMALL_FAST_MODEL='us.anthropic.claude-haiku-4-5-20251001-v1:0'
The Bedrock provider preserves Claude `reasoningContent` blocks in streamed
Converse responses when Bedrock returns them.
+> **Note:** Some newer Claude models on Bedrock reject the `temperature`
+> inference parameter. When a model returns this rejection, the provider
+> automatically retries the request without `temperature` and remembers the
+> model for the rest of the session, so subsequent calls skip the parameter.
+> For these models any configured temperature value is not applied.
+
#### Option B: Anthropic API
```bash
@@ -395,6 +401,19 @@ Memory is stored locally at `.data/memory.json`
(co-located with the CLI package
| **Auto-fix** | LLM-powered | Up to 3 rounds of automatic error correction
during generation |
| **Auto-repair** | LLM-powered | Automatic diagnosis and config patching on
`/check` or `/run` failure |
+Local validation flags unresolved `${VAR}` placeholders as missing environment
+variables, with a field-aware exemption for SeaTunnel's engine-resolved file
+sink template placeholders (see the
+[LocalFile sink
docs](https://seatunnel.apache.org/docs/connectors/sink/LocalFile)):
+
+| Field | Engine placeholders allowed |
+|-------|-----------------------------|
+| `file_name_expression` | `${now}`, `${uuid}`, `${transactionId}` |
+| `partition_dir_expression` | `${k0}`, `${v0}`, `${k1}`, `${v1}`, ... |
+
+The same names used in any other field (URLs, credentials, paths) are treated
+as regular environment variables and reported if unset.
+
## Connector Metadata
The CLI ships with `connector_metadata.json`, exported from the SeaTunnel
engine via runtime reflection. It includes source, sink, and transform plugin
option rules, conditional options, and value constraints. No extra steps needed.
diff --git a/seatunnel-cli/seatunnel_cli/agents.py
b/seatunnel-cli/seatunnel_cli/agents.py
index 89c5010a4c..14caee8a04 100644
--- a/seatunnel-cli/seatunnel_cli/agents.py
+++ b/seatunnel-cli/seatunnel_cli/agents.py
@@ -526,12 +526,31 @@ def validate_hocon(config_str: str) -> str:
# Check for unresolved ${ENV_VAR} placeholders — these will be passed as
literal
# strings to connectors at runtime, causing authentication/connection
failures.
+ # SeaTunnel's engine resolves certain placeholders itself, but only in
+ # specific file sink fields (see docs/en/connectors/sink/LocalFile.md):
+ # file_name_expression -> ${now}, ${uuid}, ${transactionId}
+ # partition_dir_expression -> ${k0}=${v0}/... (kN = partition field,
+ # vN = partition value)
+ # The exemption is field-aware so that the same names used in unrelated
+ # fields (URLs, credentials, ...) are still diagnosed as env vars.
+ engine_template_fields = {
+ "file_name_expression": re.compile(r"now|uuid|transactionId"),
+ "partition_dir_expression": re.compile(r"[kv]\d+"),
+ }
env_var_pattern = re.compile(r'\$\{([A-Za-z_][A-Za-z0-9_]*)\}')
+ # HOCON accepts both `key = value` and `key : value` separators
+ field_pattern = re.compile(r'^\s*"?([\w.]+)"?\s*[=:]')
unresolved_vars = set()
- for m in env_var_pattern.finditer(config_str):
- var_name = m.group(1)
- if not os.environ.get(var_name):
- unresolved_vars.add(var_name)
+ for line in config_str.splitlines():
+ field_match = field_pattern.match(line)
+ allowed = engine_template_fields.get(
+ field_match.group(1)) if field_match else None
+ for m in env_var_pattern.finditer(line):
+ var_name = m.group(1)
+ if allowed and allowed.fullmatch(var_name):
+ continue
+ if not os.environ.get(var_name):
+ unresolved_vars.add(var_name)
if unresolved_vars:
var_list = ", ".join(sorted(unresolved_vars))
errors.append(
diff --git a/seatunnel-cli/seatunnel_cli/llm_provider.py
b/seatunnel-cli/seatunnel_cli/llm_provider.py
index ef91498c0b..8d547666bc 100644
--- a/seatunnel-cli/seatunnel_cli/llm_provider.py
+++ b/seatunnel-cli/seatunnel_cli/llm_provider.py
@@ -29,6 +29,11 @@ Selection: set AI_PROVIDER env var, or run --init for
interactive setup.
"""
import abc
+
+try:
+ import botocore.exceptions
+except ImportError: # boto3/botocore optional (bedrock extra)
+ botocore = None
import json
import logging
import os
@@ -330,6 +335,43 @@ class BedrockProvider(LLMProvider):
if endpoint_url:
client_kwargs["endpoint_url"] = endpoint_url
self.client = session.client(**client_kwargs)
+ # Models that rejected `temperature` (newer Claude models deprecate
it).
+ # Populated on first rejection so subsequent calls skip the retry.
+ self._no_temperature_models: set[str] = set()
+
+ def _build_kwargs(self, model_id, messages, system, temperature,
max_tokens, tools) -> dict:
+ inference_config = {"maxTokens": max_tokens}
+ if model_id not in self._no_temperature_models:
+ inference_config["temperature"] = temperature
+ kwargs = {
+ "modelId": model_id,
+ "messages": messages,
+ "inferenceConfig": inference_config,
+ }
+ if system:
+ kwargs["system"] = [{"text": system}]
+ if tools:
+ kwargs["toolConfig"] = {"tools": tools}
+ return kwargs
+
+ @staticmethod
+ def _is_temperature_rejection(error: Exception) -> bool:
+ """True if this is Bedrock's structured rejection of `temperature`.
+
+ Checks the botocore ClientError code (precise) and only uses the
+ message to discriminate that `temperature` is the rejected parameter
+ (tolerant to wording changes such as "deprecated" vs "not supported").
+ """
+ try:
+ from botocore.exceptions import ClientError
+ except ImportError:
+ return False
+ if not isinstance(error, ClientError):
+ return False
+ code = error.response.get("Error", {}).get("Code", "")
+ if code != "ValidationException":
+ return False
+ return "temperature" in str(error).lower()
@property
def provider_name(self) -> str:
@@ -353,18 +395,15 @@ class BedrockProvider(LLMProvider):
tools: list[dict] | None = None,
) -> dict:
model_id = model or self._model_id
- kwargs = {
- "modelId": model_id,
- "messages": messages,
- "inferenceConfig": {"temperature": temperature, "maxTokens":
max_tokens},
- }
- if system:
- kwargs["system"] = [{"text": system}]
- if tools:
- kwargs["toolConfig"] = {"tools": tools}
-
- response = self.client.converse(**kwargs)
- return response
+ kwargs = self._build_kwargs(model_id, messages, system, temperature,
max_tokens, tools)
+ try:
+ return self.client.converse(**kwargs)
+ except botocore.exceptions.ClientError as e:
+ if self._is_temperature_rejection(e):
+ self._no_temperature_models.add(model_id)
+ kwargs["inferenceConfig"].pop("temperature", None)
+ return self.client.converse(**kwargs)
+ raise
def chat_stream(
self,
@@ -376,17 +415,16 @@ class BedrockProvider(LLMProvider):
tools: list[dict] | None = None,
) -> Generator[dict, None, None]:
model_id = model or self._model_id
- kwargs = {
- "modelId": model_id,
- "messages": messages,
- "inferenceConfig": {"temperature": temperature, "maxTokens":
max_tokens},
- }
- if system:
- kwargs["system"] = [{"text": system}]
- if tools:
- kwargs["toolConfig"] = {"tools": tools}
-
- response = self.client.converse_stream(**kwargs)
+ kwargs = self._build_kwargs(model_id, messages, system, temperature,
max_tokens, tools)
+ try:
+ response = self.client.converse_stream(**kwargs)
+ except botocore.exceptions.ClientError as e:
+ if self._is_temperature_rejection(e):
+ self._no_temperature_models.add(model_id)
+ kwargs["inferenceConfig"].pop("temperature", None)
+ response = self.client.converse_stream(**kwargs)
+ else:
+ raise
current_tool_id = None
for event in response.get("stream", []):
if "contentBlockStart" in event:
diff --git a/seatunnel-cli/tests/test_llm_provider_bedrock.py
b/seatunnel-cli/tests/test_llm_provider_bedrock.py
new file mode 100644
index 0000000000..b73589c4fb
--- /dev/null
+++ b/seatunnel-cli/tests/test_llm_provider_bedrock.py
@@ -0,0 +1,177 @@
+#
+# 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.
+#
+
+"""Regression tests for the BedrockProvider temperature-rejection retry
+(chat and chat_stream paths)."""
+
+import copy
+from unittest import mock
+
+import pytest
+
+botocore = pytest.importorskip("botocore")
+from botocore.exceptions import ClientError # noqa: E402
+
+OK_RESPONSE = {
+ "output": {"message": {"role": "assistant", "content": [{"text": "OK"}]}},
+ "stopReason": "end_turn",
+}
+
+OK_STREAM = {"stream": []}
+
+
+def _temperature_rejection() -> ClientError:
+ return ClientError(
+ error_response={
+ "Error": {
+ "Code": "ValidationException",
+ "Message": "The model returned the following errors: "
+ "`temperature` is deprecated for this model.",
+ }
+ },
+ operation_name="Converse",
+ )
+
+
+def _other_validation_error() -> ClientError:
+ return ClientError(
+ error_response={
+ "Error": {
+ "Code": "ValidationException",
+ "Message": "The provided model identifier is invalid.",
+ }
+ },
+ operation_name="Converse",
+ )
+
+
+def _throttling_error() -> ClientError:
+ return ClientError(
+ error_response={
+ "Error": {"Code": "ThrottlingException",
+ "Message": "Too many requests, temperature unrelated"}
+ },
+ operation_name="Converse",
+ )
+
+
+def _make_provider(converse_effects=None, stream_effects=None):
+ """Build a BedrockProvider with a stubbed boto3 client.
+
+ The retry path mutates the kwargs dict in place before the second call,
+ so mock's recorded call_args (which stores references) can't be used to
+ inspect the first call's inferenceConfig. Record deep snapshots instead.
+ """
+ with mock.patch.dict("sys.modules", {"boto3": mock.MagicMock()}):
+ from seatunnel_cli.llm_provider import BedrockProvider
+ provider = BedrockProvider.__new__(BedrockProvider)
+ provider._model_id = "test-model"
+ provider._fast_model_id = "test-model"
+ provider._no_temperature_models = set()
+ provider.client = mock.MagicMock()
+ provider.converse_snapshots = []
+ provider.stream_snapshots = []
+
+ def _run(effects, snapshots):
+ def call(**kwargs):
+ snapshots.append(copy.deepcopy(kwargs))
+ effect = effects.pop(0)
+ if isinstance(effect, Exception):
+ raise effect
+ return effect
+ return call
+
+ if converse_effects is not None:
+ provider.client.converse.side_effect = _run(
+ list(converse_effects), provider.converse_snapshots)
+ if stream_effects is not None:
+ provider.client.converse_stream.side_effect = _run(
+ list(stream_effects), provider.stream_snapshots)
+ return provider
+
+
+# ── chat (non-streaming) ──
+
+def test_chat_temperature_rejection_retries_without_temperature():
+ provider = _make_provider(
+ converse_effects=[_temperature_rejection(), OK_RESPONSE, OK_RESPONSE])
+
+ resp = provider.chat([{"role": "user", "content": [{"text": "hi"}]}])
+ assert resp["stopReason"] == "end_turn"
+
+ snaps = provider.converse_snapshots
+ assert len(snaps) == 2
+ assert "temperature" in snaps[0]["inferenceConfig"]
+ assert "temperature" not in snaps[1]["inferenceConfig"]
+ assert "test-model" in provider._no_temperature_models
+
+ # model id cached: the next call must not send temperature at all
+ provider.chat([{"role": "user", "content": [{"text": "hi"}]}])
+ assert "temperature" not in
provider.converse_snapshots[2]["inferenceConfig"]
+
+
+def test_chat_other_validation_errors_are_not_swallowed():
+ provider = _make_provider(converse_effects=[_other_validation_error()])
+ with pytest.raises(ClientError):
+ provider.chat([{"role": "user", "content": [{"text": "hi"}]}])
+ assert len(provider.converse_snapshots) == 1
+ assert not provider._no_temperature_models
+
+
+def test_chat_non_validation_client_errors_are_not_swallowed():
+ provider = _make_provider(converse_effects=[_throttling_error()])
+ with pytest.raises(ClientError):
+ provider.chat([{"role": "user", "content": [{"text": "hi"}]}])
+ assert len(provider.converse_snapshots) == 1
+ assert not provider._no_temperature_models
+
+
+def test_chat_non_client_errors_propagate():
+ provider = _make_provider(converse_effects=[RuntimeError("boom")])
+ with pytest.raises(RuntimeError):
+ provider.chat([{"role": "user", "content": [{"text": "hi"}]}])
+ assert not provider._no_temperature_models
+
+
+# ── chat_stream (the CLI's interactive path — scenario from #11508) ──
+
+def test_chat_stream_temperature_rejection_retries_without_temperature():
+ provider = _make_provider(
+ stream_effects=[_temperature_rejection(), OK_STREAM, OK_STREAM])
+
+ events = list(provider.chat_stream(
+ [{"role": "user", "content": [{"text": "hi"}]}]))
+ assert events == [] # empty stream consumed without error
+
+ snaps = provider.stream_snapshots
+ assert len(snaps) == 2
+ assert "temperature" in snaps[0]["inferenceConfig"]
+ assert "temperature" not in snaps[1]["inferenceConfig"]
+ assert "test-model" in provider._no_temperature_models
+
+ # cached: next stream call must not send temperature
+ list(provider.chat_stream([{"role": "user", "content": [{"text": "hi"}]}]))
+ assert "temperature" not in provider.stream_snapshots[2]["inferenceConfig"]
+
+
+def test_chat_stream_other_errors_are_not_swallowed():
+ provider = _make_provider(stream_effects=[_throttling_error()])
+ with pytest.raises(ClientError):
+ list(provider.chat_stream(
+ [{"role": "user", "content": [{"text": "hi"}]}]))
+ assert len(provider.stream_snapshots) == 1
+ assert not provider._no_temperature_models
diff --git a/seatunnel-cli/tests/test_validate_hocon_placeholders.py
b/seatunnel-cli/tests/test_validate_hocon_placeholders.py
new file mode 100644
index 0000000000..4e9a32a634
--- /dev/null
+++ b/seatunnel-cli/tests/test_validate_hocon_placeholders.py
@@ -0,0 +1,105 @@
+#
+# 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.
+#
+
+"""Regression tests for field-aware engine placeholder handling in
validate_hocon."""
+
+import os
+from unittest import mock
+
+import pytest
+
+from seatunnel_cli.agents import validate_hocon
+
+
+def _config_with_sink(sink_body: str) -> str:
+ return f"""
+env {{ parallelism = 1
+ job.mode = "BATCH" }}
+source {{ FakeSource {{ row.num = 5
+ schema {{ fields {{ id = "bigint" }} }}
+ plugin_output = "f" }} }}
+sink {{ LocalFile {{ plugin_input = "f"
+ path = "/tmp/out"
+ {sink_body} }} }}
+"""
+
+
+# ── engine placeholders accepted only in their own fields ──
+
+def test_file_name_expression_engine_placeholders_accepted():
+ for expr in ("${now}", "${uuid}", "${transactionId}",
"out_${uuid}_${now}"):
+ config = _config_with_sink(f'file_name_expression = "{expr}"')
+ result = validate_hocon(config)
+ assert "Unresolved environment variables" not in result, expr
+
+
+def test_partition_dir_expression_engine_placeholders_accepted():
+ config = _config_with_sink(
+ 'partition_dir_expression = "${k0}=${v0}/${k1}=${v1}"')
+ result = validate_hocon(config)
+ assert "Unresolved environment variables" not in result
+
+
+# ── the same names in unrelated fields are still diagnosed ──
+
[email protected]("field_line, var", [
+ ('url = "jdbc:mysql://${now}:3306/db"', "now"),
+ ('password = "${uuid}"', "uuid"),
+ ('topic = "${transactionId}"', "transactionId"),
+ ('path = "/data/${k0}"', "k0"),
+])
+def test_engine_placeholder_names_rejected_outside_their_fields(field_line,
var):
+ assert var not in os.environ
+ config = _config_with_sink(field_line)
+ result = validate_hocon(config)
+ assert "Unresolved environment variables" in result
+ assert var in result
+
+
+def test_unset_env_var_still_rejected_in_expression_fields():
+ # A non-engine placeholder inside file_name_expression is still an env var
+ assert "MY_UNSET_PREFIX" not in os.environ
+ config = _config_with_sink(
+ 'file_name_expression = "${MY_UNSET_PREFIX}_${now}"')
+ result = validate_hocon(config)
+ assert "Unresolved environment variables" in result
+ assert "MY_UNSET_PREFIX" in result
+
+
+def test_set_env_var_accepted_anywhere():
+ with mock.patch.dict(os.environ, {"MYSQL_PASSWORD": "x"}):
+ config = _config_with_sink('password = "${MYSQL_PASSWORD}"')
+ result = validate_hocon(config)
+ assert "Unresolved environment variables" not in result
+
+# ── HOCON colon separator (key : value) is equally valid ──
+
+def test_colon_separator_engine_placeholders_accepted():
+ config = _config_with_sink('file_name_expression: "${now}"')
+ result = validate_hocon(config)
+ assert "Unresolved environment variables" not in result
+
+ config = _config_with_sink('partition_dir_expression: "${k0}=${v0}"')
+ result = validate_hocon(config)
+ assert "Unresolved environment variables" not in result
+
+
+def test_colon_separator_still_rejects_env_vars_elsewhere():
+ assert "now" not in os.environ
+ config = _config_with_sink('topic: "${now}"')
+ result = validate_hocon(config)
+ assert "Unresolved environment variables" in result
