O Google usa tecnologia de IA na tradução de conteúdos para seu idioma de preferência. As traduções com IA podem ter erros.

Criar um cluster do serviço gerenciado do Google Cloud para Apache Kafka

Um cluster do Serviço Gerenciado para Apache Kafka oferece um ambiente para armazenar e processar fluxos de mensagens organizados em tópicos.

Para criar um cluster, use o console Google Cloud , a Google Cloud CLI, a biblioteca de cliente ou a API Managed Kafka. Não é possível usar a API Apache Kafka de código aberto para criar um cluster.

Antes de começar

Verifique se você tem familiaridade com o seguinte:

Papéis e permissões necessárias para criar um cluster

Para receber as permissões necessárias para criar um cluster, peça ao administrador para conceder a você o papel do IAM de Editor de cluster gerenciado do Kafka (roles/managedkafka.clusterEditor) no projeto. Para mais informações sobre a concessão de papéis, consulte Gerenciar o acesso a projetos, pastas e organizações.

Esse papel predefinido contém as permissões necessárias para criar um cluster. Para acessar as permissões exatas necessárias, expanda a seção Permissões necessárias:

Permissões necessárias

As seguintes permissões são necessárias para criar um cluster:

Crie um cluster: managedkafka.clusters.create

Essas permissões também podem ser concedidas com funções personalizadas ou outros papéis predefinidos.

Propriedades de um cluster do serviço gerenciado para Apache Kafka

Ao criar ou atualizar um cluster do Serviço gerenciado para Apache Kafka, é necessário especificar as seguintes propriedades.

Nome do cluster

O nome ou ID do cluster do serviço gerenciado para Apache Kafka que você está criando. Para diretrizes sobre como nomear um cluster, consulte Diretrizes para nomear um recurso do Serviço gerenciado para Apache Kafka. O nome de um cluster é imutável.

Local

O local em que você está criando o disco. O local precisa ser uma das regiões Google Cloud aceitas. O local de um cluster não pode ser mudado depois. Para uma lista dos locais disponíveis, consulte Locais do Serviço gerenciado para Apache Kafka.

Configuração de capacidade

A capacidade de computação, memória e armazenamento do cluster do Kafka. Ao criar um cluster, especifique os seguintes valores:

vCPUs: o número de vCPUs no cluster. São necessárias pelo menos três vCPUs por cluster.
Memória: a quantidade de memória atribuída ao cluster. É necessário provisionar entre 1 GiB e 8 GiB por vCPU.

Por exemplo, se você criar um cluster com 6 vCPUs, a memória mínima que poderá alocar para o cluster será de 6 GiB (1 GiB por vCPU), e a máxima será de 48 GiB (8 GiB por vCPU).

O serviço provisiona automaticamente brokers, incluindo volumes de Persistent Disk para eles. Se quiser, ative o Armazenamento local personalizado, que permite especificar a capacidade do disco para cada broker. Para mais informações, consulte Configurar o tamanho do disco do broker.

Para diretrizes sobre como estimar a capacidade necessária do cluster, consulte Planejar o tamanho do cluster do Kafka.

Configuração de rede

A configuração de rede é uma lista de sub-redes da VPC em que o cluster está acessível. Para produzir ou consumir mensagens, os clientes precisam acessar uma dessas sub-redes.

Confira algumas diretrizes para a configuração de rede:

É necessário ter pelo menos uma sub-rede para um cluster. O valor máximo é 10.
Exatamente uma sub-rede por rede é permitida para qualquer cluster.
Cada sub-rede precisa estar na mesma região que o cluster. O projeto e a rede podem ser diferentes.
Os endereços IP dos brokers e do servidor bootstrap são alocados automaticamente em cada sub-rede. Além disso, as entradas de DNS para esses endereços IP são criadas nas redes VPC correspondentes.
Se você adicionar uma sub-rede de um projeto diferente, conceda permissões à conta de serviço gerenciado pelo Google associada ao cluster. Para mais informações, consulte Conectar um cluster entre projetos.

Depois de criar o cluster, é possível atualizar a lista de sub-redes. Para mais informações sobre rede, consulte Configurar a rede para o Serviço Gerenciado para Apache Kafka.

Clusters públicos

Você pode configurar o cluster como público para permitir que aplicativos cliente se conectem a ele e executem operações do Kafka (incluindo envio e recebimento de mensagens) pela Internet pública. Para ativar o acesso público, especifique um ou mais intervalos de IP de origem permitidos.

