[DON'T MERGE] Java SDK All

.dockerignore

@@ -39,6 +39,7 @@
 !task-sdk/
 !airflow-ctl/
 !go-sdk/
+!sdk/
 
 # Add all "test" distributions
 !tests

.github/boring-cyborg.yml

@@ -189,6 +189,9 @@ labelPRBasedOnFilePath:
   provider:keycloak:
     - providers/keycloak/**
 
+  provider:sdk-java:
+    - providers/sdk/java/**
+
   provider:microsoft-azure:
     - providers/microsoft/azure/**
 

.pre-commit-config.yaml

@@ -492,6 +492,12 @@ repos:
         language: python
         pass_filenames: false
         files: ^dev/registry/registry_tools/types\.py$|^registry/src/_data/types\.json$
+      - id: check-task-instance-dto-sync
+        name: Check BaseTaskInstanceDTO duplicate is in sync between core and task-sdk
+        entry: ./scripts/ci/prek/check_task_instance_dto_sync.py
+        language: python
+        pass_filenames: false
+        files: ^airflow-core/src/airflow/executors/workloads/task\.py$|^task-sdk/src/airflow/sdk/execution_time/workloads/task\.py$
       - id: ruff
         name: Run 'ruff' for extremely fast Python linting
         description: "Run 'ruff' for extremely fast Python linting"
@@ -1111,6 +1117,22 @@ repos:
         files: ^airflow-core/src/airflow/migrations/versions/.*\.py$
         additional_dependencies: [rich>=13.6.0]
         require_serial: true
+      - id: check-java-serialization-compatibility
+        name: Check Java SDK serialization compatibility
+        description: >-
+          Verify that the Java and Python SDKs produce identical DAG serialization
+          for all test cases defined in java-sdk/validation/serialization/test_dags.yaml
+        entry: ./scripts/ci/prek/check_java_serialization_compatibility.py
+        language: python
+        stages: ['manual']
+        pass_filenames: false
+        require_serial: true
+        files: >
+          (?x)
+          ^java-sdk/sdk/src/.*\.(kt|java)$|
+          ^java-sdk/validation/serialization/.*$|
+          ^airflow-core/src/airflow/serialization/.*\.py$|
+          ^task-sdk/src/airflow/sdk/definitions/.*\.py$
       ## ADD MOST PREK HOOK ABOVE THAT LINE
       # The below prek hooks are those requiring CI image to be built
       ## ONLY ADD PREK HOOKS HERE THAT REQUIRE CI IMAGE

Dockerfile

@@ -1048,8 +1048,8 @@ function install_airflow_and_providers_from_docker_context_files(){
         install_airflow_core_distribution=("apache-airflow-core==${AIRFLOW_VERSION}")
     fi
 
-    # Find Provider/TaskSDK/CTL distributions in docker-context files
-    readarray -t airflow_distributions< <(python /scripts/docker/get_distribution_specs.py /docker-context-files/apache?airflow?{providers,task?sdk,airflowctl}*.{whl,tar.gz} 2>/dev/null || true)
+    # Find Provider/TaskSDK/CTL/Coordinator distributions in docker-context files
+    readarray -t airflow_distributions< <(python /scripts/docker/get_distribution_specs.py /docker-context-files/apache?airflow?{providers,task?sdk,airflowctl,coordinators?java}*.{whl,tar.gz} 2>/dev/null || true)
     echo
     echo "${COLOR_BLUE}Found provider distributions in docker-context-files folder: ${airflow_distributions[*]}${COLOR_RESET}"
     echo

Dockerfile.ci

@@ -1222,6 +1222,30 @@ function environment_initialization() {
         export AIRFLOW__SCHEDULER__GO_WORKER=True
     fi
 
+    if [[ ${JAVA_SDK=} == "true" ]]; then
+        echo
+        echo "${COLOR_BLUE}Setting up Java SDK${COLOR_RESET}"
+        echo
+
+        # Install Java and the Java SDK provider
+        bash /opt/airflow/scripts/in_container/java_sdk_setup.sh
+
+        # Set JAVA_HOME and PATH before Gradle build so gradlew can find java
+        export JAVA_HOME=/files/openjdk
+        export PATH=/files/openjdk/bin:${PATH}
+
+        # Build both Java SDK bundles (stub + pure Java)
+        export JAVA_SDK_SRC_DIR=/opt/airflow/java-sdk
+        export BUNDLES_OUTPUT_DIR=/files/java-sdk-bundles
+        bash /opt/airflow/scripts/in_container/java_sdk_build.sh
+
+        # Source the generated environment configuration
+        # shellcheck disable=SC1091
+        source /files/java-sdk-bundles/java_sdk_env.sh
+
+        echo "${COLOR_BLUE}Java SDK setup complete.${COLOR_RESET}"
+    fi
+
     RUN_TESTS=${RUN_TESTS:="false"}
     CI=${CI:="false"}
 

airflow-core/docs/extra-packages-ref.rst

@@ -178,6 +178,17 @@ all the ``airflow`` packages together - similarly to what happened in Airflow 2.
 ``airflow-task-sdk`` separately, if you want to install providers, you need to install them separately as
 ``apache-airflow-providers-*`` distribution packages.
 
+Multi-Language extras
+=====================
+
+These are extras that add dependencies needed for integration with other languages runtimes. Currently we have only Java SDK related extra, but in the future we might add more extras related to other languages runtimes.
+
++----------+------------------------------------------+------------------------------------------------------------------+
+| extra    | install command                          | enables                                                          |
++==========+==========================================+==================================================================+
+| sdk.java | ``pip install apache-airflow[sdk.java]`` | JavaCoordinator for both dag processing and workload execution.  |
++----------+------------------------------------------+------------------------------------------------------------------+
+
 Apache Software extras
 ======================
 

airflow-core/pyproject.toml

@@ -249,6 +249,7 @@ exclude = [
 "../shared/serialization/src/airflow_shared/serialization" = "src/airflow/_shared/serialization"
 "../shared/state/src/airflow_shared/state" = "src/airflow/_shared/state"
 "../shared/timezones/src/airflow_shared/timezones" = "src/airflow/_shared/timezones"
+"../shared/workloads/src/airflow_shared/workloads" = "src/airflow/_shared/workloads"
 "../shared/listeners/src/airflow_shared/listeners" = "src/airflow/_shared/listeners"
 "../shared/plugins_manager/src/airflow_shared/plugins_manager" = "src/airflow/_shared/plugins_manager"
 "../shared/providers_discovery/src/airflow_shared/providers_discovery" = "src/airflow/_shared/providers_discovery"
@@ -337,6 +338,7 @@ shared_distributions = [
     "apache-airflow-shared-serialization",
     "apache-airflow-shared-state",
     "apache-airflow-shared-timezones",
+    "apache-airflow-shared-workloads",
     "apache-airflow-shared-plugins-manager",
     "apache-airflow-shared-providers-discovery",
 ]

airflow-core/src/airflow/_shared/workloads

@@ -0,0 +1 @@
+../../../../shared/workloads/src/airflow_shared/workloads

airflow-core/src/airflow/config_templates/config.yml

@@ -1967,6 +1967,43 @@ workers:
       type: integer
       example: ~
       default: "60"
+sdk:
+  description: Settings for non-Python SDK runtime coordination
+  options:
+    coordinators:
+      description: |
+        JSON list of runtime coordinator entries.
+
+        Each entry is an object with ``name``, ``classpath`` and optional
+        ``kwargs``.  ``classpath`` is resolved via ``import_string`` and
+        constructed with ``kwargs`` once per process.  Entries are
+        independent instances, so the same ``classpath`` can be configured
+        multiple times with different ``kwargs`` (for example, two
+        ``JavaCoordinator`` instances pinned to different JDK versions).
+      version_added: 3.1.7
+      type: string
+      example: |
+        [
+          {
+            "name": "jdk-17",
+            "classpath": "airflow.sdk.coordinators.java.JavaCoordinator",
+            "kwargs": {"java_executable": "/usr/lib/jvm/java-17-openjdk/bin/java", "jvm_args": ["-Xmx1024m"]}
+          }
+        ]
+      default: ~
+    queue_to_coordinator:
+      description: |
+        JSON mapping of queue names to coordinator ``name`` from
+        ``[sdk] coordinators``.
+
+        When a task's ``language`` field is not set, this mapping is checked
+        to route the task to a configured coordinator instance based on its
+        queue.  This is useful when queues are used as environment or
+        isolation identifiers (e.g. ``legacy-java``, ``modern-java``).
+      version_added: 3.1.7
+      type: string
+      example: '{"legacy-java": "jdk-11", "modern-java": "jdk-17"}'
+      default: ~
 api_auth:
   description: Settings relating to authentication on the Airflow APIs
   options:

airflow-core/src/airflow/dag_processing/manager.py

@@ -267,6 +267,9 @@ class DagFileProcessorManager(LoggingMixin):
         factory=_config_get_factory("dag_processor", "file_parsing_sort_mode")
     )
 
+    _runtime_file_extensions: tuple[str, ...] | None = attrs.field(default=None, init=False)
+    """File extensions registered by runtime coordinators (e.g. ".jar"). Lazily populated."""
+
     _api_server: InProcessExecutionAPI = attrs.field(init=False, factory=InProcessExecutionAPI)
     """API server to interact with Metadata DB"""
 
@@ -815,14 +818,28 @@ def _find_files_in_bundle(self, bundle: BaseDagBundle) -> list[Path]:
 
         return rel_paths
 
+    def _get_runtime_file_extensions(self) -> tuple[str, ...]:
+        """Collect file extensions from configured runtime coordinators (cached after first call)."""
+        if self._runtime_file_extensions is not None:
+            return self._runtime_file_extensions
+
+        from airflow.sdk.execution_time.coordinator import get_coordinator_manager
+
+        self._runtime_file_extensions = get_coordinator_manager().file_extensions()
+        return self._runtime_file_extensions
+
     def _get_observed_filelocs(self, present: set[DagFileInfo]) -> set[str]:
         """
         Return observed DAG source paths for bundle entries.
 
         For regular files this includes the relative file path.
         For ZIP archives this includes DAG-like inner paths such as
         ``archive.zip/dag.py``.
+
+        Runtime coordinator file extensions (e.g. ``.jar``) are treated as
+        opaque files rather than ZIP archives.
         """
+        runtime_extensions = self._get_runtime_file_extensions()
 
         def find_zipped_dags(abs_path: os.PathLike) -> Iterator[str]:
             """Yield absolute paths for DAG-like files inside a ZIP archive."""
@@ -837,7 +854,7 @@ def find_zipped_dags(abs_path: os.PathLike) -> Iterator[str]:
         observed_filelocs: set[str] = set()
         for info in present:
             abs_path = str(info.absolute_path)
-            if abs_path.endswith(".py") or not zipfile.is_zipfile(abs_path):
+            if abs_path.endswith((".py", *runtime_extensions)) or not zipfile.is_zipfile(abs_path):
                 observed_filelocs.add(str(info.rel_path))
             else:
                 if TYPE_CHECKING:

airflow-core/src/airflow/dag_processing/processor.py

@@ -17,6 +17,7 @@
 from __future__ import annotations
 
 import contextlib
+import functools
 import importlib
 import logging
 import os
@@ -76,8 +77,6 @@
 from airflow.utils.state import TaskInstanceState
 
 if TYPE_CHECKING:
-    from socket import socket
-
     from structlog.typing import FilteringBoundLogger
 
     from airflow.api_fastapi.execution_api.app import InProcessExecutionAPI
@@ -86,6 +85,7 @@
     from airflow.sdk.definitions.context import Context
     from airflow.sdk.definitions.dag import DAG
     from airflow.sdk.definitions.mappedoperator import MappedOperator
+    from airflow.sdk.execution_time.supervisor import SelectorCallback
     from airflow.typing_compat import Self
 
 
@@ -553,7 +553,14 @@ def start(  # type: ignore[override]
     ) -> Self:
         logger = kwargs["logger"]
 
-        _pre_import_airflow_modules(os.fspath(path), logger)
+        # Check if a configured runtime coordinator should handle this file
+        logger.debug("Checking for runtime coordinator entrypoint for file", path=path)
+        resolved_target = cls._resolve_processor_target(path, bundle_name, bundle_path, logger)
+        if resolved_target is not None:
+            target = resolved_target
+            logger.debug("Resolved runtime coordinator entrypoint for file", path=path)
+        else:
+            _pre_import_airflow_modules(os.fspath(path), logger)
 
         proc: Self = super().start(
             target=target,
@@ -566,6 +573,35 @@ def start(  # type: ignore[override]
         proc._on_child_started(callbacks, path, bundle_path, bundle_name)
         return proc
 
+    @staticmethod
+    def _resolve_processor_target(
+        path: str | os.PathLike[str],
+        bundle_name: str,
+        bundle_path: Path,
+        log: FilteringBoundLogger,
+    ) -> Callable[[], None] | None:
+        """
+        Return the entrypoint of the first runtime coordinator that can handle *path*.
+
+        The returned callable is a ``functools.partial`` that binds *path*, *bundle_name*
+        and *bundle_path* so the supervisor can pass it as a no-arg ``target`` to
+        ``WatchedSubprocess.start``.
+        """
+        from airflow.sdk.execution_time.coordinator import get_coordinator_manager
+
+        coordinator = get_coordinator_manager().for_dag_file(bundle_name, path)
+        if coordinator is None:
+            log.debug("No runtime coordinator found for file %s, using default processor", path)
+            return None
+
+        log.debug("Using runtime coordinator %s for file %s", type(coordinator).__qualname__, path)
+        return functools.partial(
+            coordinator.run_dag_parsing,
+            path=os.fspath(path),
+            bundle_name=bundle_name,
+            bundle_path=os.fspath(bundle_path),
+        )
+
     def _on_child_started(
         self,
         callbacks: list[CallbackRequest],
@@ -591,7 +627,7 @@ def _get_target_loggers(self) -> tuple[FilteringBoundLogger, ...]:
 
     def _create_log_forwarder(
         self, loggers: tuple[FilteringBoundLogger, ...], name: str, log_level: int = logging.INFO
-    ) -> Callable[[socket], bool]:
+    ) -> SelectorCallback:
         return super()._create_log_forwarder(loggers, name.replace("task.", "dag_processor.", 1), log_level)
 
     def _handle_request(self, msg: ToManager, log: FilteringBoundLogger, req_id: int) -> None:

airflow-core/src/airflow/executors/base_executor.py

@@ -650,11 +650,10 @@ def run_workload(
 
         if isinstance(workload, ExecuteTask):
             from airflow.sdk.execution_time.supervisor import supervise_task
+            from airflow.sdk.execution_time.workloads.task import TaskInstanceDTO as SDKTaskInstanceDTO
 
-            # workload.ti is a TaskInstanceDTO which duck-types as TaskInstance.
-            # TODO: Create a protocol for this.
             return supervise_task(
-                ti=workload.ti,  # type: ignore[arg-type]
+                ti=SDKTaskInstanceDTO.model_validate(workload.ti, from_attributes=True),
                 bundle_info=workload.bundle_info,
                 dag_rel_path=workload.dag_rel_path,
                 token=workload.token,

airflow-core/src/airflow/executors/workloads/task.py

@@ -33,8 +33,14 @@
     from airflow.models.taskinstancekey import TaskInstanceKey
 
 
-class TaskInstanceDTO(BaseModel):
-    """Schema for TaskInstance with minimal required fields needed for Executors and Task SDK."""
+class BaseTaskInstanceDTO(BaseModel):
+    """
+    Base schema for TaskInstance with the minimal fields shared by Executors and the Task SDK.
+
+    This definition is duplicated in :mod:`airflow.sdk.execution_time.workloads.task`
+    and the two are kept in sync by the ``check-task-instance-dto-sync`` prek
+    hook. Update both files together.
+    """
 
     id: uuid.UUID
     dag_version_id: uuid.UUID
@@ -48,11 +54,16 @@ class TaskInstanceDTO(BaseModel):
     queue: str
     priority_weight: int
     executor_config: dict | None = Field(default=None, exclude=True)
-    external_executor_id: str | None = Field(default=None, exclude=True)
 
     parent_context_carrier: dict | None = None
     context_carrier: dict | None = None
 
+
+class TaskInstanceDTO(BaseTaskInstanceDTO):
+    """TaskInstanceDTO with executor-specific ``external_executor_id`` field and ``key`` property."""
+
+    external_executor_id: str | None = Field(default=None, exclude=True)
+
     # TODO: Task-SDK: Can we replace TaskInstanceKey with just the uuid across the codebase?
     @property
     def key(self) -> TaskInstanceKey:

airflow-core/src/airflow/models/dagcode.py

@@ -119,6 +119,14 @@ def code(cls, dag_id, session: Session = NEW_SESSION) -> str:
 
     @staticmethod
     def get_code_from_file(fileloc):
+        # Try from runtime coordinator first.
+        from airflow.sdk.execution_time.coordinator import get_coordinator_manager
+
+        coordinator = get_coordinator_manager().for_dag_file("", fileloc)
+        if coordinator is not None:
+            return coordinator.get_code_from_file(fileloc)
+
+        # Then fallback to python native
         try:
             with open_maybe_zipped(fileloc, "r") as f:
                 code = f.read()

airflow-core/src/airflow/provider.yaml.schema.json

@@ -624,6 +624,13 @@
                 }
             }
         },
+        "coordinators": {
+            "type": "array",
+            "description": "Runtime Coordinator class names (BaseCoordinator subclasses)",
+            "items": {
+                "type": "string"
+            }
+        },
         "source-date-epoch": {
             "type": "integer",
             "description": "Source date epoch - seconds since epoch (gmtime) when the release documentation was prepared. Used to generate reproducible package builds with flint.",