Update a Google Cloud Managed Service for Apache Kafka cluster

Console

1. In the Google Cloud console, go to the Clusters page.

Go to Clusters

1. From the list of clusters, click the cluster whose properties you want to edit.

The console displays the cluster details page.

1. On the cluster details page, click Edit.

2. Edit the properties as required. You can edit the following properties of a cluster from the console:

   • Memory
   • vCPUs
   • Subnet
   • Rebalancing configuration
   • mTLS configuration
   • Labels

3. Click Save.

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. Before using any of the command data below, make the following replacements:

   • PROJECT_ID: the project ID.
   • LOCATION: the location of the cluster.
   • CLUSTER_ID: the ID of the cluster.
   • CPU_COUNT: the number of vCPUs for the cluster.
   • MEMORY: the amount of memory for the cluster. Example: 10GiB.
   • SUBNET_ID: the subnet ID of the subnet to connect to. Example: default.
   • LABELS: the labels to associate with the cluster.
   • ALLOWED_SOURCE_IP_RANGES: the allowed source IPv4 CIDR ranges for public cluster internet access.

   Execute the following command:

   Linux, macOS, or Cloud Shell

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

   Windows (PowerShell)

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

   Windows (cmd.exe)

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

   You should receive a response similar to the following:

   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: update
   name: projects/PROJECT_ID/locations/LOCATION/operations/OPERATION_ID
   

   • To disable public access, use the --no-public-cluster flag.
   • If you use the --async flag with your command, the system sends the update request and immediately returns a response, without waiting for the operation to complete. With the --async flag, you can continue with other tasks while the cluster update happens in the background. If you don't use the --async flag, the system waits for the operation to complete before returning a response. You must wait until the cluster is fully updated before you can continue with other tasks.

REST

Before using any of the request data, make the following replacements:

• PROJECT_ID: your Google Cloud project ID
• LOCATION: the location of the cluster.
• CLUSTER_ID: the ID of the cluster.
• UPDATE_MASK: which fields to update, as a comma-separated list of fully qualified names. Example: capacityConfig.vcpuCount,capacityConfig.memoryBytes
• CPU_COUNT: the number of vCPUs for the cluster.
• MEMORY: the amount of memory for the cluster, in bytes. Example: 3221225472.
• SUBNET_ID: subnet ID of the subnet to connect to. Example: default.

HTTP method and URL:

PATCH https://managedkafka.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID?updateMask=UPDATE_MASK

Request JSON body:

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

To send your request, expand one of these options:

curl -X PATCH \
     -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/CLUSTER_ID?updateMask=UPDATE_MASK"

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

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

You should receive a JSON response similar to the following:

{
  "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": "update",
    "requestedCancellation": false,
    "apiVersion": "v1"
  },
  "done": false
}

In the request body, include only the fields that you are updating, as specified in the UPDATE_MASK query parameter.

• To add a subnet, append a new entry to networkConfigs in the following format: projects/PROJECT_ID/regions/LOCATION/subnetworks/SUBNET_ID. Example: projects/sample-project/regions/us-central1/subnetworks/default.

• To enable public access or update the allowed source IP ranges, include gcpConfig.accessConfig.publicClusterConfig in the UPDATE_MASK query parameter and specify the allowedSourceIpRanges array in the request body. Example request body:

  {
    "gcpConfig": {
      "accessConfig": {
        "publicClusterConfig": {
          "allowedSourceIpRanges": [
            "203.0.113.0/24"
          ]
        }
      }
    }
  }
  

