Mettez à jour la durée de conservation, le facteur de scaling (capacité de débit) et le nombre d'octets de partition (capacité de stockage) d'un sujet Lite.
En savoir plus
Pour obtenir une documentation détaillée incluant cet exemple de code, consultez les articles suivants :
Exemple de code
Go
import (
"context"
"fmt"
"io"
"time"
"cloud.google.com/go/pubsublite"
)
func updateTopic(w io.Writer, projectID, region, location, topicID, reservation string) error {
// projectID := "my-project-id"
// region := "us-central1"
// NOTE: location can be either a region ("us-central1") or a zone ("us-central1-a")
// For a list of valid locations, see https://cloud.google.com/pubsub/lite/docs/locations.
// location := "us-central1"
// topicID := "my-topic"
// reservation := "projects/my-project-id/reservations/my-reservation"
ctx := context.Background()
client, err := pubsublite.NewAdminClient(ctx, region)
if err != nil {
return fmt.Errorf("pubsublite.NewAdminClient: %v", err)
}
defer client.Close()
topicPath := fmt.Sprintf("projects/%s/locations/%s/topics/%s", projectID, location, topicID)
// For ranges of fields in TopicConfigToUpdate, see https://pkg.go.dev/cloud.google.com/go/pubsublite/#TopicConfigToUpdate
config := pubsublite.TopicConfigToUpdate{
Name: topicPath,
PartitionCount: 3, // Partition count cannot decrease.
PublishCapacityMiBPerSec: 8,
SubscribeCapacityMiBPerSec: 16,
PerPartitionBytes: 60 * 1024 * 1024 * 1024,
RetentionDuration: 24 * time.Hour,
ThroughputReservation: reservation,
}
updatedCfg, err := client.UpdateTopic(ctx, config)
if err != nil {
return fmt.Errorf("client.UpdateTopic got err: %v", err)
}
fmt.Fprintf(w, "Updated topic: %v\n", updatedCfg)
return nil
}
Java
import com.google.api.gax.rpc.NotFoundException;
import com.google.cloud.pubsublite.AdminClient;
import com.google.cloud.pubsublite.AdminClientSettings;
import com.google.cloud.pubsublite.CloudRegion;
import com.google.cloud.pubsublite.CloudRegionOrZone;
import com.google.cloud.pubsublite.CloudZone;
import com.google.cloud.pubsublite.ProjectNumber;
import com.google.cloud.pubsublite.ReservationName;
import com.google.cloud.pubsublite.ReservationPath;
import com.google.cloud.pubsublite.TopicName;
import com.google.cloud.pubsublite.TopicPath;
import com.google.cloud.pubsublite.proto.Topic;
import com.google.cloud.pubsublite.proto.Topic.PartitionConfig;
import com.google.cloud.pubsublite.proto.Topic.PartitionConfig.Capacity;
import com.google.cloud.pubsublite.proto.Topic.ReservationConfig;
import com.google.cloud.pubsublite.proto.Topic.RetentionConfig;
import com.google.protobuf.FieldMask;
import com.google.protobuf.util.Durations;
import java.util.Arrays;
import java.util.concurrent.ExecutionException;
public class UpdateTopicExample {
public static void main(String... args) throws Exception {
// TODO(developer): Replace these variables before running the sample.
String cloudRegion = "your-cloud-region";
char zoneId = 'b';
String topicId = "your-topic-id";
String reservationId = "your-reservation-id";
long projectNumber = Long.parseLong("123456789");
// True if using a regional location. False if using a zonal location.
// https://cloud.google.com/pubsub/lite/docs/topics
boolean regional = true;
updateTopicExample(cloudRegion, zoneId, projectNumber, topicId, reservationId, regional);
}
public static void updateTopicExample(
String cloudRegion,
char zoneId,
long projectNumber,
String topicId,
String reservationId,
boolean regional)
throws Exception {
CloudRegionOrZone location;
if (regional) {
location = CloudRegionOrZone.of(CloudRegion.of(cloudRegion));
} else {
location = CloudRegionOrZone.of(CloudZone.of(CloudRegion.of(cloudRegion), zoneId));
}
TopicPath topicPath =
TopicPath.newBuilder()
.setProject(ProjectNumber.of(projectNumber))
.setLocation(location)
.setName(TopicName.of(topicId))
.build();
ReservationPath reservationPath =
ReservationPath.newBuilder()
.setProject(ProjectNumber.of(projectNumber))
.setLocation(CloudRegion.of(cloudRegion))
.setName(ReservationName.of(reservationId))
.build();
Iterable<String> iterablePaths =
Arrays.asList(
"partition_config.scale",
"retention_config.per_partition_bytes",
"retention_config.period",
"reservation_config.throughput_reservation");
FieldMask fieldMask = FieldMask.newBuilder().addAllPaths(iterablePaths).build();
Topic topic =
Topic.newBuilder()
.setPartitionConfig(
PartitionConfig.newBuilder()
.setCapacity(
Capacity.newBuilder()
.setPublishMibPerSec(16)
.setSubscribeMibPerSec(32)
.build())
.build())
.setRetentionConfig(
RetentionConfig.newBuilder()
// Set storage per partition to 32 GiB. This must be 30 GiB-10 TiB.
// If the number of bytes stored in any of the topic's partitions grows
// beyond this value, older messages will be dropped to make room for
// newer ones, regardless of the value of `period`.
// Be careful when decreasing storage per partition as it may cause
// lost messages.
.setPerPartitionBytes(32 * 1024 * 1024 * 1024L)
.setPeriod(Durations.fromDays(7)))
.setReservationConfig(
ReservationConfig.newBuilder()
.setThroughputReservation(reservationPath.toString())
.build())
.setName(topicPath.toString())
.build();
AdminClientSettings adminClientSettings =
AdminClientSettings.newBuilder().setRegion(CloudRegion.of(cloudRegion)).build();
try (AdminClient adminClient = AdminClient.create(adminClientSettings)) {
Topic topicBeforeUpdate = adminClient.getTopic(topicPath).get();
System.out.println("Before update: " + topicBeforeUpdate.getAllFields());
Topic topicAfterUpdate = adminClient.updateTopic(topic, fieldMask).get();
System.out.println("After update: " + topicAfterUpdate.getAllFields());
} catch (ExecutionException e) {
try {
throw e.getCause();
} catch (NotFoundException notFound) {
System.out.println("This topic is not found.");
} catch (Throwable throwable) {
throwable.printStackTrace();
}
}
}
}Python
from google.api_core.exceptions import NotFound
from google.cloud.pubsublite import AdminClient, Topic
from google.cloud.pubsublite.types import (
CloudRegion,
CloudZone,
ReservationPath,
TopicPath,
)
from google.protobuf.duration_pb2 import Duration
from google.protobuf.field_mask_pb2 import FieldMask
# TODO(developer):
# project_number = 1122334455
# cloud_region = "us-central1"
# zone_id = "a"
# topic_id = "your-topic-id"
# reservation_id = "your-reservation-id"
# regional = True
location = None
if regional:
# A region.
location = CloudRegion(cloud_region)
else:
# A zone.
location = CloudZone(CloudRegion(cloud_region), zone_id)
topic_path = TopicPath(project_number, location, topic_id)
reservation_path = ReservationPath(project_number, cloud_region, reservation_id)
# Defines which topic fields to update.
field_mask = FieldMask(
paths=[
"partition_config.capacity",
"retention_config.per_partition_bytes",
"retention_config.period",
"reservation_confing.throughput_reservation",
]
)
# Defines how to update the topic fields.
topic = Topic(
name=str(topic_path),
partition_config=Topic.PartitionConfig(
capacity=Topic.PartitionConfig.Capacity(
publish_mib_per_sec=16, subscribe_mib_per_sec=32,
)
),
retention_config=Topic.RetentionConfig(
# Set storage per partition to 32 GiB. This must be in the range 30 GiB-10TiB.
# If the number of byptes stored in any of the topic's partitions grows beyond
# this value, older messages will be dropped to make room for newer ones,
# regardless of the value of `period`.
# Be careful when decreasing storage per partition as it may cuase lost messages.
per_partition_bytes=32 * 1024 * 1024 * 1024,
# Allow messages to be stored for 14 days.
period=Duration(seconds=60 * 60 * 24 * 14),
),
reservation_config=Topic.ReservationConfig(
throughput_reservation=str(reservation_path),
),
)
client = AdminClient(cloud_region)
try:
response = client.update_topic(topic, field_mask)
print(f"{response.name} updated successfully.")
except NotFound:
print(f"{topic_path} not found.")Étapes suivantes
Pour rechercher et filtrer des exemples de code pour d'autres produits Google Cloud, consultez l'exemple de navigateur Google Cloud.