Memperbarui topik Google Cloud Managed Service for Apache Kafka

Setelah topik dibuat, Anda dapat mengedit konfigurasi topik untuk memperbarui properti berikut: jumlah partisi dan konfigurasi topik yang tidak menggunakan properti default yang sudah ditetapkan di tingkat cluster. Anda hanya dapat meningkatkan jumlah partisi, tidak dapat menguranginya.

Untuk memperbarui satu topik, Anda dapat menggunakan Google Cloud konsol, Google Cloud CLI, library klien, Managed Kafka API, atau Apache Kafka API open source.

Peran dan izin yang diperlukan untuk mengedit topik

Untuk mendapatkan izin yang diperlukan untuk mengedit topik, minta administrator Anda untuk memberi Anda peran IAM Editor Topik Managed Kafka(roles/managedkafka.topicEditor) di project Anda. Untuk mengetahui informasi selengkapnya tentang pemberian peran, lihat Mengelola akses ke project, folder, dan organisasi.

Peran bawaan ini berisi izin yang diperlukan untuk mengedit topik. Untuk melihat izin yang benar-benar diperlukan, perluas bagian Izin yang diperlukan:

Izin yang diperlukan

Izin berikut diperlukan untuk mengedit topik:

  • Memperbarui topik: managedkafka.topics.update

Anda mungkin juga bisa mendapatkan izin ini dengan peran khusus atau peran bawaan lainnya.

Mengedit topik

Untuk mengedit topik, ikuti langkah-langkah berikut:

Konsol

  1. Di Google Cloud konsol, buka halaman Clusters.

    Buka Cluster

    Cluster yang Anda buat dalam project akan tercantum.

  2. Klik cluster yang berisi topik yang ingin Anda edit.

    Halaman Cluster details akan terbuka. Di halaman detail cluster, untuk tab Resources, topik akan tercantum.

  3. Klik topik yang ingin Anda edit.

    Halaman Topic details akan terbuka.

  4. Untuk melakukan pengeditan, klik Edit.

  5. Klik Save setelah melakukan perubahan.

gcloud

  1. Di konsol, aktifkan Cloud Shell. Google Cloud

    Aktifkan Cloud Shell

    Di bagian bawah konsol Google Cloud , sesi Cloud Shell akan dimulai dan menampilkan prompt command line. Cloud Shell adalah lingkungan shell dengan Google Cloud CLI yang sudah terinstal, dan dengan nilai yang sudah ditetapkan untuk project Anda saat ini. Diperlukan waktu beberapa detik untuk melakukan inisialisasi pada sesi.

  2. Jalankan perintah gcloud managed-kafka topics update:

    gcloud managed-kafka topics update TOPIC_ID \
  --cluster=CLUSTER_ID \
  --location=LOCATION_ID \
  --partitions=PARTITIONS \
  --configs=CONFIGS

    Perintah ini mengubah konfigurasi topik yang ada di cluster Managed Service untuk Apache Kafka yang ditentukan. Anda dapat menggunakan perintah ini untuk meningkatkan jumlah partisi dan memperbarui setelan konfigurasi tingkat topik.

    Ganti kode berikut:

    • TOPIC_ID: ID topik.
    • CLUSTER_ID: ID cluster yang berisi topik.
    • LOCATION_ID: Lokasi cluster.
    • PARTITIONS: Opsional: Jumlah partisi terbaru untuk topik. Anda hanya dapat meningkatkan jumlah partisi, tidak dapat menguranginya.
    • CONFIGS: Opsional: Daftar setelan konfigurasi yang akan diperbarui. Tentukan sebagai daftar pasangan kunci-nilai yang dipisahkan koma. Contoh: retention.ms=604800000,retention.bytes=10000000.

REST

Sebelum menggunakan salah satu data permintaan, lakukan penggantian berikut:

  • PROJECT_ID: ID proyek Google Cloud Anda
  • LOCATION: lokasi cluster
  • CLUSTER_ID: ID cluster
  • TOPIC_ID: ID topik
  • UPDATE_MASK: kolom yang akan diperbarui, sebagai daftar yang dipisahkan koma dari nama yang sepenuhnya memenuhi syarat. Contoh: partitionCount
  • PARTITION_COUNT: jumlah partisi terbaru untuk topik

Metode HTTP dan URL: 

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

Meminta isi JSON: 

{
  "name": "TOPIC_ID",
  "partitionCount": PARTITION_COUNT
}

Untuk mengirim permintaan Anda, perluas salah satu opsi berikut:

Anda akan melihat respons JSON seperti berikut:

{
  "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
}

Go

Sebelum mencoba contoh ini, ikuti petunjuk penyiapan Go di Menginstal library klien. Untuk mengetahui informasi selengkapnya, lihat Dokumentasi referensi API Managed Service untuk Apache Kafka Go.

