暂停、恢复、停止和重启连接器

您可以通过暂停、恢复、停止或重启连接器来控制其运行。通过这些操作，您可以管理数据流并解决问题，而无需删除并重新创建连接器。

如需暂停、恢复、停止或重启 Connect 集群中的连接器，您可以使用 Google Cloud 控制台、gcloud CLI、Managed Service for Apache Kafka 客户端库或 Managed Kafka API。您无法使用开源 Apache Kafka API 来更改连接器状态。

暂停、恢复、停止或重启连接器所需的角色和权限

如需获得暂停、恢复、停止或重启连接器所需的权限，请让管理员为您授予包含 Connect 集群的项目的 Managed Kafka Connector Editor (roles/managedkafka.connectorEditor) IAM 角色。 如需详细了解如何授予角色，请参阅管理对项目、文件夹和组织的访问权限。

此预定义角色包含暂停、恢复、停止或重启连接器所需的权限。如需查看所需的确切权限，请展开所需权限部分：

所需权限

您需要具备以下权限才能暂停、恢复、停止或重启连接器：

• 授予对所请求连接器的暂停连接器权限： managedkafka.connectors.pause
• 授予对所请求连接器的恢复连接器权限： managedkafka.connectors.resume
• 授予对所请求连接器的重启连接器权限： managedkafka.connectors.restart
• 授予对所请求连接器的停止连接器权限： managedkafka.connectors.stop

您也可以使用自定义角色或其他预定义角色来获取这些权限。

如需详细了解 Managed Kafka Connector Editor 角色，请参阅 Managed Service for Apache Kafka 预定义角色。

暂停连接器

暂停连接器时，其状态会保留。这意味着连接器会记住它在处理消息或数据时中断的位置。消息处理会暂停，直到连接器恢复为止。您可以恢复已暂停的连接器，它会从暂停的位置继续运行。这对于在不丢失连接器设置的情况下进行问题排查或执行维护非常有用。

控制台

1. 在 Google Cloud 控制台中，前往连接集群页面。

   前往“关联集群”

2. 点击托管要暂停的连接器的已连接集群。

   系统会显示连接集群详情页面。

3. 在资源标签页中，找到列表中的连接器，然后点击其名称。

   系统会将您重定向到连接器详情页面。

4. 点击暂停。

gcloud

1. In the Google Cloud console, activate Cloud Shell.

   Activate Cloud Shell

   At the bottom of the Google Cloud console, a Cloud Shell session starts and displays a command-line prompt. Cloud Shell is a shell environment with the Google Cloud CLI already installed and with values already set for your current project. It can take a few seconds for the session to initialize.

2. 使用 gcloud managed-kafka connectors pause 命令暂停连接器：

   gcloud managed-kafka connectors pause CONNECTOR_ID \
       --location=LOCATION \
       --connect-cluster=CONNECT_CLUSTER_ID
   

   替换以下内容：

   • CONNECTOR_ID：必填。您要暂停的连接器的 ID。
   • LOCATION：必填。包含连接器的 Connect 集群的位置。
   • CONNECT_CLUSTER_ID：必填。包含连接器的 Connect 集群的 ID。

Go

在尝试此示例之前，请按照 安装客户端库中的 Go 设置说明进行操作。如需了解详情，请参阅 Managed Service for Apache Kafka Go API 参考文档。

如需向 Managed Service for Apache Kafka 进行身份验证，请设置应用默认凭据(ADC)。 如需了解详情，请参阅为本地开发环境设置 ADC。

import (
	"context"
	"fmt"
	"io"

	"cloud.google.com/go/managedkafka/apiv1/managedkafkapb"
	"google.golang.org/api/option"

	managedkafka "cloud.google.com/go/managedkafka/apiv1"
)

