jason810496 commented on code in PR #73420:
URL: https://github.com/apache/airflow/pull/73420#discussion_r4193124778
##########
go-sdk/pkg/execution/client.go:
##########
@@ -57,15 +71,63 @@ func translateApiError(err error, code string, sentinel
error, key string) error
// over the comm socket using msgpack-framed IPC instead of HTTP.
type CoordinatorClient struct {
comm *CoordinatorComm
+ // Bound at construction, not per call: the Execution API scopes the
task
+ // state store to the caller's own task instance ("ti:self").
+ tiID string
}
var _ sdk.Client = (*CoordinatorClient)(nil)
// NewCoordinatorClient creates a new client backed by the comm socket.
-func NewCoordinatorClient(comm *CoordinatorComm) *CoordinatorClient {
+func NewCoordinatorClient(comm *CoordinatorComm, tiID string)
*CoordinatorClient {
return &CoordinatorClient{
comm: comm,
+ tiID: tiID,
+ }
+}
+
+// resolveDefaultExpiry returns nil ("never expires") for a retention of 0.
+// Only an absent env value falls back (the runtime was not launched by the
+// coordinator); a malformed one fails as it does in Python rather than
silently
+// retaining keys for a different period.
+func resolveDefaultExpiry(now time.Time) (any, error) {
+ days := fallbackRetentionDays
+ if raw, ok := os.LookupEnv(defaultRetentionDaysEnv); ok {
Review Comment:
The Go runtime should be able to get the worker side `conf` via the
`AIRFLOW__STATE_STORE__DEFAULT_RETENTION_DAYS` env anyway.
Let's remove the `fallbackRetentionDays` fallback to prevent the further
inconsistency.
--
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]