Ao configurar intervalos de IP de origem permitidos, as seguintes regras se aplicam:

Todos os intervalos de IP de origem permitidos precisam ser especificados na notação CIDR IPv4 (por exemplo, 203.0.113.0/24 ou 198.51.100.5/32).
Cada tamanho de sub-rede CIDR precisa estar entre /16 e /32.
Os intervalos CIDR não podem se sobrepor.
Todos os intervalos de IP de origem permitidos precisam ser roteáveis publicamente. Intervalos de IP particulares, como 10.0.0.0/8 (RFC 1918), são rejeitados.
É possível especificar no máximo 500 intervalos de IP de origem permitidos.
Os endereços IPv6 não são aceitos.

Antes de ativar o acesso público para um cluster, recomendamos que você leia Conectar clientes a um cluster público para saber como o recurso de acesso público é implementado.

Rótulos

Os rótulos são pares de chave-valor que ajudam na organização e identificação. Com os rótulos, você pode categorizar recursos com base no ambiente. Exemplos são "env:production" e "owner:data-engineering".

É possível filtrar e pesquisar recursos com base nos rótulos deles. Por exemplo, suponha que você tenha vários clusters do Serviço Gerenciado para Apache Kafka para diferentes departamentos. É possível configurar e pesquisar clusters com o marcador "department:marketing" para encontrar o relevante.

Configuração de reequilíbrio

Essa configuração determina se o serviço rebalanceia automaticamente as réplicas de partição entre os brokers.

Os modos disponíveis são:

Rebalanceamento automático no escalonamento vertical: quando essa opção está ativada, o serviço aciona automaticamente um rebalanceamento de réplicas quando você escalonar verticalmente do cluster. Esse modo ajuda a manter uma distribuição de carga uniforme, mas pode afetar temporariamente o desempenho durante a operação de rebalanceamento.
Sem rebalanceamento: quando essa opção está ativada, o serviço não rebalanceia automaticamente as réplicas.

Para mais informações, consulte Rebalancear partições de broker.

Criptografia

O Serviço gerenciado para Apache Kafka pode criptografar mensagens com Google-owned and Google-managed encryption keys (padrão) ou chaves de criptografia gerenciadas pelo cliente (CMEK). Todas as mensagens são criptografadas em repouso e em trânsito. O tipo de criptografia de um cluster é imutável.

Google-owned and Google-managed encryption keys são usados por padrão. Essas chaves são criadas, gerenciadas e armazenadas inteiramente pela Google Cloud na infraestrutura dela.

As CMEKs são chaves de criptografia gerenciadas pelo Cloud Key Management Service. Com esse recurso, você tem mais controle sobre as chaves usadas para criptografar dados em repouso nos serviços compatíveis do Google Cloud . O uso da CMEK gera custos adicionais relacionados ao Cloud Key Management Service. Para usar a CMEK, seu keyring precisa estar no mesmo local dos recursos com que você o usa. Para mais informações, consulte Configurar a criptografia de mensagens.

Configuração de mTLS

Você pode configurar o mTLS como um método de autenticação alternativo que usa certificados do cliente. A configuração inclui o seguinte:

Pools de CA: uma lista de 1 a 10 pools do Certificate Authority Service (CAS) em que o cluster confia para autenticação de cliente.
Regras de mapeamento principal do SSL: uma propriedade do broker ssl.principal.mapping.rules opcional, mas recomendada, para simplificar nomes principais de certificados longos para uso em ACLs do Kafka.

Para mais informações sobre o mTLS, consulte Configurar a autenticação mTLS.

Criar um cluster

Antes de criar um cluster, leia a documentação sobre propriedades do cluster.

A criação de um cluster geralmente leva de 20 a 30 minutos.

Para criar um cluster, siga estas etapas:

Console

No Google Cloud console, acesse a página Clusters.

Acessar Clusters
Selecione Criar.

A página Criar cluster do Kafka é aberta.
Em Nome do cluster, insira uma string.

Para mais informações sobre como nomear um cluster, consulte as Diretrizes de nomeação de um recurso do Serviço gerenciado para Apache Kafka.
Em Local, insira um local compatível.

Para mais informações sobre os locais com suporte, consulte Locais com suporte do Serviço Gerenciado para Apache Kafka.
Em Configuração de capacidade, insira valores para Memória e vCPUs.