Untuk melakukan autentikasi ke Managed Service untuk Apache Kafka, siapkan Kredensial Default Aplikasi(ADC). Untuk mengetahui informasi selengkapnya, lihat Menyiapkan ADC untuk lingkungan pengembangan lokal.

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 updateTopic(w io.Writer, projectID, region, clusterID, topicID string, partitionCount int32, configs map[string]string, opts ...option.ClientOption) error {
	// projectID := "my-project-id"
	// region := "us-central1"
	// clusterID := "my-cluster"
	// topicID := "my-topic"
	// partitionCount := 20
	// configs := map[string]string{"min.insync.replicas":"1"}
	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)
	topicPath := fmt.Sprintf("%s/topics/%s", clusterPath, topicID)
	TopicConfig := managedkafkapb.Topic{
		Name:           topicPath,
		PartitionCount: partitionCount,
		Configs:        configs,
	}
	paths := []string{"partition_count", "configs"}
	updateMask := &fieldmaskpb.FieldMask{
		Paths: paths,
	}

	req := &managedkafkapb.UpdateTopicRequest{
		UpdateMask: updateMask,
		Topic:      &TopicConfig,
	}
	topic, err := client.UpdateTopic(ctx, req)
	if err != nil {
		return fmt.Errorf("client.UpdateTopic got err: %w", err)
	}
	fmt.Fprintf(w, "Updated topic: %#v\n", topic)
	return nil
}

Java

Sebelum mencoba contoh ini, ikuti petunjuk penyiapan Java di Menginstal library klien. Untuk mengetahui informasi selengkapnya, lihat Dokumentasi referensi API Managed Service untuk Apache Kafka Java.

Untuk melakukan autentikasi ke Managed Service untuk Apache Kafka, siapkan Kredensial Default Aplikasi. Untuk mengetahui informasi selengkapnya, lihat Menyiapkan ADC untuk lingkungan pengembangan lokal.

import com.google.api.gax.rpc.ApiException;
import com.google.cloud.managedkafka.v1.ManagedKafkaClient;
import com.google.cloud.managedkafka.v1.Topic;
import com.google.cloud.managedkafka.v1.TopicName;
import com.google.cloud.managedkafka.v1.UpdateTopicRequest;
import com.google.protobuf.FieldMask;
import java.io.IOException;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;

public class UpdateTopic {

  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 topicId = "my-topic";
    int partitionCount = 200;
    Map<String, String> configs =
        new HashMap<String, String>() {
          {
            put("min.insync.replicas", "1");
          }
        };
    updateTopic(projectId, region, clusterId, topicId, partitionCount, configs);
  }

  public static void updateTopic(
      String projectId,
      String region,
      String clusterId,
      String topicId,
      int partitionCount,
      Map<String, String> configs)
      throws Exception {
    Topic topic =
        Topic.newBuilder()
            .setName(TopicName.of(projectId, region, clusterId, topicId).toString())
            .setPartitionCount(partitionCount)
            .putAllConfigs(configs)
            .build();
    String[] paths = {"partition_count", "configs"};
    FieldMask updateMask = FieldMask.newBuilder().addAllPaths(Arrays.asList(paths)).build();
    try (ManagedKafkaClient managedKafkaClient = ManagedKafkaClient.create()) {
      UpdateTopicRequest request =
          UpdateTopicRequest.newBuilder().setUpdateMask(updateMask).setTopic(topic).build();
      // This operation is being handled synchronously.
      Topic response = managedKafkaClient.updateTopic(request);
      System.out.printf("Updated topic: %s\n", response.getName());
    } catch (IOException | ApiException e) {
      System.err.printf("managedKafkaClient.updateCluster got err: %s", e.getMessage());
    }
  }
}

Python

Sebelum mencoba contoh ini, ikuti petunjuk penyiapan Python di Menginstal library klien. Untuk mengetahui informasi selengkapnya, lihat Dokumentasi referensi Managed Service untuk Apache Kafka Python API.

Untuk melakukan autentikasi ke Managed Service untuk Apache Kafka, siapkan Kredensial Default Aplikasi. Untuk mengetahui informasi selengkapnya, lihat Menyiapkan ADC untuk lingkungan pengembangan lokal.

from google.api_core.exceptions import NotFound
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"
# topic_id = "my-topic"
# partition_count = 20
# configs = {"min.insync.replicas": "1"}

client = managedkafka_v1.ManagedKafkaClient()

topic = managedkafka_v1.Topic()
topic.name = client.topic_path(project_id, region, cluster_id, topic_id)
topic.partition_count = partition_count
topic.configs = configs
update_mask = field_mask_pb2.FieldMask()
update_mask.paths.extend(["partition_count", "configs"])

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

