<!--

Welcome to Apache Airflow!  For a smooth issue process, try to answer the following questions.
Don't worry if they're not all applicable; just try to include what you can :-)

If you need to include code snippets or logs, please put them in fenced code
blocks.  If they're super-long, please use the details tag like
<details><summary>super-long log</summary> lots of stuff </details>

Please delete these comment blocks before submitting the issue.

-->

Hi,
I am facing some issues while installing PIP Packages in the Dataproc cluster using Initialization script,
I am trying to upgrade to Airflow 2.0 from 1.10.12  (where this code works fine)

``
[2021-07-09 11:35:37,587] {taskinstance.py:1454} ERROR - metadata was invalid: [('PIP_PACKAGES', 'pyyaml requests pandas openpyxl'), ('x-goog-api-client', 'gl-python/3.7.10 grpc/1.35.0 gax/1.26.0 gccl/airflow_v2.0.0+astro.3')
``
```python
 path = f"gs://goog-dataproc-initialization-actions-{self.cfg.get('region')}/python/pip-install.sh"
 
return DataprocClusterCreateOperator(
     ........
  init_actions_uris=[path],
  metadata=[('PIP_PACKAGES', 'pyyaml requests pandas openpyxl')],
    ............
      )
```

**Apache Airflow version**:
airflow_v2.0.0


**What happened**:
I am trying to migrate our codebase from Airflow v1.10.12, on the deeper analysis found that as part refactoring in of below pr #6371, we can no longer pass **metadata** in DataprocClusterCreateOperator() as this is not being  passed to ClusterGenerator() method.

**What you expected to happen**:
Operator should work as before.


airflow/providers/google/cloud/example_dags/example_dataproc.py
"placement": {"cluster_name": CLUSTER_NAME},
    "spark_r_job": {"main_r_file_uri": SPARKR_URI},
}
HIVE_JOB = {
    "reference": {"project_id": PROJECT_ID},
    "placement": {"cluster_name": CLUSTER_NAME},
    "hive_job": {"query_list": {"queries": ["SHOW DATABASES;"]}},
}
HADOOP_JOB = {
    "reference": {"project_id": PROJECT_ID},
    "placement": {"cluster_name": CLUSTER_NAME},
    "hadoop_job": {
        "main_jar_file_uri": "file:///usr/lib/hadoop-mapreduce/hadoop-mapreduce-examples.jar",
        "args": ["wordcount", "gs://pub/shakespeare/rose.txt", OUTPUT_PATH],
    },
}
WORKFLOW_NAME = "airflow-dataproc-test"
WORKFLOW_TEMPLATE = {
    "id": WORKFLOW_NAME,
    "placement": {
        "managed_cluster": {
            "cluster_name": CLUSTER_NAME,
            "config": CLUSTER_CONFIG,
        }
    },
    "jobs": [{"step_id": "pig_job_1", "pig_job": PIG_JOB["pig_job"]}],
}
with models.DAG(
    "example_gcp_dataproc",
    schedule_interval='@once',