Skip to content

Remove workers.celery breaking change #61049

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jedcunningham merged 2 commits into from
Jan 26, 2026
Merged

Remove workers.celery breaking change #61049

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
cc5452e
Change default to nil
Miretpl Jan 25, 2026
e0c0cc3
Update tests
Miretpl Jan 25, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
90 changes: 57 additions & 33 deletions chart/values.schema.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2688,8 +2688,11 @@
"properties": {
"replicas": {
"description": "Number of Airflow Celery workers.",
"type": "integer",
"default": 1
"type": [
"integer",
"null"
],
"default": null
},
"revisionHistoryLimit": {
"description": "Max number of old Airflow Celery workers ReplicaSets to retain.",
Expand Down Expand Up @@ -2720,11 +2723,7 @@
"items": {
"type": "string"
},
"default": [
"bash",
"-c",
"exec \\\nairflow {{ semverCompare \">=2.0.0\" .Values.airflowVersion | ternary \"celery worker\" \"worker\" }}\n{{- if and .Values.workers.queue (ne .Values.workers.queue \"default\") }}\n{{- \" -q \" }}{{ .Values.workers.queue }}\n{{- end }}"
]
"default": null
},
"livenessProbe": {
"description": "Liveness probe configuration for Airflow Celery worker containers.",
Expand All @@ -2733,28 +2732,43 @@
"properties": {
"enabled": {
"description": "Enable liveness probe for Airflow Celery workers.",
"type": "boolean",
"default": true
"type": [
"boolean",
"null"
],
"default": null
},
"initialDelaySeconds": {
"description": "Number of seconds after the container has started before liveness probes are initiated.",
"type": "integer",
"default": 10
"type": [
"integer",
"null"
],
"default": null
},
"timeoutSeconds": {
"description": "Number of seconds after which the probe times out. Minimum value is 1 seconds.",
"type": "integer",
"default": 20
"type": [
"integer",
"null"
],
"default": null
},
"failureThreshold": {
"description": "Minimum consecutive failures for the probe to be considered failed after having succeeded. Minimum value is 1.",
"type": "integer",
"default": 5
"type": [
"integer",
"null"
],
"default": null
},
"periodSeconds": {
"description": "How often (in seconds) to perform the probe. Minimum value is 1.",
"type": "integer",
"default": 60
"type": [
"integer",
"null"
],
"default": null
},
"command": {
"description": "Command for livenessProbe",
Expand Down Expand Up @@ -2782,12 +2796,7 @@
"null",
"object"
],
"default": {
"rollingUpdate": {
"maxSurge": "100%",
"maxUnavailable": "50%"
}
}
"default": null
},
"podManagementPolicy": {
"description": "Specifies the policy for managing pods within the Airflow Celery worker. Only applicable to StatefulSet.",
Expand Down Expand Up @@ -2863,17 +2872,23 @@
"properties": {
"enabled": {
"description": "Enable persistent volumes.",
"type": "boolean",
"default": true
"type": [
"boolean",
"null"
],
"default": null
},
"persistentVolumeClaimRetentionPolicy": {
"$ref": "#/definitions/persistentVolumeClaimRetentionPolicy",
"description": "PersistentVolumeClaim retention policy to be used in the lifecycle of a StatefulSet."
},
"size": {
"description": "Volume size for Airflow Celery worker StatefulSet.",
"type": "string",
"default": "100Gi"
"type": [
"string",
"null"
],
"default": null
},
"storageClassName": {
"description": "If using a custom StorageClass, pass name ref to all StatefulSets here (templated).",
Expand All @@ -2885,8 +2900,11 @@
},
"fixPermissions": {
"description": "Execute init container to chown log directory. This is currently only needed in kind, due to usage of local-path provisioner.",
"type": "boolean",
"default": false
"type": [
"boolean",
"null"
],
"default": null
},
"annotations": {
"description": "Annotations to add to Airflow Celery worker volumes.",
Expand Down Expand Up @@ -2929,8 +2947,11 @@
"properties": {
"enabled": {
"description": "Enable Kerberos init container.",
"type": "boolean",
"default": false
"type": [
"boolean",
"null"
],
"default": null
},
"resources": {
"description": "Resources on kerberos init container.",
Expand Down Expand Up @@ -3066,8 +3087,11 @@
"properties": {
"enabled": {
"description": "Enable kerberos init container.",
"type": "boolean",
"default": false
"type": [
"boolean",
"null"
],
"default": null
},
"resources": {
"description": "Resources on kerberos init container.",
Expand Down
37 changes: 13 additions & 24 deletions chart/values.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1041,7 +1041,7 @@ workers:

celery:
# Number of Airflow Celery workers
replicas: 1
replicas: ~

# Max number of old Airflow Celery workers ReplicaSets to retain
revisionHistoryLimit: ~
Expand All @@ -1050,24 +1050,16 @@ workers:
command: ~

# Args to use when running Airflow Celery workers (templated)
args:
- "bash"
- "-c"
- |-
exec \
airflow {{ semverCompare ">=2.0.0" .Values.airflowVersion | ternary "celery worker" "worker" }}
{{- if and .Values.workers.queue (ne .Values.workers.queue "default") }}
{{- " -q " }}{{ .Values.workers.queue }}
{{- end }}
args: ~

# If the Airflow Celery worker stops responding for 5 minutes (5*60s)
# kill the worker and let Kubernetes restart it
livenessProbe:
enabled: true
initialDelaySeconds: 10
timeoutSeconds: 20
failureThreshold: 5
periodSeconds: 60
enabled: ~
initialDelaySeconds: ~
timeoutSeconds: ~
failureThreshold: ~
periodSeconds: ~
command: ~

# Enable the default workers defined by the root `workers` and `workers.celery`
Expand Down Expand Up @@ -1106,10 +1098,7 @@ workers:
# Update Strategy when Airflow Celery worker is deployed as a StatefulSet
updateStrategy: ~
# Update Strategy when Airflow Celery worker is deployed as a Deployment
strategy:
rollingUpdate:
maxSurge: "100%"
maxUnavailable: "50%"
strategy: ~

# Allow relaxing ordering guarantees for Airflow Celery worker
# while preserving its uniqueness and identity
Expand All @@ -1124,7 +1113,7 @@ workers:
# Persistence volume configuration for Airflow Celery workers
persistence:
# Enable persistent volumes
enabled: true
enabled: ~

# This policy determines whether PVCs should be deleted when StatefulSet is scaled down or removed
persistentVolumeClaimRetentionPolicy: ~
Expand All @@ -1133,15 +1122,15 @@ workers:
# whenScaled: Delete

# Volume size for Airflow Celery worker StatefulSet
size: 100Gi
size: ~

# If using a custom storageClass, pass name ref to all StatefulSets here
storageClassName:

# Execute init container to chown log directory.
# This is currently only needed in kind, due to usage
# of local-path provisioner.
fixPermissions: false
fixPermissions: ~

# Annotations to add to Airflow Celery worker volumes
annotations: {}
Expand All @@ -1155,7 +1144,7 @@ workers:
kerberosInitContainer:
# Enable kerberos init container
# If workers.kerberosInitContainer.enabled is set to True, this flag has no effect
enabled: false
enabled: ~

resources: {}
# limits:
Expand Down Expand Up @@ -1187,7 +1176,7 @@ workers:
kerberosInitContainer:
# Enable kerberos init container
# If workers.kerberosInitContainer.enabled is set to True, this flag has no effect
enabled: false
enabled: ~

resources: {}
# limits:
Expand Down
57 changes: 3 additions & 54 deletions helm-tests/tests/helm_tests/airflow_core/test_scheduler.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,68 +47,17 @@ class TestScheduler:
"StatefulSet",
),
("LocalExecutor", {"celery": {"persistence": {"enabled": False}}}, "Deployment"),
# Test workers.persistence.enabled flag when celery one is default (expected no impact on kind)
# Test workers.persistence.enabled flag when celery one is default
("CeleryExecutor", {"persistence": {"enabled": False}}, "Deployment"),
("CeleryExecutor", {"persistence": {"enabled": True}}, "Deployment"),
("CeleryKubernetesExecutor", {"persistence": {"enabled": True}}, "Deployment"),
("CeleryExecutor,KubernetesExecutor", {"persistence": {"enabled": True}}, "Deployment"),
("KubernetesExecutor", {"persistence": {"enabled": True}}, "Deployment"),
("LocalKubernetesExecutor", {"persistence": {"enabled": False}}, "StatefulSet"),
("LocalKubernetesExecutor", {"persistence": {"enabled": False}}, "Deployment"),
("LocalKubernetesExecutor", {"persistence": {"enabled": True}}, "StatefulSet"),
("LocalExecutor", {"persistence": {"enabled": True}}, "StatefulSet"),
("LocalExecutor,KubernetesExecutor", {"persistence": {"enabled": True}}, "StatefulSet"),
("LocalExecutor", {"persistence": {"enabled": False}}, "StatefulSet"),
# Test workers.persistence.enabled flag when celery one is unset
(
"CeleryExecutor",
{"persistence": {"enabled": False}, "celery": {"persistence": {"enabled": None}}},
"Deployment",
),
(
"CeleryExecutor",
{"persistence": {"enabled": True}, "celery": {"persistence": {"enabled": None}}},
"Deployment",
),
(
"CeleryKubernetesExecutor",
{"persistence": {"enabled": True}, "celery": {"persistence": {"enabled": None}}},
"Deployment",
),
(
"CeleryExecutor,KubernetesExecutor",
{"persistence": {"enabled": True}, "celery": {"persistence": {"enabled": None}}},
"Deployment",
),
(
"KubernetesExecutor",
{"persistence": {"enabled": True}, "celery": {"persistence": {"enabled": None}}},
"Deployment",
),
(
"LocalKubernetesExecutor",
{"persistence": {"enabled": False}, "celery": {"persistence": {"enabled": None}}},
"Deployment",
),
(
"LocalKubernetesExecutor",
{"persistence": {"enabled": True}, "celery": {"persistence": {"enabled": None}}},
"StatefulSet",
),
(
"LocalExecutor",
{"persistence": {"enabled": True}, "celery": {"persistence": {"enabled": None}}},
"StatefulSet",
),
(
"LocalExecutor,KubernetesExecutor",
{"persistence": {"enabled": True}, "celery": {"persistence": {"enabled": None}}},
"StatefulSet",
),
(
"LocalExecutor",
{"persistence": {"enabled": False}, "celery": {"persistence": {"enabled": None}}},
"Deployment",
),
("LocalExecutor", {"persistence": {"enabled": False}}, "Deployment"),
],
)
def test_scheduler_kind(self, executor, workers_values, kind):
Expand Down
Loading
Loading