func pauseConnector(w io.Writer, projectID, region, connectClusterID, connectorID string, opts ...option.ClientOption) error {
	// projectID := "my-project-id"
	// region := "us-central1"
	// connectClusterID := "my-connect-cluster"
	// connectorID := "my-connector"
	ctx := context.Background()
	client, err := managedkafka.NewManagedKafkaConnectClient(ctx, opts...)
	if err != nil {
		return fmt.Errorf("managedkafka.NewManagedKafkaConnectClient got err: %w", err)
	}
	defer client.Close()

	connectorPath := fmt.Sprintf("projects/%s/locations/%s/connectClusters/%s/connectors/%s", projectID, region, connectClusterID, connectorID)
	req := &managedkafkapb.PauseConnectorRequest{
		Name: connectorPath,
	}
	resp, err := client.PauseConnector(ctx, req)
	if err != nil {
		return fmt.Errorf("client.PauseConnector got err: %w", err)
	}
	fmt.Fprintf(w, "Paused connector: %#v\n", resp)
	return nil
}

Java

在尝试此示例之前，请按照 安装客户端库中的 Java 设置说明进行操作。如需了解详情，请参阅 Managed Service for Apache Kafka Java API 参考文档。

如需向 Managed Service for Apache Kafka 进行身份验证，请设置应用默认凭据。 如需了解详情，请参阅 为本地开发环境设置 ADC。

import com.google.api.gax.rpc.ApiException;
import com.google.cloud.managedkafka.v1.ConnectorName;
import com.google.cloud.managedkafka.v1.ManagedKafkaConnectClient;
import com.google.cloud.managedkafka.v1.PauseConnectorRequest;
import java.io.IOException;

public class PauseConnector {

  public static void main(String[] args) throws Exception {
    // TODO(developer): Replace these variables before running the example.
    String projectId = "my-project-id";
    String region = "my-region"; // e.g. us-east1
    String connectClusterId = "my-connect-cluster";
    String connectorId = "my-connector";
    pauseConnector(projectId, region, connectClusterId, connectorId);
  }

  public static void pauseConnector(
      String projectId, String region, String connectClusterId, String connectorId)
      throws Exception {
    try (ManagedKafkaConnectClient managedKafkaConnectClient = 
        ManagedKafkaConnectClient.create()) {
      ConnectorName connectorName = ConnectorName.of(projectId, region, connectClusterId, 
          connectorId);
      PauseConnectorRequest request = PauseConnectorRequest.newBuilder()
          .setName(connectorName.toString()).build();

      // This operation is being handled synchronously.
      managedKafkaConnectClient.pauseConnector(request);
      System.out.printf("Connector %s paused successfully.\n", connectorId);
    } catch (IOException | ApiException e) {
      System.err.printf("managedKafkaConnectClient.pauseConnector got err: %s\n", 
          e.getMessage());
    }
  }
}

Python

在尝试此示例之前，请按照 安装客户端库中的 Python 设置说明进行操作。如需了解详情，请参阅 Managed Service for Apache Kafka Python API 参考文档。

如需向 Managed Service for Apache Kafka 进行身份验证，请设置应用默认凭据。 如需了解详情，请参阅为本地开发环境设置 ADC。

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.PauseConnectorRequest(
    name=connect_client.connector_path(project_id, region, connect_cluster_id, connector_id),
)

try:
    operation = connect_client.pause_connector(request=request)
    print(f"Waiting for operation {operation.operation.name} to complete...")
    operation.result()
    print(f"Paused connector {connector_id}")
except GoogleAPICallError as e:
    print(f"Failed to pause connector {connector_id} with error: {e}")

恢复连接器

恢复暂停的连接器会从上次中断的位置重新开始运行。

控制台

1. 在 Google Cloud 控制台中，前往连接集群页面。

   前往“关联集群”

2. 点击托管要恢复的连接器的 Connect 集群。

   系统会显示连接集群详情页面。

3. 在资源标签页中，找到列表中的已暂停连接器，然后点击其名称。

   系统会将您重定向到连接器详情页面。

4. 点击继续。

gcloud

1. In the Google Cloud console, activate Cloud Shell.

   Activate Cloud Shell

   At the bottom of the Google Cloud console, a Cloud Shell session starts and displays a command-line prompt. Cloud Shell is a shell environment with the Google Cloud CLI already installed and with values already set for your current project. It can take a few seconds for the session to initialize.

