Create Lite topic

Create a single-partition Lite topic and configure its retention period, per partition storage, and publishing throughput scaling factor.

Documentation pages that include this code sample

To view the code sample used in context, see the following documentation:

Code sample

Java

import com.google.cloud.pubsublite.AdminClient;
import com.google.cloud.pubsublite.AdminClientSettings;
import com.google.cloud.pubsublite.CloudRegion;
import com.google.cloud.pubsublite.CloudZone;
import com.google.cloud.pubsublite.ProjectNumber;
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.RetentionConfig;
import com.google.protobuf.util.Durations;

public class CreateTopicExample {

  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";
    long projectNumber = Long.parseLong("123456789");
    int partitions = 1;

    createTopicExample(cloudRegion, zoneId, projectNumber, topicId, partitions);
  }

  public static void createTopicExample(
      String cloudRegion, char zoneId, long projectNumber, String topicId, int partitions)
      throws Exception {

    TopicPath topicPath =
        TopicPath.newBuilder()
            .setProject(ProjectNumber.of(projectNumber))
            .setLocation(CloudZone.of(CloudRegion.of(cloudRegion), zoneId))
            .setName(TopicName.of(topicId))
            .build();

    Topic topic =
        Topic.newBuilder()
            .setPartitionConfig(
                PartitionConfig.newBuilder()
                    // Set throughput capacity per partition in MiB/s.
                    .setCapacity(
                        PartitionConfig.Capacity.newBuilder()
                            // Must be 4-16 MiB/s.
                            .setPublishMibPerSec(4)
                            // Must be 4-32 MiB/s.
                            .setSubscribeMibPerSec(8)
                            .build())
                    .setCount(partitions))
            .setRetentionConfig(
                RetentionConfig.newBuilder()
                    // How long messages are retained.
                    .setPeriod(Durations.fromDays(1))
                    // Set storage per partition to 30 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`.
                    .setPerPartitionBytes(30 * 1024 * 1024 * 1024L))
            .setName(topicPath.toString())
            .build();

    AdminClientSettings adminClientSettings =
        AdminClientSettings.newBuilder().setRegion(CloudRegion.of(cloudRegion)).build();

    try (AdminClient adminClient = AdminClient.create(adminClientSettings)) {
      Topic response = adminClient.createTopic(topic).get();
      System.out.println(response.getAllFields() + "created successfully.");
    }
  }
}

Python

from google.api_core.exceptions import AlreadyExists
from google.cloud.pubsublite import AdminClient, Topic
from google.cloud.pubsublite.types import CloudRegion, CloudZone, TopicPath
from google.protobuf.duration_pb2 import Duration

# TODO(developer):
# project_number = 1122334455
# cloud_region = "us-central1"
# zone_id = "a"
# topic_id = "your-topic-id"
# num_partitions = 1

cloud_region = CloudRegion(cloud_region)
location = CloudZone(cloud_region, zone_id)
topic_path = TopicPath(project_number, location, topic_id)
topic = Topic(
    name=str(topic_path),
    partition_config=Topic.PartitionConfig(
        # A topic must have at least one partition.
        count=num_partitions,
        # Set throughput capacity per partition in MiB/s.
        capacity=Topic.PartitionConfig.Capacity(
            # Set publish throughput capacity per partition to 4 MiB/s. Must be >= 4 and <= 16.
            publish_mib_per_sec=4,
            # Set subscribe throughput capacity per partition to 4 MiB/s. Must be >= 4 and <= 32.
            subscribe_mib_per_sec=8,
        ),
    ),
    retention_config=Topic.RetentionConfig(
        # Set storage per partition to 30 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`.
        per_partition_bytes=30 * 1024 * 1024 * 1024,
        # Allow messages to be retained for 7 days.
        period=Duration(seconds=60 * 60 * 24 * 7),
    ),
)

client = AdminClient(cloud_region)
try:
    response = client.create_topic(topic)
    print(f"{response.name} created successfully.")
except AlreadyExists:
    print(f"{topic_path} already exists.")