<!--

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
)
    
    sparkr_task = DataprocSubmitJobOperator(
        task_id="sparkr_task", job=SPARKR_JOB, region=REGION, project_id=PROJECT_ID
    )
    hive_task = DataprocSubmitJobOperator(
        task_id="hive_task", job=HIVE_JOB, region=REGION, project_id=PROJECT_ID
    )
    hadoop_task = DataprocSubmitJobOperator(
        task_id="hadoop_task", job=HADOOP_JOB, region=REGION, project_id=PROJECT_ID
    )
    
    delete_cluster = DataprocDeleteClusterOperator(
        task_id="delete_cluster", project_id=PROJECT_ID, cluster_name=CLUSTER_NAME, region=REGION
    )
    
    create_cluster >> scale_cluster
    scale_cluster >> create_workflow_template >> trigger_workflow >> delete_cluster
    scale_cluster >> hive_task >> delete_cluster
    scale_cluster >> pig_task >> delete_cluster
    scale_cluster >> spark_sql_task >> delete_cluster
    scale_cluster >> spark_task >> delete_cluster
    scale_cluster >> spark_task_async
    spark_task_async_sensor >> delete_cluster
    scale_cluster >> pyspark_task >> delete_cluster
    scale_cluster >> sparkr_task >> delete_cluster
    scale_cluster >> hadoop_task >> delete_cluster