<!--

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
trigger_workflow = DataprocInstantiateWorkflowTemplateOperator(
        task_id="trigger_workflow", region=REGION, project_id=PROJECT_ID, template_id=WORKFLOW_NAME
    )
    
    pig_task = DataprocSubmitJobOperator(
        task_id="pig_task", job=PIG_JOB, region=REGION, project_id=PROJECT_ID
    )
    spark_sql_task = DataprocSubmitJobOperator(
        task_id="spark_sql_task", job=SPARK_SQL_JOB, region=REGION, project_id=PROJECT_ID
    )
    spark_task = DataprocSubmitJobOperator(
        task_id="spark_task", job=SPARK_JOB, region=REGION, project_id=PROJECT_ID
    )
    
    spark_task_async = DataprocSubmitJobOperator(
        task_id="spark_task_async", job=SPARK_JOB, region=REGION, project_id=PROJECT_ID, asynchronous=True
    )
    spark_task_async_sensor = DataprocJobSensor(
        task_id='spark_task_async_sensor_task',
        region=REGION,
        project_id=PROJECT_ID,
        dataproc_job_id=spark_task_async.output,
        poke_interval=10,
    )
    
    
    pyspark_task = DataprocSubmitJobOperator(
        task_id="pyspark_task", job=PYSPARK_JOB, region=REGION, project_id=PROJECT_ID