2. 使用 gcloud managed-kafka connectors resume 命令恢复连接器：

   gcloud managed-kafka connectors resume CONNECTOR_ID \
       --location=LOCATION \
       --connect-cluster=CONNECT_CLUSTER_ID
   

   替换以下内容：

   • CONNECTOR_ID：必填。您要恢复的连接器的 ID。
   • LOCATION：必填。包含连接器的 Connect 集群的位置。
   • CONNECT_CLUSTER_ID：必填。包含连接器的 Connect 集群的 ID。

Go

在尝试此示例之前，请按照 安装客户端库中的 Go 设置说明进行操作。如需了解详情，请参阅 Managed Service for Apache Kafka Go API 参考文档。

如需向 Managed Service for Apache Kafka 进行身份验证，请设置应用默认凭据(ADC)。 如需了解详情，请参阅为本地开发环境设置 ADC。

import (
	"context"
	"fmt"
	"io"

	"cloud.google.com/go/managedkafka/apiv1/managedkafkapb"
	"google.golang.org/api/option"

	managedkafka "cloud.google.com/go/managedkafka/apiv1"
)

func resumeConnector(w io.Writer, projectID, region, connectClusterID, connectorID string, opts ...option.ClientOption) error {
	// projectID := "my-project-id"
	// region := "us-central1"
	// connectClusterID := "my-connect-cluster"
	// connectorID := "my-connector"
	ctx := context.Background()
	client, err := managedkafka.NewManagedKafkaConnectClient(ctx, opts...)
	if err != nil {
		return fmt.Errorf("managedkafka.NewManagedKafkaConnectClient got err: %w", err)
	}
	defer client.Close()

	connectorPath := fmt.Sprintf("projects/%s/locations/%s/connectClusters/%s/connectors/%s", projectID, region, connectClusterID, connectorID)
	req := &managedkafkapb.ResumeConnectorRequest{
		Name: connectorPath,
	}
	resp, err := client.ResumeConnector(ctx, req)
	if err != nil {
		return fmt.Errorf("client.ResumeConnector got err: %w", err)
	}
	fmt.Fprintf(w, "Resumed connector: %#v\n", resp)
	return nil
}

Java

在尝试此示例之前，请按照 安装客户端库中的 Java 设置说明进行操作。如需了解详情，请参阅 Managed Service for Apache Kafka Java API 参考文档。

如需向 Managed Service for Apache Kafka 进行身份验证，请设置应用默认凭据。 如需了解详情，请参阅 为本地开发环境设置 ADC。

import com.google.api.gax.rpc.ApiException;
import com.google.cloud.managedkafka.v1.ConnectorName;
import com.google.cloud.managedkafka.v1.ManagedKafkaConnectClient;
import com.google.cloud.managedkafka.v1.ResumeConnectorRequest;
import java.io.IOException;

public class ResumeConnector {

  public static void main(String[] args) throws Exception {
    // TODO(developer): Replace these variables before running the example.
    String projectId = "my-project-id";
    String region = "my-region"; // e.g. us-east1
    String connectClusterId = "my-connect-cluster";
    String connectorId = "my-connector";
    resumeConnector(projectId, region, connectClusterId, connectorId);
  }

  public static void resumeConnector(
      String projectId, String region, String connectClusterId, String connectorId)
      throws Exception {
    try (ManagedKafkaConnectClient managedKafkaConnectClient = 
        ManagedKafkaConnectClient.create()) {
      ConnectorName connectorName = ConnectorName.of(projectId, region, connectClusterId, 
          connectorId);
      ResumeConnectorRequest request = ResumeConnectorRequest.newBuilder()
          .setName(connectorName.toString()).build();

      // This operation is being handled synchronously.
      managedKafkaConnectClient.resumeConnector(request);
      System.out.printf("Connector %s resumed successfully.\n", connectorId);
    } catch (IOException | ApiException e) {
      System.err.printf("managedKafkaConnectClient.resumeConnector got err: %s\n", 
          e.getMessage());
    }
  }
}

