Issue : When trying to submit an ingestion task using DruidOperator, getting 500 error code in response from Druid. And can see no task submitted in Druid console.
In Airflow 1.10.x, everything is working fine. But when upgraded to 2.0.1, it is failing to submit the task. There is absolutely no change in the code/files except the import statements.

Resolution : I compared DruidOperator code for both Airflow 1.10.x & 2.0.1 and found one line causing the issue.
In Airflow 2.0.x, before submitting the indexing job json string is converted to python object. But it should be json string only.
In Airflow 1.10.x there is no conversion happening and hence it is working fine. (Please see below code snippets.)

I have already tried this change in my setup and re-ran the ingestion tasks. It is all working fine.

~~hook.submit_indexing_job(json.loads(self.json_index_file))~~
**hook.submit_indexing_job(self.json_index_file)**


Airflow 1.10.x - airflow/contrib/operators/druid_operator.py
```
def execute(self, context):
    hook = DruidHook(
        druid_ingest_conn_id=self.conn_id,
        max_ingestion_time=self.max_ingestion_time
    )
    self.log.info("Submitting %s", self.index_spec_str)
    hook.submit_indexing_job(self.index_spec_str)
```

Airflow 2.0.1 - airflow/providers/apache/druid/operators/druid.py
```
def execute(self, context: Dict[Any, Any]) -> None:
    hook = DruidHook(druid_ingest_conn_id=self.conn_id, max_ingestion_time=self.max_ingestion_time)
    self.log.info("Submitting %s", self.json_index_file)
    hook.submit_indexing_job(json.loads(self.json_index_file))
```


**Apache Airflow version**: 2.0.x

**Error Logs**:

```
[2021-02-24 06:42:24,287] {{connectionpool.py:452}} DEBUG - http://druid-master:8081 "POST /druid/indexer/v1/task HTTP/1.1" 500 15714
[2021-02-24 06:42:24,287] {{taskinstance.py:570}} DEBUG - Refreshing TaskInstance <TaskInstance: druid_compact_daily 2021-02-23T01:20:00+00:00 [running]> from DB
[2021-02-24 06:42:24,296] {{taskinstance.py:605}} DEBUG - Refreshed TaskInstance <TaskInstance: druid_compact_daily 2021-02-23T01:20:00+00:00 [running]>
[2021-02-24 06:42:24,298] {{taskinstance.py:1455}} ERROR - Did not get 200 when submitting the Druid job to http://druid-master:8081/druid/indexer/v1/task
```

tests/providers/apache/druid/operators/test_druid.py
#
#
#
#
import unittest
from airflow.models import TaskInstance
from airflow.models.dag import DAG
from airflow.providers.apache.druid.operators.druid import DruidOperator
from airflow.utils import timezone
DEFAULT_DATE = timezone.datetime(2017, 1, 1)
class TestDruidOperator(unittest.TestCase):
    def setUp(self):
        args = {'owner': 'airflow', 'start_date': timezone.datetime(2017, 1, 1)}
        self.dag = DAG('test_dag_id', default_args=args)
    def test_render_template(self):
        json_str = '''
            {
                "type": "{{ params.index_type }}",
                "datasource": "{{ params.datasource }}",
                "spec": {
                    "dataSchema": {