How to get the best performance for reading messages from Kafka, then publishing to pub/sub topic?

[yufanzhang]

I develop Spring KafkaListener to read messages from Kafka topic, then publish them to GCP Pub/sub topic. The application is running in GCP Cloud Run. Our Kafka topic is configured with 9 partitions. In GCP, we configure our Cloud Run service to set both min/max instances = 9 because we find setting min/max instances = 100 has the same performance as setting min/max instances = 9 when Kafka only has 9 partitions.

For above setting, I find, for one instance, the max pub/sub publishing performance is about 2,500 messages per second (2,500/sec). I want to whether that is the max performance for GCP Pub/Sub publisher. If I want to reach 1,000,000/sec, beside increasing Kafka partition along with increasing Cloud Run instances, what else can I do on coding enhancement?

My coding for publishing messages is referencing GCP Cloud sample codes: https://cloud.google.com/pubsub/docs/batch-messaging#use_batch_messaging

My code for reading Kafka topic messages:

@KafkaListener( topics = {"my-kafka-topic" }, containerFactory = "kafkaListenerContainerFactory", groupId = "my-group-id")

public void onMessage(List<ConsumerRecord<String, String>> consumerRecords) throws Exception { 

  Publisher publisher = getPublisher();

List<ApiFuture<String>> messageIdFutures = new ArrayList<>();

  try {

// Assume the Kafka message batch size=1000, while Pub/Sub publishing batch size = 100

// Therefore, we run 10 publisher processes in parallel here

Lists

.partition(consumerRecords, 100)

.parallelStream()

.forEach(messages -> messageIdFutures.addAll(publishingMessages(messages, publisher)));

} finally {

// Wait on any pending publish requests.

List<String> messageIds = ApiFutures.allAsList(messageIdFutures).get();

System.out.println("Published " + messageIds.size() + " messages with batch settings.");

if (publisher != null) {

// When finished with the publisher, shutdown to free up resources.

publisher.shutdown();

publisher.awaitTermination(1, TimeUnit.MINUTES);

}

}

}