Python

在尝试此示例之前，请按照 安装客户端库中的 Python 设置说明进行操作。如需了解详情，请参阅 Managed Service for Apache Kafka Python API 参考文档。

如需向 Managed Service for Apache Kafka 进行身份验证，请设置应用默认凭据。 如需了解详情，请参阅为本地开发环境设置 ADC。

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.ResumeConnectorRequest(
    name=connect_client.connector_path(project_id, region, connect_cluster_id, connector_id),
)

try:
    operation = connect_client.resume_connector(request=request)
    print(f"Waiting for operation {operation.operation.name} to complete...")
    operation.result()
    print(f"Resumed connector {connector_id}")
except GoogleAPICallError as e:
    print(f"Failed to resume connector {connector_id} with error: {e}")

停止连接器

停止连接器会停止连接器的所有任务。停止连接器会保留其状态。如需让连接器再次运行，请重启连接器。日志和指标也会持久存储。

控制台

1. 在 Google Cloud 控制台中，前往连接集群页面。

   前往“关联集群”

2. 点击托管要停止的连接器的已连接集群。

   系统会显示连接集群详情页面。

3. 在资源标签页中，找到列表中的连接器，然后点击其名称。

   系统会将您重定向到连接器详情页面。

4. 点击停止。

gcloud

1. In the Google Cloud console, activate Cloud Shell.

   Activate Cloud Shell

   At the bottom of the Google Cloud console, a Cloud Shell session starts and displays a command-line prompt. Cloud Shell is a shell environment with the Google Cloud CLI already installed and with values already set for your current project. It can take a few seconds for the session to initialize.

2. 使用 gcloud managed-kafka connectors stop 命令停止连接器：

   gcloud managed-kafka connectors stop CONNECTOR_ID \
       --location=LOCATION \
       --connect-cluster=CONNECT_CLUSTER_ID
   

   替换以下内容：

   • CONNECTOR_ID：必填。要停止的连接器的 ID。
   • LOCATION：必填。包含连接器的 Connect 集群的位置。
   • CONNECT_CLUSTER_ID：必填。包含连接器的 Connect 集群的 ID。

Go

在尝试此示例之前，请按照 安装客户端库中的 Go 设置说明进行操作。如需了解详情，请参阅 Managed Service for Apache Kafka Go API 参考文档。

如需向 Managed Service for Apache Kafka 进行身份验证，请设置应用默认凭据(ADC)。 如需了解详情，请参阅为本地开发环境设置 ADC。

import (
	"context"
	"fmt"
	"io"

	"cloud.google.com/go/managedkafka/apiv1/managedkafkapb"
	"google.golang.org/api/option"

	managedkafka "cloud.google.com/go/managedkafka/apiv1"
)

func stopConnector(w io.Writer, projectID, region, connectClusterID, connectorID string, opts ...option.ClientOption) error {
	// projectID := "my-project-id"
	// region := "us-central1"
	// connectClusterID := "my-connect-cluster"
	// connectorID := "my-connector"
	ctx := context.Background()
	client, err := managedkafka.NewManagedKafkaConnectClient(ctx, opts...)
	if err != nil {
		return fmt.Errorf("managedkafka.NewManagedKafkaConnectClient got err: %w", err)
	}
	defer client.Close()

	connectorPath := fmt.Sprintf("projects/%s/locations/%s/connectClusters/%s/connectors/%s", projectID, region, connectClusterID, connectorID)
	req := &managedkafkapb.StopConnectorRequest{
		Name: connectorPath,
	}
	resp, err := client.StopConnector(ctx, req)
	if err != nil {
		return fmt.Errorf("client.StopConnector got err: %w", err)
	}
	fmt.Fprintf(w, "Stopped connector: %#v\n", resp)
	return nil
}

Java

在尝试此示例之前，请按照 安装客户端库中的 Java 设置说明进行操作。如需了解详情，请参阅 Managed Service for Apache Kafka Java API 参考文档。