• To disable public access, include gcpConfig.accessConfig.publicClusterConfig in the UPDATE_MASK query parameter and pass an empty JSON object {} in the request body (or omit publicClusterConfig). Example request body:

  {}
  

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/option"
	"google.golang.org/protobuf/types/known/fieldmaskpb"

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

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

	clusterPath := fmt.Sprintf("projects/%s/locations/%s/clusters/%s", projectID, region, clusterID)
	capacityConfig := &managedkafkapb.CapacityConfig{
		MemoryBytes: memory,
	}
	cluster := &managedkafkapb.Cluster{
		Name:           clusterPath,
		CapacityConfig: capacityConfig,
	}
	paths := []string{"capacity_config.memory_bytes"}
	updateMask := &fieldmaskpb.FieldMask{
		Paths: paths,
	}

	req := &managedkafkapb.UpdateClusterRequest{
		UpdateMask: updateMask,
		Cluster:    cluster,
	}
	op, err := client.UpdateCluster(ctx, req)
	if err != nil {
		return fmt.Errorf("client.UpdateCluster got err: %w", err)
	}
	resp, err := op.Wait(ctx)
	if err != nil {
		return fmt.Errorf("op.Wait got err: %w", err)
	}
	fmt.Fprintf(w, "Updated cluster: %#v\n", resp)
	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.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.TimedRetryAlgorithm;
import com.google.cloud.managedkafka.v1.CapacityConfig;
import com.google.cloud.managedkafka.v1.Cluster;
import com.google.cloud.managedkafka.v1.ClusterName;
import com.google.cloud.managedkafka.v1.ManagedKafkaClient;
import com.google.cloud.managedkafka.v1.ManagedKafkaSettings;
import com.google.cloud.managedkafka.v1.OperationMetadata;
import com.google.cloud.managedkafka.v1.UpdateClusterRequest;
import com.google.protobuf.FieldMask;
import java.time.Duration;
import java.util.concurrent.ExecutionException;

public class UpdateCluster {

  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";
    long memoryBytes = 25769803776L; // 24 GiB
    updateCluster(projectId, region, clusterId, memoryBytes);
  }

  public static void updateCluster(
      String projectId, String region, String clusterId, long memoryBytes) throws Exception {
    CapacityConfig capacityConfig = CapacityConfig.newBuilder().setMemoryBytes(memoryBytes).build();
    Cluster cluster =
        Cluster.newBuilder()
            .setName(ClusterName.of(projectId, region, clusterId).toString())
            .setCapacityConfig(capacityConfig)
            .build();
    FieldMask updateMask = FieldMask.newBuilder().addPaths("capacity_config.memory_bytes").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.updateClusterOperationSettings()
        .setPollingAlgorithm(timedRetryAlgorithm);

    try (ManagedKafkaClient managedKafkaClient = ManagedKafkaClient.create(
        settingsBuilder.build())) {
      UpdateClusterRequest request =
          UpdateClusterRequest.newBuilder().setUpdateMask(updateMask).setCluster(cluster).build();
      OperationFuture<Cluster, OperationMetadata> future =
          managedKafkaClient.updateClusterOperationCallable().futureCall(request);

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

      Cluster response = future.get();
      System.out.printf("Updated cluster: %s\n", response.getName());
    } catch (ExecutionException e) {
      System.err.printf("managedKafkaClient.updateCluster got err: %s", 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.

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.

from google.api_core.exceptions import GoogleAPICallError
from google.cloud import managedkafka_v1
from google.protobuf import field_mask_pb2

# TODO(developer)
# project_id = "my-project-id"
# region = "us-central1"
# cluster_id = "my-cluster"
# memory_bytes = 4295000000

client = managedkafka_v1.ManagedKafkaClient()

cluster = managedkafka_v1.Cluster()
cluster.name = client.cluster_path(project_id, region, cluster_id)
cluster.capacity_config.memory_bytes = memory_bytes
update_mask = field_mask_pb2.FieldMask()
update_mask.paths.append("capacity_config.memory_bytes")

# For a list of editable fields, one can check https://cloud.google.com/managed-kafka/docs/create-cluster#properties.
request = managedkafka_v1.UpdateClusterRequest(
    update_mask=update_mask,
    cluster=cluster,
)

try:
    operation = client.update_cluster(request=request)
    print(f"Waiting for operation {operation.operation.name} to complete...")
    response = operation.result()
    print("Updated cluster:", response)
except GoogleAPICallError as e:
    print(f"The operation failed with error: {e.message}")