title: Component Gateway activation on dataproc does not work with composer(airflow) operator airflow.providers.google.cloud.operators.dataproc
body: <p>I’m trying execute this dag bellow.
  It seems that the operator creating a dataproc cluster does not enable enabling the optional components to enable jupyter notebook and anaconda.
  I found this code here: <a href="https://stackoverflow.com/questions/59568043/component-gateway-with-dataprocoperator-on-airflow/71211981#71211981">Component Gateway with DataprocOperator on Airflow</a> to try to solve it, but for me it didn't solve it because i thikn the composer(airflow) version here is diferente. my version is composer - 2.0.0-preview.5, airflow-2.1.4.</p>
  <p>The operator works perfectly when creating the cluster, but it doesn't create with the optional component to enable jupyter notebook.
  Does anyone have any ideas to help me?</p>
  <pre><code>from airflow.contrib.sensors.gcs_sensor import GoogleCloudStoragePrefixSensor
  from airflow import DAG
  from datetime import datetime, timedelta
  from airflow.contrib.operators.dataproc_operator import DataprocClusterCreateOperator,DataprocClusterDeleteOperator, DataProcSparkOperator
  from airflow.contrib.operators.gcs_to_bq import GoogleCloudStorageToBigQueryOperator
  
  yesterday = datetime.combine(datetime.today() - timedelta(1),
                               datetime.min.time())
  
  
  default_args = {
      'owner': 'teste3',
      'depends_on_past': False,
      'start_date' :yesterday,
      'email': ['airflow@example.com'],
      'email_on_failure': False,
      'email_on_retry': False,
      'retries': 0,
      'retry_delay': timedelta(minutes=5),
  
  }
  
  dag = DAG(
      'teste-dag-3',catchup=False, default_args=default_args, schedule_interval=None)
  
  
  # configura os componentes
  class CustomDataprocClusterCreateOperator(DataprocClusterCreateOperator):
  
      def __init__(self, *args, **kwargs):
          super(CustomDataprocClusterCreateOperator, self).__init__(*args, **kwargs)
  
      def _build_cluster_data(self):
          cluster_data = super(CustomDataprocClusterCreateOperator, self)._build_cluster_data()
          cluster_data['config']['endpointConfig'] = {
              'enableHttpPortAccess': True
          }
          cluster_data['config']['softwareConfig']['optionalComponents'] = [ 'JUPYTER', 'ANACONDA' ]
          return cluster_data
  
  
  create_cluster=CustomDataprocClusterCreateOperator(
          dag=dag,
          task_id='start_cluster_example',
          cluster_name='teste-ge-{{ ds }}',
          project_id= &quot;sandbox-coe&quot;,
          num_workers=2,
          num_masters=1,
          master_machine_type='n2-standard-8',
          worker_machine_type='n2-standard-8',
          worker_disk_size=500,
          master_disk_size=500,
          master_disk_type='pd-ssd',
          worker_disk_type='pd-ssd',
          image_version='1.5.56-ubuntu18',
          tags=['allow-dataproc-internal'],
          region=&quot;us-central1&quot;,
          zone='us-central1-f',#Variable.get('gc_zone'),
          storage_bucket = &quot;bucket-dataproc-ge&quot;,
          labels = {'product' : 'sample-label'},
          service_account_scopes = ['https://www.googleapis.com/auth/cloud-platform'],
          #properties={&quot;yarn:yarn.nodemanager.resource.memory-mb&quot; : 15360,&quot;yarn:yarn.scheduler.maximum-allocation-mb&quot; : 15360},
          #subnetwork_uri=&quot;projects/project-id/regions/us-central1/subnetworks/dataproc-subnet&quot;,
          retries= 1,
          retry_delay=timedelta(minutes=1)
      ) #starts a dataproc cluster
  
  
  stop_cluster_example = DataprocClusterDeleteOperator(
      dag=dag,
      task_id='stop_cluster_example',
      cluster_name='teste-ge-{{ ds }}',
      project_id=&quot;sandbox-coe&quot;,
      region=&quot;us-central1&quot;,
      ) #stops a running dataproc cluster
  
  
  
  
  create_cluster  &gt;&gt; stop_cluster_example
  </code></pre>