List all connectors

Listing all connectors running in a Connect cluster provides an overview of the data integrations that are configured. You can monitor the health and status of your connectors, identify potential issues, and manage your data flows effectively.

To list all connectors in a Connect cluster, you can use the Google Cloud console, the gcloud CLI, the Managed Service for Apache Kafka client library, or the Managed Kafka API. You can't use the open source Apache Kafka API to list the connectors.

Required roles and permissions to list all connectors

To get the permissions that you need to list all connectors, ask your administrator to grant you the Managed Kafka Viewer (roles/managedkafka.viewer) IAM role on your project. For more information about granting roles, see Manage access to projects, folders, and organizations.

This predefined role contains the permissions required to list all connectors. To see the exact permissions that are required, expand the Required permissions section:

Required permissions

The following permissions are required to list all connectors:

Grant the list connectors permission on the parent Connect cluster: managedkafka.connectors.list
Grant the get connector details permission on the parent Connect cluster: managedkafka.connectors.get

You might also be able to get these permissions with custom roles or other predefined roles.

For more information about the Managed Kafka Viewer role, see Managed Service for Apache Kafka predefined roles.

View all connectors

This view provides a quick way to monitor the status of your connectors and identify any that require attention. You can then drill down into individual connectors to view their details and configurations if needed.

Console

In the Google Cloud console, go to the Connect clusters page.

Go to Connect Clusters
Click the Connect cluster for which you want to list the connectors.

The Connect cluster details page is displayed.
The Resources tab displays a list of all connectors running in the cluster. The list includes the following information for each connector:
- Name: The name of the connector.
- State: The operational state of the connector. For example, Running, Failed.
- Connector type: The type of connector plugin.
You can use the Filter option to search for specific connectors by name.

gcloud

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.

Use the gcloud alpha managed-kafka connectors list command to list connectors:

gcloud alpha managed-kafka connectors list \
    --location=LOCATION \
    --connect_cluster=CONNECT_CLUSTER_ID

To further refine the list of connectors, you can use additional flags:
```
gcloud alpha managed-kafka connectors list \
    --location=LOCATION \
    --connect_cluster=CONNECT_CLUSTER_ID \
    [--filter=EXPRESSION] \
    [--limit=LIMIT] \
    [--page-size=PAGE_SIZE] \
    [--sort-by=SORT_BY]
```
Replace the following:
- LOCATION: Required. The location of the Connect cluster containing the connectors that you want to list.
- CONNECT_CLUSTER_ID: Required. The ID of the Connect cluster containing the connectors you want to list.
- EXPRESSION: (Optional) A Boolean filter expression to apply to the list. If the expression evaluates to True, the item is included in the list. For more details and examples, run gcloud topic filters.
  
  Examples:
  - To list only connectors that are in the 'RUNNING' state:
```
--filter="state=RUNNING"
```
  - To list only 'Pub/Sub Sink' connectors:
```
--filter="connector_plugin='Pub/Sub Sink'"
```
  - To list connectors with a name containing 'prod':
```
--filter="name ~ 'prod'"
```
  - To list connectors that are either 'FAILED' or are 'Pub/Sub Source' plugins:
```
--filter="state=FAILED OR connector_plugin='Pub/Sub Source'"
```
- LIMIT: (Optional) The maximum number of connectors to display. If not specified, all connectors are listed.
- PAGE_SIZE: (Optional) The number of results to display per page. If not specified, the service determines a suitable page size.
- SORT_BY: (Optional) A comma-separated list of fields to sort by. The default sort order is ascending. To sort in descending order, prefix the field with ~. Supported fields are likely name and state.

Example command with sorting:

gcloud alpha managed-kafka connectors list \
    --location=us-central1 \
    --connect_cluster=test-connect-cluster \
    --sort-by=~state,name

Example command with filtering and limiting:

gcloud alpha managed-kafka connectors list \
    --location=us-central1 \
    --connect_cluster=test-connect-cluster \
    --filter="state=RUNNING AND connector_plugin='Pub/Sub Sink'" \
    --limit=5

Example output:

NAME                                    STATE     CONNECTOR_PLUGIN
pubsub-sink-connector                   RUNNING   Pub/Sub Sink
another-pubsub-sink                     RUNNING   Pub/Sub Sink
prod-pubsub-sink                        RUNNING   Pub/Sub Sink

Go

Before trying this sample, follow the Go setup instructions in Install the client libraries. For more information, see the Managed Service for Apache Kafka Go API reference documentation.

To authenticate to Managed Service for Apache Kafka, set up Application Default Credentials(ADC). For more information, see Set up ADC for a local development environment.

import (
	"context"
	"fmt"
	"io"

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

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

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

	parent := fmt.Sprintf("projects/%s/locations/%s/connectClusters/%s", projectID, region, connectClusterID)
	req := &managedkafkapb.ListConnectorsRequest{
		Parent: parent,
	}
	connectorIter := client.ListConnectors(ctx, req)
	for {
		connector, err := connectorIter.Next()
		if err == iterator.Done {
			break
		}
		if err != nil {
			return fmt.Errorf("connectorIter.Next() got err: %w", err)
		}
		fmt.Fprintf(w, "Got connector: %v", connector)
	}
	return nil
}

Java

Before trying this sample, follow the Java setup instructions in Install the client libraries. For more information, see the Managed Service for Apache Kafka Java API reference documentation.

To authenticate to Managed Service for Apache Kafka, set up Application Default Credentials. For more information, see Set up ADC for a local development environment.

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

public class ListConnectors {

  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 clusterId = "my-connect-cluster";
    listConnectors(projectId, region, clusterId);
  }

  public static void listConnectors(String projectId, String region, String clusterId)
      throws IOException {
    try (ManagedKafkaConnectClient managedKafkaConnectClient = ManagedKafkaConnectClient.create()) {
      ConnectClusterName parent = ConnectClusterName.of(projectId, region, clusterId);
      // This operation is handled synchronously.
      for (Connector connector : managedKafkaConnectClient.listConnectors(parent).iterateAll()) {
        System.out.println(connector.getAllFields());
      }
    } catch (IOException | ApiException e) {
      System.err.printf("managedKafkaConnectClient.listConnectors got err: %s\n", e.getMessage());
    }
  }
}

Python

Before trying this sample, follow the Python setup instructions in Install the client libraries. For more information, see the Managed Service for Apache Kafka Python API reference documentation.