Para mais informações sobre como dimensionar um cluster, consulte Planejar o tamanho do cluster do Kafka.
Opcional. Para especificar a capacidade do disco do broker, faça o seguinte:
1. Clique no botão Ativar tamanho do armazenamento local personalizado para a posição ativada.
  
  Importante: depois de clicar em Salvar, não será possível desativar essa configuração novamente para esse cluster.
2. No campo Armazenamento local personalizado, insira o tamanho do disco por broker em GiB.
Se você deixar a opção Ativar o tamanho do armazenamento local personalizado desativada, a capacidade padrão do disco será provisionada. Para mais informações, consulte Configurar o tamanho do disco do broker.
Em Configuração de rede, insira os seguintes detalhes:
1. Projeto: o projeto em que a sub-rede está localizada. A sub-rede precisa estar na mesma região do cluster, mas o projeto pode ser diferente.
2. Rede: a rede a que a sub-rede está conectada.
3. Sub-rede: o nome da sub-rede.
4. Caminho de URI da sub-rede: este campo é preenchido automaticamente. Ou insira o caminho da sub-rede aqui. O nome da sub-rede precisa estar no formato: projects/PROJECT_ID/regions/REGION/subnetworks/SUBNET_ID.
5. Clique em Concluído.
Opcional: para adicionar mais sub-redes, clique em Adicionar uma sub-rede conectada.

É possível adicionar até 10 sub-redes.
Opcional: ative o acesso público ao cluster.
1. Na seção Configuração do cluster público, em O cluster pode ser acessado, selecione Em uma rede particular e usando a Internet pública.
2. No campo Lista separada por vírgulas de intervalos IPv4 de origem permitidos na notação CIDR, insira os intervalos de IP dos seus aplicativos cliente na notação CIDR IPv4 (por exemplo, 203.0.113.0/24). Os endereços IPv6 não são aceitos.
Opcional: defina uma chave de criptografia gerenciada pelo cliente (CMEK).
1. Em Criptografia, selecione Chave do Cloud KMS.
2. Em Tipo de chave, selecione Cloud KMS.
3. Em Selecionar uma chave gerenciada pelo cliente, selecione ou insira a CMEK.
  
  Se essa opção não aparecer, verifique se você ativou a API Cloud KMS para o projeto.
Para mais informações, consulte Configurar a criptografia de mensagens.
Clique em Criar.

gcloud

No console do Google Cloud , ative o Cloud Shell.

Ativar o Cloud Shell

Na parte de baixo do console Google Cloud , uma sessão do Cloud Shell é iniciada e exibe um prompt de linha de comando. O Cloud Shell é um ambiente shell com a CLI do Google Cloud já instalada e com valores já definidos para o projeto atual. A inicialização da sessão pode levar alguns segundos.

Antes de usar os dados do comando abaixo, faça estas substituições:

PROJECT_ID: o ID do projeto;
LOCATION: O local do cluster.
CLUSTER_ID: o ID do cluster.
CPU_COUNT: o número de vCPUs para o cluster.
MEMORY: a quantidade de memória para o cluster. Exemplo: 3GiB.
SUBNET_ID: o ID da sub-rede a ser conectada. Exemplo: default.
ENCRYPTION_KEY: o ID da chave de criptografia gerenciada pelo cliente a ser usada no cluster. Isso é opcional.
LABELS: os rótulos a serem associados ao cluster.
ALLOWED_SOURCE_IP_RANGES: os intervalos de IP de origem permitidos para o cluster público. Isso é opcional.

Execute o seguinte comando:

Linux, macOS ou Cloud Shell

gcloud managed-kafka clusters create CLUSTER_ID \
    --location=LOCATION \
    --cpu=CPU_COUNT \
    --memory=MEMORY \
    --subnets=projects/PROJECT_ID/regions/LOCATION/subnetworks/SUBNET_ID \
    --auto-rebalance \
    --encryption-key=ENCRYPTION_KEY \
    --async \
    --labels=LABELS \
    --public-cluster \
    --allowed-source-ip-ranges=ALLOWED_SOURCE_IP_RANGES

Windows (PowerShell)

