packagemainimport("context""flag""fmt""log""sync""cloud.google.com/go/pubsub""cloud.google.com/go/pubsublite/pscompat""golang.org/x/sync/errgroup")funcmain(){// NOTE: Set these flags for an existing Pub/Sub Lite topic when running this// sample.projectID:=flag.String("project_id","","Cloud Project ID")zone:=flag.String("zone","","Cloud Zone where the topic resides, e.g. us-central1-a")topicID:=flag.String("topic_id","","Existing Pub/Sub Lite topic")messageCount:=flag.Int("message_count",100,"The number of messages to send")flag.Parse()ctx:=context.Background()topicPath:=fmt.Sprintf("projects/%s/locations/%s/topics/%s",*projectID,*zone,*topicID)// Create the publisher client.publisher,err:=pscompat.NewPublisherClient(ctx,topicPath)iferr!=nil{log.Fatalf("pscompat.NewPublisherClient error: %v",err)}// Ensure the publisher will be shut down.deferpublisher.Stop()// Collect any messages that need to be republished with a new publisher// client.vartoRepublish[]*pubsub.MessagevartoRepublishMusync.Mutex// Publish messages. Messages are automatically batched.g:=new(errgroup.Group)fori:=0;i <*messageCount;i++{msg:=&pubsub.Message{Data:[]byte(fmt.Sprintf("message-%d",i)),}result:=publisher.Publish(ctx,msg)g.Go(func()error{// Get blocks until the result is ready.id,err:=result.Get(ctx)iferr!=nil{// NOTE: A failed PublishResult indicates that the publisher client// encountered a fatal error and has permanently terminated. After the// fatal error has been resolved, a new publisher client instance must// be created to republish failed messages.fmt.Printf("Publish error: %v\n",err)toRepublishMu.Lock()toRepublish=append(toRepublish,msg)toRepublishMu.Unlock()returnerr}// Metadata decoded from the id contains the partition and offset.metadata,err:=pscompat.ParseMessageMetadata(id)iferr!=nil{fmt.Printf("Failed to parse message metadata %q: %v\n",id,err)returnerr}fmt.Printf("Published: partition=%d, offset=%d\n",metadata.Partition,metadata.Offset)returnnil})}iferr:=g.Wait();err!=nil{fmt.Printf("Publishing finished with error: %v\n",err)}fmt.Printf("Published %d messages\n",*messageCount-len(toRepublish))// Print the error that caused the publisher client to terminate (if any),// which may contain more context than PublishResults.iferr:=publisher.Error();err!=nil{fmt.Printf("Publisher client terminated due to error: %v\n",publisher.Error())}}

importcom.google.api.core.ApiFuture;importcom.google.api.core.ApiFutures;importcom.google.api.gax.rpc.ApiException;importcom.google.cloud.pubsublite.CloudRegion;importcom.google.cloud.pubsublite.CloudRegionOrZone;importcom.google.cloud.pubsublite.CloudZone;importcom.google.cloud.pubsublite.MessageMetadata;importcom.google.cloud.pubsublite.ProjectNumber;importcom.google.cloud.pubsublite.TopicName;importcom.google.cloud.pubsublite.TopicPath;importcom.google.cloud.pubsublite.cloudpubsub.Publisher;importcom.google.cloud.pubsublite.cloudpubsub.PublisherSettings;importcom.google.protobuf.ByteString;importcom.google.pubsub.v1.PubsubMessage;importjava.util.ArrayList;importjava.util.List;importjava.util.concurrent.ExecutionException;publicclassPublisherExample{publicstaticvoidmain(String...args)throwsException{// TODO(developer): Replace these variables before running the sample.StringcloudRegion="your-cloud-region";charzoneId='b';// Choose an existing topic for the publish example to work.StringtopicId="your-topic-id";longprojectNumber=Long.parseLong("123456789");intmessageCount=100;// True if using a regional location. False if using a zonal location.// https://cloud.google.com/pubsub/lite/docs/topicsbooleanregional=false;publisherExample(cloudRegion,zoneId,projectNumber,topicId,messageCount,regional);}// Publish messages to a topic.publicstaticvoidpublisherExample(StringcloudRegion,charzoneId,longprojectNumber,StringtopicId,intmessageCount,booleanregional)throwsApiException,ExecutionException,InterruptedException{CloudRegionOrZonelocation;if(regional){location=CloudRegionOrZone.of(CloudRegion.of(cloudRegion));}else{location=CloudRegionOrZone.of(CloudZone.of(CloudRegion.of(cloudRegion),zoneId));}TopicPathtopicPath=TopicPath.newBuilder().setProject(ProjectNumber.of(projectNumber)).setLocation(location).setName(TopicName.of(topicId)).build();Publisherpublisher=null;List<ApiFuture<String>>futures=newArrayList<>();try{PublisherSettingspublisherSettings=PublisherSettings.newBuilder().setTopicPath(topicPath).build();publisher=Publisher.create(publisherSettings);// Start the publisher. Upon successful starting, its state will become RUNNING.publisher.startAsync().awaitRunning();for(inti=0;i <messageCount;i++){Stringmessage="message-"+i;// Convert the message to a byte string.ByteStringdata=ByteString.copyFromUtf8(message);PubsubMessagepubsubMessage=PubsubMessage.newBuilder().setData(data).build();// Publish a message. Messages are automatically batched.ApiFuture<String>future=publisher.publish(pubsubMessage);futures.add(future);}}finally{ArrayList<MessageMetadata>metadata=newArrayList<>();List<String>ackIds=ApiFutures.allAsList(futures).get();for(Stringid:ackIds){// Decoded metadata contains partition and offset.metadata.add(MessageMetadata.decode(id));}System.out.println(metadata+"\nPublished "+ackIds.size()+" messages.");if(publisher!=null){// Shut down the publisher.publisher.stopAsync().awaitTerminated();System.out.println("Publisher is shut down.");}}}}

fromgoogle.cloud.pubsublite.cloudpubsubimportPublisherClientfromgoogle.cloud.pubsublite.typesimport(CloudRegion,CloudZone,MessageMetadata,TopicPath,)# TODO(developer):# project_number = 1122334455# cloud_region = "us-central1"# zone_id = "a"# topic_id = "your-topic-id"# regional = Trueifregional:location=CloudRegion(cloud_region)else:location=CloudZone(CloudRegion(cloud_region),zone_id)topic_path=TopicPath(project_number,location,topic_id)# PublisherClient() must be used in a `with` block or have __enter__() called before use.withPublisherClient()aspublisher_client:data="Hello world!"api_future=publisher_client.publish(topic_path,data.encode("utf-8"))# result() blocks. To resolve API futures asynchronously, use add_done_callback().message_id=api_future.result()message_metadata=MessageMetadata.decode(message_id)print(f"Published a message to{topic_path} with partition{message_metadata.partition.value} and offset{message_metadata.cursor.offset}.")

import("context""fmt""io""cloud.google.com/go/pubsub""cloud.google.com/go/pubsublite/pscompat")funcpublishWithOrderingKey(wio.Writer,projectID,zone,topicIDstring,messageCountint)error{// projectID := "my-project-id"// zone := "us-central1-a"// topicID := "my-topic"// messageCount := 10ctx:=context.Background()topicPath:=fmt.Sprintf("projects/%s/locations/%s/topics/%s",projectID,zone,topicID)// Create the publisher client.publisher,err:=pscompat.NewPublisherClient(ctx,topicPath)iferr!=nil{returnfmt.Errorf("pscompat.NewPublisherClient error: %w",err)}// Ensure the publisher will be shut down.deferpublisher.Stop()// Messages of the same ordering key will always get published to the same// partition. When OrderingKey is unset, messages can get published to// different partitions if more than one partition exists for the topic.varresults[]*pubsub.PublishResultfori:=0;i <messageCount;i++{r:=publisher.Publish(ctx,&pubsub.Message{OrderingKey:"test_ordering_key",Data:[]byte(fmt.Sprintf("message-%d",i)),})results=append(results,r)}// Print publish results.varpublishedCountintfor_,r:=rangeresults{// Get blocks until the result is ready.id,err:=r.Get(ctx)iferr!=nil{// NOTE: A failed PublishResult indicates that the publisher client// encountered a fatal error and has permanently terminated. After the// fatal error has been resolved, a new publisher client instance must be// created to republish failed messages.fmt.Fprintf(w,"Publish error: %v\n",err)continue}// Metadata decoded from the id contains the partition and offset.metadata,err:=pscompat.ParseMessageMetadata(id)iferr!=nil{returnfmt.Errorf("failed to parse message metadata %q: %w",id,err)}fmt.Fprintf(w,"Published: partition=%d, offset=%d\n",metadata.Partition,metadata.Offset)publishedCount++}fmt.Fprintf(w,"Published %d messages with ordering key\n",publishedCount)returnpublisher.Error()}

importcom.google.api.core.ApiFuture;importcom.google.api.gax.rpc.ApiException;importcom.google.cloud.pubsublite.CloudRegion;importcom.google.cloud.pubsublite.CloudRegionOrZone;importcom.google.cloud.pubsublite.CloudZone;importcom.google.cloud.pubsublite.MessageMetadata;importcom.google.cloud.pubsublite.ProjectNumber;importcom.google.cloud.pubsublite.TopicName;importcom.google.cloud.pubsublite.TopicPath;importcom.google.cloud.pubsublite.cloudpubsub.Publisher;importcom.google.cloud.pubsublite.cloudpubsub.PublisherSettings;importcom.google.protobuf.ByteString;importcom.google.pubsub.v1.PubsubMessage;importjava.util.concurrent.ExecutionException;publicclassPublishWithOrderingKeyExample{publicstaticvoidmain(String...args)throwsException{// TODO(developer): Replace these variables before running the sample.StringcloudRegion="your-cloud-region";charzoneId='b';// Choose an existing topic for the publish example to work.StringtopicId="your-topic-id";longprojectNumber=Long.parseLong("123456789");// True if using a regional location. False if using a zonal location.// https://cloud.google.com/pubsub/lite/docs/topicsbooleanregional=false;publishWithOrderingKeyExample(cloudRegion,zoneId,projectNumber,topicId,regional);}// Publish a message to a topic with an ordering key.publicstaticvoidpublishWithOrderingKeyExample(StringcloudRegion,charzoneId,longprojectNumber,StringtopicId,booleanregional)throwsApiException,ExecutionException,InterruptedException{CloudRegionOrZonelocation;if(regional){location=CloudRegionOrZone.of(CloudRegion.of(cloudRegion));}else{location=CloudRegionOrZone.of(CloudZone.of(CloudRegion.of(cloudRegion),zoneId));}TopicPathtopicPath=TopicPath.newBuilder().setProject(ProjectNumber.of(projectNumber)).setLocation(location).setName(TopicName.of(topicId)).build();PublisherSettingspublisherSettings=PublisherSettings.newBuilder().setTopicPath(topicPath).build();Publisherpublisher=Publisher.create(publisherSettings);// Start the publisher. Upon successful starting, its state will become RUNNING.publisher.startAsync().awaitRunning();Stringmessage="message-with-ordering-key";// Convert the message to a byte string.ByteStringdata=ByteString.copyFromUtf8(message);PubsubMessagepubsubMessage=PubsubMessage.newBuilder().setData(data)// Messages of the same ordering key will always get published to the// same partition. When OrderingKey is unset, messages can get published// to different partitions if more than one partition exists for the topic..setOrderingKey("testing").build();// Publish a message.ApiFuture<String>future=publisher.publish(pubsubMessage);// Shut down the publisher.publisher.stopAsync().awaitTerminated();StringackId=future.get();MessageMetadatametadata=MessageMetadata.decode(ackId);System.out.println("Published a message with ordering key:\n"+metadata);}}

fromgoogle.cloud.pubsublite.cloudpubsubimportPublisherClientfromgoogle.cloud.pubsublite.typesimport(CloudRegion,CloudZone,MessageMetadata,TopicPath,)# TODO(developer):# project_number = 1122334455# cloud_region = "us-central1"# zone_id = "a"# topic_id = "your-topic-id"# num_messages = 100# regional = Trueifregional:location=CloudRegion(cloud_region)else:location=CloudZone(CloudRegion(cloud_region),zone_id)topic_path=TopicPath(project_number,location,topic_id)# PublisherClient() must be used in a `with` block or have __enter__() called before use.withPublisherClient()aspublisher_client:formessageinrange(num_messages):data=f"{message}"# Messages of the same ordering key will always get published to the same partition.# When ordering_key is unset, messsages can get published ot different partitions if# more than one partition exists for the topic.api_future=publisher_client.publish(topic_path,data.encode("utf-8"),ordering_key="testing")# result() blocks. To resolve api futures asynchronously, use add_done_callback().message_id=api_future.result()message_metadata=MessageMetadata.decode(message_id)print(f"Published{data} to partition{message_metadata.partition.value} and offset{message_metadata.cursor.offset}.")print(f"Finished publishing{num_messages} messages with an ordering key to{str(topic_path)}.")

import("context""fmt""io""cloud.google.com/go/pubsub""cloud.google.com/go/pubsublite/pscompat")funcpublishWithCustomAttributes(wio.Writer,projectID,zone,topicIDstring)error{// projectID := "my-project-id"// zone := "us-central1-a"// topicID := "my-topic"ctx:=context.Background()topicPath:=fmt.Sprintf("projects/%s/locations/%s/topics/%s",projectID,zone,topicID)// Create the publisher client.publisher,err:=pscompat.NewPublisherClient(ctx,topicPath)iferr!=nil{returnfmt.Errorf("pscompat.NewPublisherClient error: %w",err)}// Ensure the publisher will be shut down.deferpublisher.Stop()// Publish a message with custom attributes.result:=publisher.Publish(ctx,&pubsub.Message{Data:[]byte("message-with-custom-attributes"),Attributes:map[string]string{"year":"2020","author":"unknown",},})// Get blocks until the result is ready.id,err:=result.Get(ctx)iferr!=nil{returnfmt.Errorf("publish error: %w",err)}fmt.Fprintf(w,"Published a message with custom attributes: %v\n",id)returnpublisher.Error()}

importcom.google.api.core.ApiFuture;importcom.google.api.gax.rpc.ApiException;importcom.google.cloud.pubsublite.CloudRegion;importcom.google.cloud.pubsublite.CloudRegionOrZone;importcom.google.cloud.pubsublite.CloudZone;importcom.google.cloud.pubsublite.MessageMetadata;importcom.google.cloud.pubsublite.ProjectNumber;importcom.google.cloud.pubsublite.TopicName;importcom.google.cloud.pubsublite.TopicPath;importcom.google.cloud.pubsublite.cloudpubsub.MessageTransforms;importcom.google.cloud.pubsublite.cloudpubsub.Publisher;importcom.google.cloud.pubsublite.cloudpubsub.PublisherSettings;importcom.google.common.collect.ImmutableMap;importcom.google.protobuf.ByteString;importcom.google.protobuf.util.Timestamps;importcom.google.pubsub.v1.PubsubMessage;importjava.time.Instant;importjava.util.concurrent.ExecutionException;publicclassPublishWithCustomAttributesExample{publicstaticvoidmain(String...args)throwsException{// TODO(developer): Replace these variables before running the sample.StringcloudRegion="your-cloud-region";charzoneId='b';// Choose an existing topic for the publish example to work.StringtopicId="your-topic-id";longprojectNumber=Long.parseLong("123456789");// True if using a regional location. False if using a zonal location.// https://cloud.google.com/pubsub/lite/docs/topicsbooleanregional=false;publishWithCustomAttributesExample(cloudRegion,zoneId,projectNumber,topicId,regional);}// Publish messages to a topic with custom attributes.publicstaticvoidpublishWithCustomAttributesExample(StringcloudRegion,charzoneId,longprojectNumber,StringtopicId,booleanregional)throwsApiException,ExecutionException,InterruptedException{CloudRegionOrZonelocation;if(regional){location=CloudRegionOrZone.of(CloudRegion.of(cloudRegion));}else{location=CloudRegionOrZone.of(CloudZone.of(CloudRegion.of(cloudRegion),zoneId));}TopicPathtopicPath=TopicPath.newBuilder().setProject(ProjectNumber.of(projectNumber)).setLocation(location).setName(TopicName.of(topicId)).build();PublisherSettingspublisherSettings=PublisherSettings.newBuilder().setTopicPath(topicPath).build();Publisherpublisher=Publisher.create(publisherSettings);// Start the publisher. Upon successful starting, its state will become RUNNING.publisher.startAsync().awaitRunning();// Prepare the message data as a byte string.StringmessageData="message-with-custom-attributes";ByteStringdata=ByteString.copyFromUtf8(messageData);// Prepare a protobuf-encoded event timestamp for the message.Instantnow=Instant.now();StringeventTime=MessageTransforms.encodeAttributeEventTime(Timestamps.fromMillis(now.toEpochMilli()));PubsubMessagepubsubMessage=PubsubMessage.newBuilder().setData(data)// Add two sets of custom attributes to the message..putAllAttributes(ImmutableMap.of("year","2020","author","unknown"))// Add an event timestamp as an attribute..putAttributes(MessageTransforms.PUBSUB_LITE_EVENT_TIME_TIMESTAMP_PROTO,eventTime).build();// Publish a message.ApiFuture<String>future=publisher.publish(pubsubMessage);// Shut down the publisher.publisher.stopAsync().awaitTerminated();StringackId=future.get();MessageMetadatametadata=MessageMetadata.decode(ackId);System.out.println("Published a message with custom attributes:\n"+metadata);}}

fromgoogle.cloud.pubsublite.cloudpubsubimportPublisherClientfromgoogle.cloud.pubsublite.typesimport(CloudRegion,CloudZone,MessageMetadata,TopicPath,)# TODO(developer):# project_number = 1122334455# cloud_region = "us-central1"# zone_id = "a"# topic_id = "your-topic-id"# regional = Trueifregional:location=CloudRegion(cloud_region)else:location=CloudZone(CloudRegion(cloud_region),zone_id)topic_path=TopicPath(project_number,location,topic_id)# PublisherClient() must be used in a `with` block or have __enter__() called before use.withPublisherClient()aspublisher_client:data="Hello world!"api_future=publisher_client.publish(topic_path,data.encode("utf-8"),year="2020",author="unknown",)# result() blocks. To resolve api futures asynchronously, use add_done_callback().message_id=api_future.result()message_metadata=MessageMetadata.decode(message_id)print(f"Published{data} to partition{message_metadata.partition.value} and offset{message_metadata.cursor.offset}.")print(f"Finished publishing a message with custom attributes to{str(topic_path)}.")

import("context""fmt""io""time""cloud.google.com/go/pubsub""cloud.google.com/go/pubsublite/pscompat")funcpublishWithBatchSettings(wio.Writer,projectID,zone,topicIDstring,messageCountint)error{// projectID := "my-project-id"// zone := "us-central1-a"// topicID := "my-topic"// messageCount := 10ctx:=context.Background()topicPath:=fmt.Sprintf("projects/%s/locations/%s/topics/%s",projectID,zone,topicID)// Batch settings control how the publisher batches messages. These settings// apply per partition.// See https://pkg.go.dev/cloud.google.com/go/pubsublite/pscompat#pkg-variables// for DefaultPublishSettings.settings:=pscompat.PublishSettings{ByteThreshold:5*1024,// 5 KiBCountThreshold:1000,// 1,000 messagesDelayThreshold:100*time.Millisecond,}// Create the publisher client.publisher,err:=pscompat.NewPublisherClientWithSettings(ctx,topicPath,settings)iferr!=nil{returnfmt.Errorf("pscompat.NewPublisherClientWithSettings error: %w",err)}// Ensure the publisher will be shut down.deferpublisher.Stop()// Publish requests are sent to the server based on request size, message// count and time since last publish, whichever condition is met first.varresults[]*pubsub.PublishResultfori:=0;i <messageCount;i++{r:=publisher.Publish(ctx,&pubsub.Message{Data:[]byte(fmt.Sprintf("message-%d",i)),})results=append(results,r)}// Print publish results.varpublishedCountintfor_,r:=rangeresults{// Get blocks until the result is ready.id,err:=r.Get(ctx)iferr!=nil{// NOTE: A failed PublishResult indicates that the publisher client// encountered a fatal error and has permanently terminated. After the// fatal error has been resolved, a new publisher client instance must be// created to republish failed messages.fmt.Fprintf(w,"Publish error: %v\n",err)continue}fmt.Fprintf(w,"Published: %v\n",id)publishedCount++}fmt.Fprintf(w,"Published %d messages with batch settings\n",publishedCount)returnpublisher.Error()}

importcom.google.api.core.ApiFuture;importcom.google.api.core.ApiFutures;importcom.google.api.gax.batching.BatchingSettings;importcom.google.api.gax.rpc.ApiException;importcom.google.cloud.pubsublite.CloudRegion;importcom.google.cloud.pubsublite.CloudRegionOrZone;importcom.google.cloud.pubsublite.CloudZone;importcom.google.cloud.pubsublite.MessageMetadata;importcom.google.cloud.pubsublite.ProjectNumber;importcom.google.cloud.pubsublite.TopicName;importcom.google.cloud.pubsublite.TopicPath;importcom.google.cloud.pubsublite.cloudpubsub.Publisher;importcom.google.cloud.pubsublite.cloudpubsub.PublisherSettings;importcom.google.protobuf.ByteString;importcom.google.pubsub.v1.PubsubMessage;importjava.util.ArrayList;importjava.util.List;importjava.util.concurrent.ExecutionException;importorg.threeten.bp.Duration;publicclassPublishWithBatchSettingsExample{publicstaticvoidmain(String...args)throwsException{// TODO(developer): Replace these variables before running the sample.StringcloudRegion="your-cloud-region";charzoneId='b';// Choose an existing topic for the publish example to work.StringtopicId="your-topic-id";longprojectNumber=Long.parseLong("123456789");intmessageCount=100;// True if using a regional location. False if using a zonal location.// https://cloud.google.com/pubsub/lite/docs/topicsbooleanregional=false;publishWithBatchSettingsExample(cloudRegion,zoneId,projectNumber,topicId,messageCount,regional);}// Publish messages to a topic with batch settings.publicstaticvoidpublishWithBatchSettingsExample(StringcloudRegion,charzoneId,longprojectNumber,StringtopicId,intmessageCount,booleanregional)throwsApiException,ExecutionException,InterruptedException{CloudRegionOrZonelocation;if(regional){location=CloudRegionOrZone.of(CloudRegion.of(cloudRegion));}else{location=CloudRegionOrZone.of(CloudZone.of(CloudRegion.of(cloudRegion),zoneId));}TopicPathtopicPath=TopicPath.newBuilder().setProject(ProjectNumber.of(projectNumber)).setLocation(location).setName(TopicName.of(topicId)).build();Publisherpublisher=null;List<ApiFuture<String>>futures=newArrayList<>();try{// Batch settings control how the publisher batches messageslongrequestBytesThreshold=5000L;// default : 3_500_000 byteslongmessageCountBatchSize=100L;// default : 1000L messageDurationpublishDelayThreshold=Duration.ofMillis(100);// default : 50 ms// Publish request get triggered based on request size, messages count & time since last// publish, whichever condition is met first.BatchingSettingsbatchingSettings=BatchingSettings.newBuilder().setRequestByteThreshold(requestBytesThreshold).setElementCountThreshold(messageCountBatchSize).setDelayThreshold(publishDelayThreshold).build();PublisherSettingspublisherSettings=PublisherSettings.newBuilder().setTopicPath(topicPath).setBatchingSettings(batchingSettings).build();publisher=Publisher.create(publisherSettings);// Start the publisher. Upon successful starting, its state will become RUNNING.publisher.startAsync().awaitRunning();for(inti=0;i <messageCount;i++){Stringmessage="message-"+i;// Convert the message to a byte string.ByteStringdata=ByteString.copyFromUtf8(message);PubsubMessagepubsubMessage=PubsubMessage.newBuilder().setData(data).build();// Publish a message.ApiFuture<String>future=publisher.publish(pubsubMessage);futures.add(future);}}finally{ArrayList<MessageMetadata>metadata=newArrayList<>();List<String>ackIds=ApiFutures.allAsList(futures).get();System.out.println("Published "+ackIds.size()+" messages with batch settings.");if(publisher!=null){// Shut down the publisher.publisher.stopAsync().awaitTerminated();}}}}

fromgoogle.cloud.pubsub_v1.typesimportBatchSettingsfromgoogle.cloud.pubsublite.cloudpubsubimportPublisherClientfromgoogle.cloud.pubsublite.typesimport(CloudRegion,CloudZone,MessageMetadata,TopicPath,)# TODO(developer):# project_number = 1122334455# cloud_region = "us-central1"# zone_id = "a"# topic_id = "your-topic-id"# num_messages = 100# regional = Trueifregional:location=CloudRegion(cloud_region)else:location=CloudZone(CloudRegion(cloud_region),zone_id)topic_path=TopicPath(project_number,location,topic_id)batch_setttings=BatchSettings(# 2 MiB. Default to 3 MiB. Must be less than 4 MiB gRPC's per-message limit.max_bytes=2*1024*1024,# 100 ms. Default to 50 ms.max_latency=0.1,# Default to 1000.max_messages=100,)# PublisherClient() must be used in a `with` block or have __enter__() called before use.withPublisherClient(per_partition_batching_settings=batch_setttings)aspublisher_client:formessageinrange(num_messages):data=f"{message}"api_future=publisher_client.publish(topic_path,data.encode("utf-8"))# result() blocks. To resolve API futures asynchronously, use add_done_callback().message_id=api_future.result()message_metadata=MessageMetadata.decode(message_id)print(f"Published{data} to partition{message_metadata.partition.value} and offset{message_metadata.cursor.offset}.")print(f"Finished publishing{num_messages} messages with batch settings to{str(topic_path)}.")