**Apache Airflow version**: 2.0.1

**Kubernetes version (if you are using kubernetes)** (use `kubectl version`):

**Environment**:

- **Cloud provider or hardware configuration**: Various (GCP & local Python)
- **OS** (e.g. from /etc/os-release): Various (linux, OSX)

**What happened**:

I, a certifiable idiot, accidentally passed a string into a task's `priority_weight` parameter in production.

There was no error at DAG evaluation time. However, upon __running__ the task, the scheduler immediately crashed. Because it was a scheduled run, the scheduler continued to restart and immediately crash until the offending DAG was paused.

The stack trace for my local repro is:
```
Traceback (most recent call last):
  File "/Users/tomyedwab/.venv/khanflow/lib/python3.8/site-packages/flask/app.py", line 2447, in wsgi_app
    response = self.full_dispatch_request()
  File "/Users/tomyedwab/.venv/khanflow/lib/python3.8/site-packages/flask/app.py", line 1952, in full_dispatch_request
    rv = self.handle_user_exception(e)
  File "/Users/tomyedwab/.venv/khanflow/lib/python3.8/site-packages/flask/app.py", line 1821, in handle_user_exception
    reraise(exc_type, exc_value, tb)
  File "/Users/tomyedwab/.venv/khanflow/lib/python3.8/site-packages/flask/_compat.py", line 39, in reraise
    raise value
  File "/Users/tomyedwab/.venv/khanflow/lib/python3.8/site-packages/flask/app.py", line 1950, in full_dispatch_request
    rv = self.dispatch_request()
  File "/Users/tomyedwab/.venv/khanflow/lib/python3.8/site-packages/flask/app.py", line 1936, in dispatch_request
    return self.view_functions[rule.endpoint](**req.view_args)
  File "/Users/tomyedwab/.venv/khanflow/lib/python3.8/site-packages/airflow/www/auth.py", line 34, in decorated
    return func(*args, **kwargs)
  File "/Users/tomyedwab/.venv/khanflow/lib/python3.8/site-packages/airflow/www/decorators.py", line 60, in wrapper
    return f(*args, **kwargs)
  File "/Users/tomyedwab/.venv/khanflow/lib/python3.8/site-packages/airflow/utils/session.py", line 65, in wrapper
    return func(*args, session=session, **kwargs)
  File "/Users/tomyedwab/.venv/khanflow/lib/python3.8/site-packages/airflow/www/views.py", line 1459, in trigger
    dag.create_dagrun(
  File "/Users/tomyedwab/.venv/khanflow/lib/python3.8/site-packages/airflow/utils/session.py", line 65, in wrapper
    return func(*args, session=session, **kwargs)
  File "/Users/tomyedwab/.venv/khanflow/lib/python3.8/site-packages/airflow/models/dag.py", line 1787, in create_dagrun
    run.verify_integrity(session=session)
  File "/Users/tomyedwab/.venv/khanflow/lib/python3.8/site-packages/airflow/utils/session.py", line 62, in wrapper
    return func(*args, **kwargs)
  File "/Users/tomyedwab/.venv/khanflow/lib/python3.8/site-packages/airflow/models/dagrun.py", line 663, in verify_integrity
    ti = TI(task, self.execution_date)
  File "<string>", line 4, in __init__
  File "/Users/tomyedwab/.venv/khanflow/lib/python3.8/site-packages/sqlalchemy/orm/state.py", line 433, in _initialize_instance
    manager.dispatch.init_failure(self, args, kwargs)
  File "/Users/tomyedwab/.venv/khanflow/lib/python3.8/site-packages/sqlalchemy/util/langhelpers.py", line 68, in __exit__
    compat.raise_(
  File "/Users/tomyedwab/.venv/khanflow/lib/python3.8/site-packages/sqlalchemy/util/compat.py", line 182, in raise_
    raise exception
  File "/Users/tomyedwab/.venv/khanflow/lib/python3.8/site-packages/sqlalchemy/orm/state.py", line 430, in _initialize_instance
    return manager.original_init(*mixed[1:], **kwargs)
  File "/Users/tomyedwab/.venv/khanflow/lib/python3.8/site-packages/airflow/models/taskinstance.py", line 286, in __init__
    self.refresh_from_task(task)
  File "/Users/tomyedwab/.venv/khanflow/lib/python3.8/site-packages/airflow/models/taskinstance.py", line 619, in refresh_from_task
    self.priority_weight = task.priority_weight_total
  File "/Users/tomyedwab/.venv/khanflow/lib/python3.8/site-packages/airflow/models/baseoperator.py", line 751, in priority_weight_total
    return self.priority_weight + sum(
TypeError: can only concatenate str (not "int") to str
```

**What you expected to happen**:

I would hope that simple mistakes like this wouldn't be able to take down the Airflow scheduler. Ideally, this type of exception would cause a task to fail and trigger the task failure logic rather than relying on monitoring uptime for the scheduler process.

Separately, it would be nice to have a validation check in BaseOperator that the priority_weights are integers so we get quicker feedback if we supply an invalid value as soon as the DAG is deployed, rather than when it is supposed to run.

**How to reproduce it**:

I can reproduce this easily by adding a bad `priority_weight` parameter to any task, i.e.:

```
PythonOperator(task_id='hello', python_callable=_print_hello, priority_weight="X")
```

How often does this problem occur? Once? Every time etc?

This problem occurs every time.

tests/models/test_baseoperator.py
"""Verify if operator attributes are correctly templated to Native Python objects."""
        with DAG("test-dag", start_date=DEFAULT_DATE, render_template_as_native_obj=True):
            task = MockOperator(task_id="op1", arg1="{{ foo }}", arg2="{{ bar }}")
        
        assert task.arg1 == "{{ foo }}"
        assert task.arg2 == "{{ bar }}"
        
        task.render_template_fields(context={"foo": ["item1", "item2"], "bar": 3})
        assert task.arg1 == ["item1", "item2"]
        assert task.arg2 == 3
    @parameterized.expand(
        [
            ({"user_defined_macros": {"foo": "bar"}}, "{{ foo }}", {}, "bar"),
            ({"user_defined_macros": {"foo": "bar"}}, 1, {}, 1),
            (
                {"user_defined_filters": {"hello": lambda name: f"Hello {name}"}},
                "{{ 'world' | hello }}",
                {},
                "Hello world",
            ),
        ]
    )
    def test_render_template_fields_with_dag_settings(self, dag_kwargs, content, context, expected_output):
        """Test render_template with additional DAG settings."""
        with DAG("test-dag", start_date=DEFAULT_DATE, **dag_kwargs):
            task = DummyOperator(task_id="op1")
        result = task.render_template(content, context)
        assert result == expected_output
    @parameterized.expand([(object(),), (uuid.uuid4(),)])
    def test_render_template_fields_no_change(self, content):