Create operators for working with Clusters for GCP Apache Kafka

generated/provider_dependencies.json

@@ -661,6 +661,7 @@
       "google-cloud-kms>=2.15.0",
       "google-cloud-language>=2.9.0",
       "google-cloud-logging>=3.5.0",
+      "google-cloud-managedkafka>=0.1.6",
       "google-cloud-memcache>=1.7.0",
       "google-cloud-monitoring>=2.18.0",
       "google-cloud-orchestration-airflow>=1.10.0",

providers/google/README.rst

@@ -95,6 +95,7 @@ PIP package                                 Version required
 ``google-cloud-kms``                        ``>=2.15.0``
 ``google-cloud-language``                   ``>=2.9.0``
 ``google-cloud-logging``                    ``>=3.5.0``
+``google-cloud-managedkafka``               ``>=0.1.6``
 ``google-cloud-memcache``                   ``>=1.7.0``
 ``google-cloud-monitoring``                 ``>=2.18.0``
 ``google-cloud-orchestration-airflow``      ``>=1.10.0``

providers/google/docs/operators/cloud/managed_kafka.rst

@@ -0,0 +1,78 @@
+ .. Licensed to the Apache Software Foundation (ASF) under one
+    or more contributor license agreements.  See the NOTICE file
+    distributed with this work for additional information
+    regarding copyright ownership.  The ASF licenses this file
+    to you under the Apache License, Version 2.0 (the
+    "License"); you may not use this file except in compliance
+    with the License.  You may obtain a copy of the License at
+
+ ..   http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+    software distributed under the License is distributed on an
+    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+    KIND, either express or implied.  See the License for the
+    specific language governing permissions and limitations
+    under the License.
+
+Google Cloud Managed Service for Apache Kafka Operators
+=======================================================
+
+The `Google Cloud Managed Service for Apache Kafka <https://cloud.google.com/managed-service-for-apache-kafka/docs>`__
+helps you set up, secure, maintain, and scale Apache Kafka clusters.
+
+Interacting with Apache Kafka Cluster
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+To create an Apache Kafka cluster you can use
+:class:`~airflow.providers.google.cloud.operators.managed_kafka.ManagedKafkaCreateClusterOperator`.
+
+.. exampleinclude:: /../../providers/google/tests/system/google/cloud/managed_kafka/example_managed_kafka_cluster.py
+    :language: python
+    :dedent: 4
+    :start-after: [START how_to_cloud_managed_kafka_create_cluster_operator]
+    :end-before: [END how_to_cloud_managed_kafka_create_cluster_operator]
+
+To delete cluster you can use
+:class:`~airflow.providers.google.cloud.operators.managed_kafka.ManagedKafkaDeleteClusterOperator`.
+
+.. exampleinclude:: /../../providers/google/tests/system/google/cloud/managed_kafka/example_managed_kafka_cluster.py
+    :language: python
+    :dedent: 4
+    :start-after: [START how_to_cloud_managed_kafka_delete_cluster_operator]
+    :end-before: [END how_to_cloud_managed_kafka_delete_cluster_operator]
+
+To get cluster you can use
+:class:`~airflow.providers.google.cloud.operators.managed_kafka.ManagedKafkaGetClusterOperator`.
+
+.. exampleinclude:: /../../providers/google/tests/system/google/cloud/managed_kafka/example_managed_kafka_cluster.py
+    :language: python
+    :dedent: 4
+    :start-after: [START how_to_cloud_managed_kafka_get_cluster_operator]
+    :end-before: [END how_to_cloud_managed_kafka_get_cluster_operator]
+
+To get a list of clusters you can use
+:class:`~airflow.providers.google.cloud.operators.managed_kafka.ManagedKafkaListClustersOperator`.
+
+.. exampleinclude:: /../../providers/google/tests/system/google/cloud/managed_kafka/example_managed_kafka_cluster.py
+    :language: python
+    :dedent: 4
+    :start-after: [START how_to_cloud_managed_kafka_list_cluster_operator]
+    :end-before: [END how_to_cloud_managed_kafka_list_cluster_operator]
+
+To update cluster you can use
+:class:`~airflow.providers.google.cloud.operators.managed_kafka.ManagedKafkaUpdateClusterOperator`.
+
+.. exampleinclude:: /../../providers/google/tests/system/google/cloud/managed_kafka/example_managed_kafka_cluster.py
+    :language: python
+    :dedent: 4
+    :start-after: [START how_to_cloud_managed_kafka_update_cluster_operator]
+    :end-before: [END how_to_cloud_managed_kafka_update_cluster_operator]
+
+Reference
+^^^^^^^^^
+
+For further information, look at:
+
+* `Client Library Documentation <https://cloud.google.com/managed-service-for-apache-kafka/docs/reference/libraries>`__
+* `Product Documentation <https://cloud.google.com/managed-service-for-apache-kafka/docs>`__

providers/google/provider.yaml

@@ -199,6 +199,11 @@ integrations:
       - /docs/apache-airflow-providers-google/operators/cloud/life_sciences.rst
     logo: /docs/integration-logos/Google-Cloud-Life-Sciences.png
     tags: [gcp]
+  - integration-name: Google Cloud Managed Service for Apache Kafka
+    external-doc-url: https://cloud.google.com/managed-service-for-apache-kafka/docs/
+    how-to-guide:
+      - /docs/apache-airflow-providers-google/operators/cloud/managed_kafka.rst
+    tags: [gcp]
   - integration-name: Google Cloud Memorystore
     external-doc-url: https://cloud.google.com/memorystore/
     how-to-guide:
@@ -599,6 +604,9 @@ operators:
   - integration-name: Google Cloud Batch
     python-modules:
       - airflow.providers.google.cloud.operators.cloud_batch
+  - integration-name: Google Cloud Managed Service for Apache Kafka
+    python-modules:
+      - airflow.providers.google.cloud.operators.managed_kafka
 
 sensors:
   - integration-name: Google BigQuery
@@ -877,6 +885,9 @@ hooks:
   - integration-name: Google Cloud Batch
     python-modules:
       - airflow.providers.google.cloud.hooks.cloud_batch
+  - integration-name: Google Cloud Managed Service for Apache Kafka
+    python-modules:
+      - airflow.providers.google.cloud.hooks.managed_kafka
 
 triggers:
   - integration-name: Google BigQuery Data Transfer Service
@@ -1215,6 +1226,8 @@ extra-links:
   - airflow.providers.google.cloud.links.translate.TranslationModelsListLink
   - airflow.providers.google.cloud.links.translate.TranslateResultByOutputConfigLink
   - airflow.providers.google.cloud.links.translate.TranslationGlossariesListLink
+  - airflow.providers.google.cloud.links.managed_kafka.ApacheKafkaClusterLink
+  - airflow.providers.google.cloud.links.managed_kafka.ApacheKafkaClusterListLink
 
 
 secrets-backends:

providers/google/pyproject.toml

@@ -93,6 +93,7 @@ dependencies = [
     "google-cloud-kms>=2.15.0",
     "google-cloud-language>=2.9.0",
     "google-cloud-logging>=3.5.0",
+    "google-cloud-managedkafka>=0.1.6",
     "google-cloud-memcache>=1.7.0",
     "google-cloud-monitoring>=2.18.0",
     "google-cloud-orchestration-airflow>=1.10.0",