gcloud managed-kafka clusters create CLUSTER_ID `
    --location=LOCATION `
    --cpu=CPU_COUNT `
    --memory=MEMORY `
    --subnets=projects/PROJECT_ID/regions/LOCATION/subnetworks/SUBNET_ID `
    --auto-rebalance `
    --encryption-key=ENCRYPTION_KEY `
    --async `
    --labels=LABELS `
    --public-cluster `
    --allowed-source-ip-ranges=ALLOWED_SOURCE_IP_RANGES

Windows (cmd.exe)

gcloud managed-kafka clusters create CLUSTER_ID ^
    --location=LOCATION ^
    --cpu=CPU_COUNT ^
    --memory=MEMORY ^
    --subnets=projects/PROJECT_ID/regions/LOCATION/subnetworks/SUBNET_ID ^
    --auto-rebalance ^
    --encryption-key=ENCRYPTION_KEY ^
    --async ^
    --labels=LABELS ^
    --public-cluster ^
    --allowed-source-ip-ranges=ALLOWED_SOURCE_IP_RANGES

Você receberá uma resposta semelhante a esta:

done: false
metadata:
  '@type': type.googleapis.com/google.cloud.managedkafka.v1.OperationMetadata
  apiVersion: v1
  createTime: 'CREATE_TIME'
  requestedCancellation: false
  target: projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID
  verb: create
name: projects/PROJECT_ID/locations/LOCATION/operations/OPERATION_ID

Armazene o OPERATION_ID para rastrear progress.

REST

Antes de usar os dados da solicitação abaixo, faça as substituições a seguir:

PROJECT_ID: o ID do projeto do Google Cloud
LOCATION: O local do cluster.
CLUSTER_ID: o ID do cluster.
CPU_COUNT: o número de vCPUs para o cluster.
MEMORY: a quantidade de memória para o cluster, em bytes. Exemplo: 3221225472.
SUBNET_ID: ID da sub-rede a ser conectada. Exemplo: default.

Método HTTP e URL:

POST https://managedkafka.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/clusters?clusterId=CLUSTER_ID

Corpo JSON da solicitação:

{
  "capacityConfig": {
    "vcpuCount": CPU_COUNT,
    "memoryBytes": MEMORY
  },
  "gcpConfig": {
    "accessConfig": {
      "networkConfigs": [
        {
          "subnet": "projects/PROJECT_ID/regions/LOCATION/subnetworks/SUBNET_ID"
        }
      ]
    }
  }
}

Para enviar a solicitação, expanda uma destas opções:

curl (Linux, macOS ou Cloud Shell)

Observação: o comando a seguir pressupõe que você tenha feito login na CLI do gcloud com sua conta de usuário executando gcloud init ou gcloud auth login, ou usando o Cloud Shell, que faz login automaticamente na CLI gcloud. . Para saber qual é a conta ativa no momento, execute o comando gcloud auth list.

Salve o corpo da solicitação em um arquivo com o nome request.json e execute o comando a seguir:

curl -X POST \
     -H "Authorization: Bearer $(gcloud auth print-access-token)" \
     -H "Content-Type: application/json; charset=utf-8" \
     -d @request.json \
     "https://managedkafka.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/clusters?clusterId=CLUSTER_ID"

PowerShell (Windows)

Observação: o comando a seguir pressupõe que você tenha feito login na CLI gcloud com sua conta de usuário executando gcloud init ou gcloud auth login . Para saber qual é a conta ativa no momento, execute o comando gcloud auth list.

Salve o corpo da solicitação em um arquivo com o nome request.json e execute o comando abaixo:

$cred = gcloud auth print-access-token
$headers = @{ "Authorization" = "Bearer $cred" }