如需向 Managed Service for Apache Kafka 进行身份验证，请设置应用默认凭据。 如需了解详情，请参阅 为本地开发环境设置 ADC。

import com.google.api.gax.rpc.ApiException;
import com.google.cloud.managedkafka.v1.ConnectorName;
import com.google.cloud.managedkafka.v1.ManagedKafkaConnectClient;
import com.google.cloud.managedkafka.v1.StopConnectorRequest;
import java.io.IOException;

public class StopConnector {

  public static void main(String[] args) throws Exception {
    // TODO(developer): Replace these variables before running the example.
    String projectId = "my-project-id";
    String region = "my-region"; // e.g. us-east1
    String connectClusterId = "my-connect-cluster";
    String connectorId = "my-connector";
    stopConnector(projectId, region, connectClusterId, connectorId);
  }

  public static void stopConnector(
      String projectId, String region, String connectClusterId, String connectorId)
      throws Exception {
    try (ManagedKafkaConnectClient managedKafkaConnectClient = 
        ManagedKafkaConnectClient.create()) {
      ConnectorName connectorName = ConnectorName.of(projectId, region, connectClusterId, 
          connectorId);
      StopConnectorRequest request = StopConnectorRequest.newBuilder()
          .setName(connectorName.toString()).build();

      // This operation is being handled synchronously.
      managedKafkaConnectClient.stopConnector(request);
      System.out.printf("Connector %s stopped successfully.\n", connectorId);
    } catch (IOException | ApiException e) {
      System.err.printf("managedKafkaConnectClient.stopConnector got err: %s\n", e.getMessage());
    }
  }
}

Python

在尝试此示例之前，请按照 安装客户端库中的 Python 设置说明进行操作。如需了解详情，请参阅 Managed Service for Apache Kafka Python API 参考文档。

如需向 Managed Service for Apache Kafka 进行身份验证，请设置应用默认凭据。 如需了解详情，请参阅为本地开发环境设置 ADC。

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.StopConnectorRequest(
    name=connect_client.connector_path(project_id, region, connect_cluster_id, connector_id),
)

try:
    operation = connect_client.stop_connector(request=request)
    print(f"Waiting for operation {operation.operation.name} to complete...")
    operation.result()
    print(f"Stopped connector {connector_id}")
except GoogleAPICallError as e:
    print(f"Failed to stop connector {connector_id} with error: {e}")

重启连接器

重启连接器会完全停止并重新启动其任务。这对于刷新连接器的状态或应用配置更改非常有用。

注意：重新启动连接器可能会导致数据流短暂中断。

控制台

1. 在 Google Cloud 控制台中，前往连接集群页面。

   前往“关联集群”

2. 点击要重启的连接器所在的已连接集群。

   系统会显示连接集群详情页面。

3. 在资源标签页中，找到列表中的连接器，然后点击其名称。

   系统会将您重定向到连接器详情页面。

4. 点击重新启动。

gcloud

1. In the Google Cloud console, activate Cloud Shell.

   Activate Cloud Shell

   At the bottom of the Google Cloud console, a Cloud Shell session starts and displays a command-line prompt. Cloud Shell is a shell environment with the Google Cloud CLI already installed and with values already set for your current project. It can take a few seconds for the session to initialize.

2. 使用 gcloud managed-kafka connectors restart 命令重启连接器：

   gcloud managed-kafka connectors restart CONNECTOR_ID \
       --location=LOCATION \
       --connect-cluster=CONNECT_CLUSTER_ID
   

   替换以下内容：

   • CONNECTOR_ID：必填。要重启的连接器的 ID。
   • LOCATION：必填。包含连接器的 Connect 集群的位置。
   • CONNECT_CLUSTER_ID：必填。包含连接器的 Connect 集群的 ID。

Go

在尝试此示例之前，请按照 安装客户端库中的 Go 设置说明进行操作。如需了解详情，请参阅 Managed Service for Apache Kafka Go API 参考文档。

如需向 Managed Service for Apache Kafka 进行身份验证，请设置应用默认凭据(ADC)。 如需了解详情，请参阅为本地开发环境设置 ADC。

