MKC Examples - Connector Operations (#13527)

* Add Managed Kafka Connect code samples for
clusters

* Adds code examples for creating, deleting,
getting, listing and updating Managed Kafka
Connect clusters

* Update google-cloud-managedkafka version to 0.1.12

* Update managedkafka/snippets/connect/clusters/create_connect_cluster.py

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>

* Update managedkafka/snippets/connect/clusters/create_connect_cluster.py

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>

* Update managedkafka/snippets/connect/clusters/delete_connect_cluster.py

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>

* Update managedkafka/snippets/connect/clusters/get_connect_cluster.py

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>

* Update managedkafka/snippets/connect/clusters/list_connect_clusters.py

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>

* Update managedkafka/snippets/connect/clusters/update_connect_cluster.py

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>

* Add timeouts and improve error handling.

* Addressed PR comments.

* Adds requirements.txt file for samples.

As per the [Authoring Guide]
(https://github.com/GoogleCloudPlatform/python-docs-samples/blob/main/AUTHORING_GUIDE.md#dependencies), each sample is
required to have a requirements.txt file that
lists the dependencies needed to run the sample.

* Adds code examples for Managed Kafka Connect
Connectors:
* Creating BigQuery Sink Connector
* Creating MirrorMaker Source Connector
* Creating Pub/Sub Sink Connector
* Creating Pub/Sub Source Connector
* Creating Storage Sink Connector
* Getting, updating, deleting, and listing connectors
* Starting, stopping, pausing, restarting and resuming connectors

* Minor fixes.

* Remove redundant file.

* Fix merge issues.

* Add example for `bootstrap.servers`

* Remove obsolete MM2 connector example.

* Clarify use for restart_connector.

* Fix comment on bootstrap.servers for MM2 example.

* Remove obsolete file.

* Fix variable naming and tasks.max value.

* Fix connectors_test.py linting errors.

* Fix linting errors for connector samples.

* linting (black)

* Remove requirements.txt to fix linting conflicts.

* sort order

---------

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
Co-authored-by: Katie McLaughlin <[email protected]>

Author: 3 people

managedkafka/snippets/connect/clusters/delete_connect_cluster.py

@@ -41,14 +41,14 @@ def delete_connect_cluster(
     # region = "us-central1"
     # connect_cluster_id = "my-connect-cluster"
 
-    client = ManagedKafkaConnectClient()
+    connect_client = ManagedKafkaConnectClient()
 
     request = managedkafka_v1.DeleteConnectClusterRequest(
-        name=client.connect_cluster_path(project_id, region, connect_cluster_id),
+        name=connect_client.connect_cluster_path(project_id, region, connect_cluster_id),
     )
 
     try:
-        operation = client.delete_connect_cluster(request=request)
+        operation = connect_client.delete_connect_cluster(request=request)
         print(f"Waiting for operation {operation.operation.name} to complete...")
         operation.result()
         print("Deleted Connect cluster")

managedkafka/snippets/connect/connectors/connectors_test.py

@@ -20,14 +20,23 @@
 import create_mirrormaker2_source_connector
 import create_pubsub_sink_connector
 import create_pubsub_source_connector
+import delete_connector
+import get_connector
 from google.api_core.operation import Operation
 from google.cloud import managedkafka_v1
+import list_connectors
+import pause_connector
 import pytest
+import restart_connector
+import resume_connector
+import stop_connector
+import update_connector
 
 
 PROJECT_ID = "test-project-id"
 REGION = "us-central1"
 CONNECT_CLUSTER_ID = "test-connect-cluster-id"
+CONNECTOR_ID = "test-connector-id"
 
 
 @mock.patch(
@@ -194,3 +203,203 @@ def test_create_bigquery_sink_connector(
     assert "Created Connector" in out
     assert connector_id in out
     mock_method.assert_called_once()
+
+
+@mock.patch(
+    "google.cloud.managedkafka_v1.services.managed_kafka_connect.ManagedKafkaConnectClient.list_connectors"
+)
+def test_list_connectors(
+    mock_method: MagicMock,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    connector = managedkafka_v1.types.Connector()
+    connector.name = managedkafka_v1.ManagedKafkaConnectClient.connector_path(
+        PROJECT_ID, REGION, CONNECT_CLUSTER_ID, CONNECTOR_ID
+    )
+    mock_method.return_value = [connector]
+
+    list_connectors.list_connectors(
+        project_id=PROJECT_ID,
+        region=REGION,
+        connect_cluster_id=CONNECT_CLUSTER_ID,
+    )
+
+    out, _ = capsys.readouterr()
+    assert "Got connector" in out
+    assert CONNECTOR_ID in out
+    mock_method.assert_called_once()
+
+
+@mock.patch(
+    "google.cloud.managedkafka_v1.services.managed_kafka_connect.ManagedKafkaConnectClient.get_connector"
+)
+def test_get_connector(
+    mock_method: MagicMock,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    connector = managedkafka_v1.types.Connector()
+    connector.name = managedkafka_v1.ManagedKafkaConnectClient.connector_path(
+        PROJECT_ID, REGION, CONNECT_CLUSTER_ID, CONNECTOR_ID
+    )
+    mock_method.return_value = connector
+
+    get_connector.get_connector(
+        project_id=PROJECT_ID,
+        region=REGION,
+        connect_cluster_id=CONNECT_CLUSTER_ID,
+        connector_id=CONNECTOR_ID,
+    )
+
+    out, _ = capsys.readouterr()
+    assert "Got connector" in out
+    assert CONNECTOR_ID in out
+    mock_method.assert_called_once()
+
+
+@mock.patch(
+    "google.cloud.managedkafka_v1.services.managed_kafka_connect.ManagedKafkaConnectClient.update_connector"
+)
+def test_update_connector(
+    mock_method: MagicMock,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    configs = {"tasks.max": "6", "value.converter.schemas.enable": "true"}
+    operation = mock.MagicMock(spec=Operation)
+    connector = managedkafka_v1.types.Connector()
+    connector.name = managedkafka_v1.ManagedKafkaConnectClient.connector_path(
+        PROJECT_ID, REGION, CONNECT_CLUSTER_ID, CONNECTOR_ID
+    )
+    operation.result = mock.MagicMock(return_value=connector)
+    mock_method.return_value = operation
+
+    update_connector.update_connector(
+        project_id=PROJECT_ID,
+        region=REGION,
+        connect_cluster_id=CONNECT_CLUSTER_ID,
+        connector_id=CONNECTOR_ID,
+        configs=configs,
+    )
+
+    out, _ = capsys.readouterr()
+    assert "Updated connector" in out
+    assert CONNECTOR_ID in out
+    mock_method.assert_called_once()
+
+
+@mock.patch(
+    "google.cloud.managedkafka_v1.services.managed_kafka_connect.ManagedKafkaConnectClient.delete_connector"
+)
+def test_delete_connector(
+    mock_method: MagicMock,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    operation = mock.MagicMock(spec=Operation)
+    operation.result = mock.MagicMock(return_value=None)
+    mock_method.return_value = operation
+
+    delete_connector.delete_connector(
+        project_id=PROJECT_ID,
+        region=REGION,
+        connect_cluster_id=CONNECT_CLUSTER_ID,
+        connector_id=CONNECTOR_ID,
+    )
+
+    out, _ = capsys.readouterr()
+    assert "Deleted connector" in out
+    mock_method.assert_called_once()
+
+
+@mock.patch(
+    "google.cloud.managedkafka_v1.services.managed_kafka_connect.ManagedKafkaConnectClient.pause_connector"
+)
+def test_pause_connector(
+    mock_method: MagicMock,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    operation = mock.MagicMock(spec=Operation)
+    operation.result = mock.MagicMock(return_value=None)
+    mock_method.return_value = operation
+
+    pause_connector.pause_connector(
+        project_id=PROJECT_ID,
+        region=REGION,
+        connect_cluster_id=CONNECT_CLUSTER_ID,
+        connector_id=CONNECTOR_ID,
+    )
+
+    out, _ = capsys.readouterr()
+    assert "Paused connector" in out
+    assert CONNECTOR_ID in out
+    mock_method.assert_called_once()
+
+
+@mock.patch(
+    "google.cloud.managedkafka_v1.services.managed_kafka_connect.ManagedKafkaConnectClient.resume_connector"
+)
+def test_resume_connector(
+    mock_method: MagicMock,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    operation = mock.MagicMock(spec=Operation)
+    operation.result = mock.MagicMock(return_value=None)
+    mock_method.return_value = operation
+
+    resume_connector.resume_connector(
+        project_id=PROJECT_ID,
+        region=REGION,
+        connect_cluster_id=CONNECT_CLUSTER_ID,
+        connector_id=CONNECTOR_ID,
+    )
+
+    out, _ = capsys.readouterr()
+    assert "Resumed connector" in out
+    assert CONNECTOR_ID in out
+    mock_method.assert_called_once()
+
+
+@mock.patch(
+    "google.cloud.managedkafka_v1.services.managed_kafka_connect.ManagedKafkaConnectClient.stop_connector"
+)
+def test_stop_connector(
+    mock_method: MagicMock,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    operation = mock.MagicMock(spec=Operation)
+    operation.result = mock.MagicMock(return_value=None)
+    mock_method.return_value = operation
+
+    stop_connector.stop_connector(
+        project_id=PROJECT_ID,
+        region=REGION,
+        connect_cluster_id=CONNECT_CLUSTER_ID,
+        connector_id=CONNECTOR_ID,
+    )
+
+    out, _ = capsys.readouterr()
+    assert "Stopped connector" in out
+    assert CONNECTOR_ID in out
+    mock_method.assert_called_once()
+
+
+@mock.patch(
+    "google.cloud.managedkafka_v1.services.managed_kafka_connect.ManagedKafkaConnectClient.restart_connector"
+)
+def test_restart_connector(
+    mock_method: MagicMock,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    operation = mock.MagicMock(spec=Operation)
+    operation.result = mock.MagicMock(return_value=None)
+    mock_method.return_value = operation
+
+    restart_connector.restart_connector(
+        project_id=PROJECT_ID,
+        region=REGION,
+        connect_cluster_id=CONNECT_CLUSTER_ID,
+        connector_id=CONNECTOR_ID,
+    )
+
+    out, _ = capsys.readouterr()
+    assert "Restarted connector" in out
+    assert CONNECTOR_ID in out
+    mock_method.assert_called_once()

managedkafka/snippets/connect/connectors/create_bigquery_sink_connector.py

@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+
 def create_bigquery_sink_connector(
     project_id: str,
     region: str,

managedkafka/snippets/connect/connectors/create_mirrormaker2_source_connector.py

@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+
 def create_mirrormaker2_source_connector(
     project_id: str,
     region: str,
@@ -75,8 +76,9 @@ def create_mirrormaker2_source_connector(
         "target.cluster.alias": target_cluster_alias,  # This is usually the primary cluster.
         # Replicate all topics from the source
         "topics": topics,
-        # The value for bootstrap.servers is a comma-separated list of hostname:port pairs
-        # for one or more Kafka brokers in the source/target cluster.
+        # The value for bootstrap.servers is a hostname:port pair for the Kafka broker in
+        # the source/target cluster.
+        # For example: "kafka-broker:9092"
         "source.cluster.bootstrap.servers": source_bootstrap_servers,
         "target.cluster.bootstrap.servers": target_bootstrap_servers,
         # You can define an exclusion policy for topics as follows:

managedkafka/snippets/connect/connectors/create_pubsub_sink_connector.py

@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+
 def create_pubsub_sink_connector(
     project_id: str,
     region: str,

managedkafka/snippets/connect/connectors/create_pubsub_source_connector.py

@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+
 def create_pubsub_source_connector(
     project_id: str,
     region: str,

managedkafka/snippets/connect/connectors/delete_connector.py

@@ -0,0 +1,61 @@
+# Copyright 2025 Google LLC
+#
+# Licensed 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.
+
+
+def delete_connector(
+    project_id: str,
+    region: str,
+    connect_cluster_id: str,
+    connector_id: str,
+) -> None:
+    """
+    Delete a connector.
+
+    Args:
+        project_id: Google Cloud project ID.
+        region: Cloud region.
+        connect_cluster_id: ID of the Kafka Connect cluster.
+        connector_id: ID of the connector.
+
+    Raises:
+        This method will raise the GoogleAPICallError exception if the operation errors.
+    """
+    # [START managedkafka_delete_connector]
+    from google.api_core.exceptions import GoogleAPICallError
+    from google.cloud.managedkafka_v1.services.managed_kafka_connect import (
+        ManagedKafkaConnectClient,
+    )
+    from google.cloud import managedkafka_v1
+
+    # TODO(developer)
+    # project_id = "my-project-id"
+    # region = "us-central1"
+    # connect_cluster_id = "my-connect-cluster"
+    # connector_id = "my-connector"
+
+    connect_client = ManagedKafkaConnectClient()
+
+    request = managedkafka_v1.DeleteConnectorRequest(
+        name=connect_client.connector_path(project_id, region, connect_cluster_id, connector_id),
+    )
+
+    try:
+        operation = connect_client.delete_connector(request=request)
+        print(f"Waiting for operation {operation.operation.name} to complete...")
+        operation.result()
+        print("Deleted connector")
+    except GoogleAPICallError as e:
+        print(f"The operation failed with error: {e}")
+
+    # [END managedkafka_delete_connector]