try:
    response = client.update_topic(request=request)
    print("Updated topic:", response)
except NotFound as e:
    print(f"Failed to update topic {topic_id} with error: {e.message}")

Mengonfigurasi retensi pesan

Kafka menyimpan pesan dalam file segmen log. Secara default, Kafka menghapus file segmen setelah periode retensi atau saat partisi melebihi batas ukuran data. Anda dapat mengubah perilaku ini dengan mengaktifkan pemadatan log. Jika pemadatan log diaktifkan, Kafka hanya akan menyimpan nilai terbaru untuk setiap kunci.

Google Cloud Managed Service untuk Apache Kafka menggunakan penyimpanan bertingkat, yang berarti segmen log yang telah selesai disimpan dari jarak jauh, bukan penyimpanan lokal. Untuk mempelajari penyimpanan bertingkat lebih lanjut, lihat Penyimpanan Bertingkat dalam dokumentasi Apache Kafka.

Menetapkan nilai retensi

Jika pemadatan log tidak diaktifkan, setelan berikut akan mengontrol cara Kafka menyimpan file segmen log:

  • retention.ms: Durasi maksimum untuk menyimpan file segmen, dalam milidetik.
  • retention.bytes: Jumlah byte maksimum yang akan disimpan per partisi. Jika data dalam partisi melebihi nilai ini, Kafka akan menghapus file segmen yang lebih lama.

Untuk memperbarui setelan ini, gunakan gcloud CLI atau Kafka CLI:

gcloud

Untuk menetapkan retensi pesan, jalankan gcloud managed-kafka topics update perintah.

gcloud managed-kafka topics update TOPIC_ID \
  --cluster=CLUSTER_ID \
  --location=LOCATION_ID \
  --configs=retention.ms=RETENTION_PERIOD,retention.bytes=MAX_BYTES

Ganti kode berikut:

  • TOPIC_ID: ID topik.
  • CLUSTER_ID: ID cluster yang berisi topik.
  • LOCATION_ID: Lokasi cluster.
  • RETENTION_PERIOD: Jumlah waktu maksimum untuk menyimpan file segmen, dalam milidetik.
  • MAX_BYTES: Jumlah byte maksimum yang akan disimpan, per partisi.

Kafka CLI

Sebelum menjalankan perintah ini, instal alat command line Kafka di VM Compute Engine. VM harus dapat menjangkau subnet yang terhubung ke cluster Managed Service untuk Apache Kafka Anda. Ikuti petunjuk di Membuat dan menggunakan pesan dengan alat command line Kafka.

Jalankan perintah kafka-configs.sh:

kafka-configs.sh --alter \
  --bootstrap-server=BOOTSTRAP_ADDRESS \
  --command-config client.properties \
  --entity-type topics \
  --entity-name TOPIC_ID \
  --add-config retention.ms=RETENTION_PERIOD,retention.bytes=MAX_BYTES

Ganti kode berikut:

Mengaktifkan pemadatan log

Jika pemadatan log diaktifkan, Kafka hanya akan menyimpan pesan terbaru untuk setiap kunci. Pemadatan log dinonaktifkan secara default. Untuk mengaktifkan pemadatan log untuk topik, tetapkan konfigurasi cleanup.policy ke "compact", sebagai berikut:

gcloud

Jalankan gcloud managed-kafka topics update perintah.

gcloud managed-kafka topics update TOPIC_ID \
  --cluster=CLUSTER_ID \
  --location=LOCATION_ID \
  --configs=cleanup.policy=compact

Ganti kode berikut:

  • TOPIC_ID: ID topik.
  • CLUSTER_ID: ID cluster yang berisi topik.
  • LOCATION_ID: Lokasi cluster.

Kafka CLI

Sebelum menjalankan perintah ini, instal alat command line Kafka di VM Compute Engine. VM harus dapat menjangkau subnet yang terhubung ke cluster Managed Service untuk Apache Kafka Anda. Ikuti petunjuk di Membuat dan menggunakan pesan dengan alat command line Kafka.

Jalankan perintah kafka-configs.sh:

kafka-configs.sh --alter \
  --bootstrap-server=BOOTSTRAP_ADDRESS \
  --command-config client.properties \
  --entity-type topics \
  --entity-name TOPIC_ID \
  --add-config cleanup.policy=compact

Ganti kode berikut:

Batasan

  • Anda tidak dapat mengganti konfigurasi topik untuk penyimpanan jarak jauh, seperti remote.storage.enable.

  • Anda tidak dapat mengganti konfigurasi topik untuk file segmen log, seperti segment.bytes.

  • Mengaktifkan pemadatan log untuk topik secara implisit akan menonaktifkan penyimpanan bertingkat untuk topik tersebut. Semua file log untuk topik disimpan secara lokal.

Apa langkah selanjutnya?