import (
	"context"
	"fmt"
	"io"

	"cloud.google.com/go/managedkafka/apiv1/managedkafkapb"
	"google.golang.org/api/option"

	managedkafka "cloud.google.com/go/managedkafka/apiv1"
)

func restartConnector(w io.Writer, projectID, region, connectClusterID, connectorID string, opts ...option.ClientOption) error {
	// projectID := "my-project-id"
	// region := "us-central1"
	// connectClusterID := "my-connect-cluster"
	// connectorID := "my-connector"
	ctx := context.Background()
	client, err := managedkafka.NewManagedKafkaConnectClient(ctx, opts...)
	if err != nil {
		return fmt.Errorf("managedkafka.NewManagedKafkaConnectClient got err: %w", err)
	}
	defer client.Close()

	connectorPath := fmt.Sprintf("projects/%s/locations/%s/connectClusters/%s/connectors/%s", projectID, region, connectClusterID, connectorID)
	req := &managedkafkapb.RestartConnectorRequest{
		Name: connectorPath,
	}
	resp, err := client.RestartConnector(ctx, req)
	if err != nil {
		return fmt.Errorf("client.RestartConnector got err: %w", err)
	}
	fmt.Fprintf(w, "Restarted connector: %#v\n", resp)
	return nil
}

Java

在尝试此示例之前，请按照 安装客户端库中的 Java 设置说明进行操作。如需了解详情，请参阅 Managed Service for Apache Kafka Java API 参考文档。

如需向 Managed Service for Apache Kafka 进行身份验证，请设置应用默认凭据。 如需了解详情，请参阅 为本地开发环境设置 ADC。

import com.google.api.gax.rpc.ApiException;
import com.google.cloud.managedkafka.v1.ConnectorName;
import com.google.cloud.managedkafka.v1.ManagedKafkaConnectClient;
import com.google.cloud.managedkafka.v1.RestartConnectorRequest;
import java.io.IOException;

public class RestartConnector {

  public static void main(String[] args) throws Exception {
    // TODO(developer): Replace these variables before running the example.
    String projectId = "my-project-id";
    String region = "my-region"; // e.g. us-east1
    String connectClusterId = "my-connect-cluster";
    String connectorId = "my-connector";
    restartConnector(projectId, region, connectClusterId, connectorId);
  }

  public static void restartConnector(
      String projectId, String region, String connectClusterId, String connectorId)
      throws Exception {
    try (ManagedKafkaConnectClient managedKafkaConnectClient = 
        ManagedKafkaConnectClient.create()) {
      ConnectorName connectorName = ConnectorName.of(projectId, region, connectClusterId, 
          connectorId);
      RestartConnectorRequest request = RestartConnectorRequest.newBuilder()
          .setName(connectorName.toString()).build();

      // This operation is being handled synchronously.
      managedKafkaConnectClient.restartConnector(request);
      System.out.printf("Connector %s restarted successfully.\n", connectorId);
    } catch (IOException | ApiException e) {
      System.err.printf("managedKafkaConnectClient.restartConnector got err: %s\n", 
          e.getMessage());
    }
  }
}

Python

在尝试此示例之前，请按照 安装客户端库中的 Python 设置说明进行操作。如需了解详情，请参阅 Managed Service for Apache Kafka Python API 参考文档。

如需向 Managed Service for Apache Kafka 进行身份验证，请设置应用默认凭据。 如需了解详情，请参阅为本地开发环境设置 ADC。

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.RestartConnectorRequest(
    name=connect_client.connector_path(project_id, region, connect_cluster_id, connector_id),
)

try:
    operation = connect_client.restart_connector(request=request)
    print(f"Waiting for operation {operation.operation.name} to complete...")
    operation.result()
    print(f"Restarted connector {connector_id}")
except GoogleAPICallError as e:
    print(f"Failed to restart connector {connector_id} with error: {e}")

Apache Kafka® 是 Apache Software Foundation 或其关联公司在美国和/或其他国家/地区的注册商标。