Add Dag run context fields to the Java SDK Context#69100
Conversation
|
I was partly doing this intentionally since all those context keys are quite redundant IMO. Maybe it’d be easier if we create a DagRun object (from the model in ti context, but not that type directly; similar to how Connection and Variable are bridged) to build Context with, and tell users to just use e.g. Note that Java SDK does not need to support Jinja templating, so a top-level key in the context is not particularly useful. |
jason810496
left a comment
There was a problem hiding this comment.
I was partly doing this intentionally since all those context keys are quite redundant IMO. Maybe it’d be easier if we create a DagRun object (from the model in ti context, but not that type directly; similar to how Connection and Variable are bridged) to build Context with, and tell users to just use e.g. context.dagRun.runId instead?
Isn't this PR doing exactly what the above part you mentioned?
Note that Java SDK does not need to support Jinja templating, so a top-level key in the context is not particularly useful.
Or do you mean since Java SDK does not need to support Jinja templating so we don't need to expose logicalDate, dataIntervalStart, etc, in the context?
IMO, no harm to expose those snapshotted attributes to user in the injected context.
9a8c2d1 to
cc2a671
Compare
Agree, I use DagRun object (
Yes, Java tasks have no Jinja/operator-templating path, so a task doing a backfill by data interval can only reach those through the context object. |
cc2a671 to
2b65abe
Compare
Java tasks could only read the dag id and run id from Context, even though the supervisor already sends the full run context (logical date, data interval, run type, trigger conf) in StartupDetails. Signed-off-by: PoAn Yang <payang@apache.org>
2b65abe to
2b008ee
Compare
| runCatching { RunType.valueOf(dagRunType.name) }.getOrElse { | ||
| logger.warning("Ignoring unrecognized run type in run context", mapOf("value" to dagRunType.name)) | ||
| null | ||
| } |
There was a problem hiding this comment.
This would drift out of sync too easily if DagRunType receives new values. Use a simple when to make the compiler check for missing items instead.
| } | ||
| } | ||
|
|
||
| enum class RunType { |
There was a problem hiding this comment.
I would name this DagRunType to match the API and the Python type.
| @Test | ||
| fun fromDecodesRunContextFields() { | ||
| // [2, msg, null] with msg coming from | ||
| // https://github.com/astronomer/airflow/blob/f39c8da8/task-sdk/tests/task_sdk/execution_time/test_comms.py#L73-L108 |
There was a problem hiding this comment.
Reference the apache/airflow repo instead.
The Java SDK's
Contextonly exposeddagIdandrunIdfrom the Dag run, even though the supervisor already delivers the full run context in the sameStartupDetailsthat Python tasks receive. Add fieldslogicalDate,dataIntervalStart/dataIntervalEnd,runAfter,ds,runType, andconftoContext.Was generative AI tooling used to co-author this PR?
{pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.