ChickenBenny commented on code in PR #76359:
URL: https://github.com/apache/airflow/pull/76359#discussion_r4549123295
##########
task-sdk/src/airflow/sdk/coordinators/_subprocess.py:
##########
@@ -562,8 +562,18 @@ def _prepare_ti_context(self, ti: TaskInstance,
ti_context: TIRunContext) -> TIR
backend's storage.
If an XCom cannot be deleted, this try fails and the task does not run.
+
+ Also fills in the worker's default ``execution_timeout`` when the task
sets none.
"""
- self._execution_timeout_seconds = ti_context.execution_timeout
+ # A task that sets no execution_timeout gets the worker's default, as
a Python task does when the
+ # worker parses its Dag. A zero means no timeout, so it does not fall
back.
+ execution_timeout = ti_context.execution_timeout
+ if (
+ execution_timeout is None
+ and (default := conf.gettimedelta("core",
"default_task_execution_timeout")) is not None
Review Comment:
Fixed in d97f392660. `SubprocessCoordinator.execute_task` now reads the
default inside the existing `_raise_as_launch_error` block, before it launches
the runtime or starts the TI, and passes the value to
`_PopenActivitySubprocess.start`. `_prepare_ti_context` only applies it. A
launch failure now happens before the TI is started, the same rule #75561
follows.
A value that is not a number raises `TaskLaunchError(..., retryable=False)`
explicitly instead of going through the catch-all, because a retry on the same
worker reads the same value.
##########
go-sdk/pkg/execution/iface.go:
##########
@@ -0,0 +1,99 @@
+// 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 execution
+
+import "time"
+
+// ifaceString returns the string carried by a nullable schema field decoded as
+// any, or "" when the value is nil or not a string.
+func ifaceString(v any) string {
+ s, _ := v.(string)
+ return s
+}
+
+// ifaceStringPtr preserves the null-vs-empty distinction for a nullable string
+// field decoded as any: nil when missing/null, else a pointer to the string.
+// Used for connection credentials, where an explicit "" must stay distinct
from
+// "not set".
+func ifaceStringPtr(v any) *string {
+ s, ok := v.(string)
+ if !ok {
+ return nil
+ }
+ return &s
+}
+
+// ifaceTimePtr returns a pointer to the time carried by a nullable date-time
+// field decoded as any, or nil when missing/null. msgpack decodes the
+// supervisor's timestamp extension straight into time.Time.
+func ifaceTimePtr(v any) *time.Time {
+ t, ok := v.(time.Time)
+ if !ok {
+ return nil
+ }
+ return &t
+}
+
+// ifaceInt converts a numeric value decoded as any into int, returning def
when
+// nil or non-numeric.
Review Comment:
Applied in 8b7cc63506.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
