cool9850311 opened a new pull request, #73420: URL: https://github.com/apache/airflow/pull/73420
## Why The language SDK specification (`contributing-docs/30_new_language_sdk.rst`) lists `task-state-store` as a runtime capability: a task must be able to read and write the per-task-instance state store. The Go SDK declared it unsupported. Python has exposed the same store since 3.3 as `context["task_state_store"]`, and it is what durable execution is built on — the store is scoped to `dag_id` + `run_id` + `task_id` + `map_index` and deliberately not to `try_number`, so a task that records an external job ID can reattach to it after a worker crash instead of submitting the job twice. The supervisor already handles `GetTaskStateStore` / `SetTaskStateStore` / `DeleteTaskStateStore` / `ClearTaskStateStore` for Python tasks, and the matching message structs were already generated in `genmodels`, but Go tasks had no API to send them. ## What - `go-sdk/sdk/sdk.go`, `go-sdk/sdk/errors.go`: add `TaskStateStoreClient` — `GetTaskState`, `UnmarshalJSONTaskState`, `SetTaskState`, `SetTaskStateWithRetention`, `DeleteTaskState`, `ClearTaskState` — composed into `sdk.Client`, plus a `TaskStateNotFound` sentinel and `NeverExpire`. Retention is split across two setters because Go has no default arguments, so a zero or negative `retention` is rejected rather than given a meaning of its own. - `go-sdk/pkg/execution/client.go`: implement them on `CoordinatorClient`. The task instance id is bound at construction rather than taken per call, because the Execution API confines the store to the caller's own task instance (the `ti:self` scope); a per-call task instance would advertise an addressing freedom the API does not grant. A value the store cannot hold — `time.Time`, a non-finite float, `[]byte` — is rejected before the frame is sent, matching the Pydantic validation Python applies on the worker. A struct is still accepted, because msgpack encodes it as a map and it arrives as a JSON object. - `task-sdk/src/airflow/sdk/coordinators/_subprocess.py`: propagate the resolved `[state_store] default_retention_days` to the subprocess in the environment at launch, next to the log levels already passed there. Expiry is computed on the worker in Python and a language SDK runtime cannot read Airflow's config, so without this every key written from Go would carry a null expiry and never be garbage collected. A misconfigured value fails the write, as it does for a Python task, rather than substituting a different lifetime. - `go-sdk/example/bundle/taskstate/`, `go-sdk/example/bundle/main.go`, `go-sdk/dags/go_examples.py`: a new `task_state_dag` example whose Go task writes with both setters, reads back with both readers, and deletes a scratch key. - `go-sdk/capabilities.yaml`: mark `task-state-store` supported since 3.4. Adding methods to `sdk.Client` means user-defined fakes that implement this interface must add the six methods. The example mocks in this repo are updated. The Go SDK does not implement the worker state backend (`[workers] state_store_backend`). A deployment that configures one gets the raw reference marker back in a Go task, which is what a Python worker without that backend configured also sees; `go.rst` says so. ## Verification - `cd go-sdk && go test ./... -race -count=1` - `uv run --project task-sdk pytest task-sdk/tests/task_sdk/coordinators/test_subprocess.py` - `prek run --from-ref upstream/main --stage pre-commit` - `breeze testing airflow-e2e-tests --e2e-test-mode go_sdk` (30 tests, the whole `go_sdk_tests` directory) - `breeze build-docs apache-airflow --docs-only` --- ##### Was generative AI tooling used to co-author this PR? - [X] Yes — Claude Code (Opus 5) -- 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]