Invoke-WebRequest `
    -Method POST `
    -Headers $headers `
    -ContentType: "application/json; charset=utf-8" `
    -InFile request.json `
    -Uri "https://managedkafka.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/clusters?clusterId=CLUSTER_ID" | Select-Object -Expand Content

Você receberá uma resposta JSON semelhante a esta:

{
  "name": "projects/PROJECT_ID/locations/LOCATION/operations/OPERATION_ID",
  "metadata": {
    "@type": "type.googleapis.com/google.cloud.managedkafka.v1.OperationMetadata",
    "createTime": "CREATE_TIME",
    "target": "projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID",
    "verb": "create",
    "requestedCancellation": false,
    "apiVersion": "v1"
  },
  "done": false
}

Terraform

É possível usar um recurso do Terraform para criar um cluster.

resource "google_managed_kafka_cluster" "default" {
  project    = data.google_project.default.project_id # Replace this with your project ID in quotes
  cluster_id = "my-cluster-id"
  location   = "us-central1"
  capacity_config {
    vcpu_count   = 3
    memory_bytes = 3221225472
  }
  gcp_config {
    access_config {
      network_configs {
        subnet = google_compute_subnetwork.default.id
      }
    }
  }
}

Para saber como aplicar ou remover uma configuração do Terraform, consulte Comandos básicos do Terraform.

Go

Antes de testar esta amostra, siga as instruções de configuração do Go em Instalar as bibliotecas de cliente. Para mais informações, consulte a documentação de referência da API Go do serviço gerenciado para Apache Kafka.

Para autenticar o serviço gerenciado para Apache Kafka, configure o Application Default Credentials(ADC). Para mais informações, consulte Configurar o ADC para um ambiente de desenvolvimento local.

import (
	"context"
	"fmt"
	"io"

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

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

func createCluster(w io.Writer, projectID, region, clusterID, subnet string, cpu, memoryBytes int64, opts ...option.ClientOption) error {
	// projectID := "my-project-id"
	// region := "us-central1"
	// clusterID := "my-cluster"
	// subnet := "projects/my-project-id/regions/us-central1/subnetworks/default"
	// cpu := 3
	// memoryBytes := 3221225472
	ctx := context.Background()
	client, err := managedkafka.NewClient(ctx, opts...)
	if err != nil {
		return fmt.Errorf("managedkafka.NewClient got err: %w", err)
	}
	defer client.Close()

	locationPath := fmt.Sprintf("projects/%s/locations/%s", projectID, region)
	clusterPath := fmt.Sprintf("%s/clusters/%s", locationPath, clusterID)

	// Memory must be between 1 GiB and 8 GiB per CPU.
	capacityConfig := &managedkafkapb.CapacityConfig{
		VcpuCount:   cpu,
		MemoryBytes: memoryBytes,
	}
	var networkConfig []*managedkafkapb.NetworkConfig
	networkConfig = append(networkConfig, &managedkafkapb.NetworkConfig{
		Subnet: subnet,
	})
	platformConfig := &managedkafkapb.Cluster_GcpConfig{
		GcpConfig: &managedkafkapb.GcpConfig{
			AccessConfig: &managedkafkapb.AccessConfig{
				NetworkConfigs: networkConfig,
			},
		},
	}
	rebalanceConfig := &managedkafkapb.RebalanceConfig{
		Mode: managedkafkapb.RebalanceConfig_AUTO_REBALANCE_ON_SCALE_UP,
	}
	cluster := &managedkafkapb.Cluster{
		Name:            clusterPath,
		CapacityConfig:  capacityConfig,
		PlatformConfig:  platformConfig,
		RebalanceConfig: rebalanceConfig,
	}

	req := &managedkafkapb.CreateClusterRequest{
		Parent:    locationPath,
		ClusterId: clusterID,
		Cluster:   cluster,
	}
	op, err := client.CreateCluster(ctx, req)
	if err != nil {
		return fmt.Errorf("client.CreateCluster got err: %w", err)
	}
	// The duration of this operation can vary considerably, typically taking 10-40 minutes.
	resp, err := op.Wait(ctx)
	if err != nil {
		return fmt.Errorf("op.Wait got err: %w", err)
	}
	fmt.Fprintf(w, "Created cluster: %s\n", resp.Name)
	return nil
}

Java

Antes de testar esta amostra, siga as instruções de configuração do Java em Instalar as bibliotecas de cliente. Para mais informações, consulte a documentação de referência da API Java do serviço gerenciado para Apache Kafka.

Para autenticar no Serviço gerenciado para Apache Kafka, configure o Application Default Credentials. Para mais informações, consulte Configurar o ADC para um ambiente de desenvolvimento local.


import com.google.api.gax.longrunning.OperationFuture;
import com.google.api.gax.longrunning.OperationSnapshot;
import com.google.api.gax.longrunning.OperationTimedPollAlgorithm;
import com.google.api.gax.retrying.RetrySettings;
import com.google.api.gax.retrying.RetryingFuture;
import com.google.api.gax.retrying.TimedRetryAlgorithm;
import com.google.cloud.managedkafka.v1.AccessConfig;
import com.google.cloud.managedkafka.v1.CapacityConfig;
import com.google.cloud.managedkafka.v1.Cluster;
import com.google.cloud.managedkafka.v1.CreateClusterRequest;
import com.google.cloud.managedkafka.v1.GcpConfig;
import com.google.cloud.managedkafka.v1.LocationName;
import com.google.cloud.managedkafka.v1.ManagedKafkaClient;
import com.google.cloud.managedkafka.v1.ManagedKafkaSettings;
import com.google.cloud.managedkafka.v1.NetworkConfig;
import com.google.cloud.managedkafka.v1.OperationMetadata;
import com.google.cloud.managedkafka.v1.RebalanceConfig;
import java.time.Duration;
import java.util.concurrent.ExecutionException;

public class CreateCluster {

  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-cluster";
    String subnet = "my-subnet"; // e.g. projects/my-project/regions/my-region/subnetworks/my-subnet
    int cpu = 3;
    long memoryBytes = 3221225472L; // 3 GiB
    createCluster(projectId, region, clusterId, subnet, cpu, memoryBytes);
  }

  public static void createCluster(
      String projectId, String region, String clusterId, String subnet, int cpu, long memoryBytes)
      throws Exception {
    CapacityConfig capacityConfig =
        CapacityConfig.newBuilder().setVcpuCount(cpu).setMemoryBytes(memoryBytes).build();
    NetworkConfig networkConfig = NetworkConfig.newBuilder().setSubnet(subnet).build();
    GcpConfig gcpConfig =
        GcpConfig.newBuilder()
            .setAccessConfig(AccessConfig.newBuilder().addNetworkConfigs(networkConfig).build())
            .build();
    RebalanceConfig rebalanceConfig =
        RebalanceConfig.newBuilder()
            .setMode(RebalanceConfig.Mode.AUTO_REBALANCE_ON_SCALE_UP)
            .build();
    Cluster cluster =
        Cluster.newBuilder()
            .setCapacityConfig(capacityConfig)
            .setGcpConfig(gcpConfig)
            .setRebalanceConfig(rebalanceConfig)
            .build();

    // Create the settings to configure the timeout for polling operations
    ManagedKafkaSettings.Builder settingsBuilder = ManagedKafkaSettings.newBuilder();
    TimedRetryAlgorithm timedRetryAlgorithm = OperationTimedPollAlgorithm.create(
        RetrySettings.newBuilder()
            .setTotalTimeoutDuration(Duration.ofHours(1L))
            .build());
    settingsBuilder.createClusterOperationSettings()
        .setPollingAlgorithm(timedRetryAlgorithm);

    try (ManagedKafkaClient managedKafkaClient = ManagedKafkaClient.create(
        settingsBuilder.build())) {

      CreateClusterRequest request =
          CreateClusterRequest.newBuilder()
              .setParent(LocationName.of(projectId, region).toString())
              .setClusterId(clusterId)
              .setCluster(cluster)
              .build();

      // The duration of this operation can vary considerably, typically taking between 10-40
      // minutes.
      OperationFuture<Cluster, OperationMetadata> future =
          managedKafkaClient.createClusterOperationCallable().futureCall(request);

      // Get the initial LRO and print details.
      OperationSnapshot operation = future.getInitialFuture().get();
      System.out.printf("Cluster creation started. Operation name: %s\nDone: %s\nMetadata: %s\n",
          operation.getName(),
          operation.isDone(),
          future.getMetadata().get().toString());

      while (!future.isDone()) {
        // The pollingFuture gives us the most recent status of the operation
        RetryingFuture<OperationSnapshot> pollingFuture = future.getPollingFuture();
        OperationSnapshot currentOp = pollingFuture.getAttemptResult().get();
        System.out.printf("Polling Operation:\nName: %s\n Done: %s\n",
            currentOp.getName(),
            currentOp.isDone());
      }

      // NOTE: future.get() blocks completion until the operation is complete (isDone =  True)
      Cluster response = future.get();
      System.out.printf("Created cluster: %s\n", response.getName());
    } catch (ExecutionException e) {
      System.err.printf("managedKafkaClient.createCluster got err: %s", e.getMessage());
    }
  }
}

Python

Antes de testar esta amostra, siga as instruções de configuração do Python em Instalar as bibliotecas de cliente. Para mais informações, consulte a documentação de referência da API Python do serviço gerenciado para Apache Kafka.