### Description

Allow an Airflow DAG to define a Databricks job with the `api/2.1/jobs/create` (or `api/2.1/jobs/reset`) endpoint then run that same job with the `api/2.1/jobs/run-now` endpoint. This would give similar capabilities as the DatabricksSubmitRun operator, but the `api/2.1/jobs/create` endpoint supports additional parameters that the `api/2.1/jobs/runs/submit` doesn't (e.g. `job_clusters`, `email_notifications`, etc.).

### Use case/motivation

Create and run a Databricks job all in the Airflow DAG. Currently, DatabricksSubmitRun operator uses the `api/2.1/jobs/runs/submit` endpoint which doesn't support all features and creates runs that aren't tied to a job in the Databricks UI. Also, DatabricksRunNow operator requires you to define the job either directly in the Databricks UI or through a separate CI/CD pipeline causing the headache of having to change code in multiple places.

### Related issues

_No response_

### Are you willing to submit a PR?

- [X] Yes I am willing to submit a PR!

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)

return str(self.__dict__)
    def to_json(self) -> str:
        return json.dumps(self.__dict__)
    @classmethod
    def from_json(cls, data: str) -> RunState:
        return RunState(**json.loads(data))
class ClusterState:
    """Utility class for the cluster state concept of Databricks cluster."""
    CLUSTER_LIFE_CYCLE_STATES = [
        "PENDING",
        "RUNNING",
        "RESTARTING",
        "RESIZING",
        "TERMINATING",
        "TERMINATED",
        "ERROR",
        "UNKNOWN",
    ]
    def __init__(self, state: str = "", state_message: str = "", *args, **kwargs) -> None:
        if state not in self.CLUSTER_LIFE_CYCLE_STATES:
            raise AirflowException(
                f"Unexpected cluster life cycle state: {state}: If the state has "
                "been introduced recently, please check the Databricks user "
                "guide for troubleshooting information"
            )
        self.state = state
        self.state_message = state_message
    @property
    def is_terminal(self) -> bool: