Test / Experiment Batching locally

[Shalima]

Hi,

I am trying to experiment with batching parameters in Producer configuration locally. Since batching is enabled by default, I tried to set maxMessages to 3. I have a consumer listening to the topic. I am using send in for loop from Producer. I tried to send 5 messages and all 5 messages were consumed same time. How do I confirm the messages are batched ? This might be more of an amateur question and I am new to Pulsar. Any help will be appreciated. Thank you.

[michaeljmarshall]

Hi @Shalima, would you mind sharing a sample of your producer and consumer code? That will likely be an efficient way to help answer your question. Thanks.

[Shalima]

Producer.

Producer<byte[]> producer = client.newProducer().topic(topicName).create();
		log.info("Created producer: " + topicName);

		for (int i = 0; i < numMsgs; i++) {
			String msg = "Message  " + i;
			byte[] bytePayload = msg.getBytes();
			MessageId msgId = producer.newMessage().value(bytePayload).send();
			log.info("Message Produced. Producer Name: " + producer.getProducerName() + " Message Id: " + msgId);
		}
		producer.close();

Consumer

Consumer consumer = client.newConsumer()
				.topic(topicName)
				.subscriptionName(subscriptionName).subscribe();
		List<String> msgReceived = new ArrayList<String>();
		while(true){
			Message msg = consumer.receive();
			
		        byte[] plainPayload = decryptor.decrypt(msg.getData());
		        String msgText = new String(plainPayload);
		        msgReceived.add(msgText);
		        log.info("Consumer Name: " + consumer.getConsumerName() + " Message : " + msgText);
		        System.out.println(String.format("Message Recieved consumer_name=%s message=%s",
				        consumer.getConsumerName(), msgText));
		        consumer.acknowledge(msg);
		}
		consumer.close();
		client.close();

[michaeljmarshall]

Thank you for this sample code. The first thing to note is that using synchronous sends by calling .send() will likely not result in batched messages because the send() call returns only when the broker sends acknowledgement that the message has been persisted, when using a persistent topic. Technically, if you share the producer across threads and call send within the message batching window, it is possible to use synchronous sends and get some benefit of batching. The main way to benefit from batching is to use the sendAsync method and then track the result with the returned future.

[Shalima]

Thank you so much for the response.

[michaeljmarshall]

How do I confirm the messages are batched ?

There are a few details here. First, message batching is an internal pulsar optimization that is transparent to the producer and consumer, so it isn't technically something that a user should have to verify. That being said, if you are not getting the throughput that you would like to see, it's always possible that the message batching could be tuned. The primary settings are set when building the producer using the newProducer() method, and they are batchingMaxPublishDelay(long, TimeUnit), batchingMaxMessages(int), and enableBatching(boolean). By default, the delay is 1 millisecond, the max messages is 1000, and batching is enabled.

Based on the code you shared, batching is enabled but is not being leveraged because synchronous sends trigger flushes.

[Shalima]

I tied sendAsync, enableBatching=true, batchingMaxMessages=5, batchingMaxPublishDelay=5 seconds. With this configuration, I sent 2 messages. My expectation was that it will have a delay in the consumer since there is not enough batchMessages. Is that a fair expectation ? I also tried sending 7 messages and I could see that the consumer behaved the same way when batching was disabled. I was hoping to see some difference if that makes sense.