RobertIndie commented on code in PR #23196:
URL: https://github.com/apache/pulsar/pull/23196#discussion_r1723062818
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProduceWithMessageIdTest.java:
##########
@@ -115,4 +125,70 @@ public CompletableFuture<MessageId> getFuture() {
// the result is true only if broker received right message id.
Awaitility.await().untilTrue(result);
}
+
+ @Test
+ public void sendWithCallBack() throws Exception {
+
+ int batchSize = 10;
+
+ String topic = "persistent://public/default/testSendWithCallBack";
+ ProducerImpl<byte[]> producer =
+ (ProducerImpl<byte[]>) pulsarClient.newProducer().topic(topic)
+ .enableBatching(true)
+ .batchingMaxMessages(batchSize)
+ .create();
+
+ AtomicBoolean result = new AtomicBoolean(false);
+ AtomicReference<OpSendMsgStats> sendMsgStats = new AtomicReference<>();
+ SendCallback sendComplete = new SendCallback() {
+ @Override
+ public void sendComplete(Throwable e, OpSendMsgStats
opSendMsgStats) {
+ log.info("sendComplete", e);
+ result.set(e == null);
+ sendMsgStats.set(opSendMsgStats);
+ }
+
+ @Override
+ public void addCallback(MessageImpl<?> msg, SendCallback scb) {
+
+ }
+
+ @Override
+ public SendCallback getNextSendCallback() {
+ return null;
+ }
+
+ @Override
+ public MessageImpl<?> getNextMessage() {
+ return null;
+ }
+
+ @Override
+ public CompletableFuture<MessageId> getFuture() {
+ return null;
+ }
+ };
+ int totalReadabled = 0;
+ int totalUncompressedSize = 0;
+ for (int i = 0; i < batchSize; i++) {
+ MessageMetadata metadata = new MessageMetadata();
+ ByteBuffer buffer =
ByteBuffer.wrap("data".getBytes(StandardCharsets.UTF_8));
+ MessageImpl<byte[]> msg = MessageImpl.create(metadata, buffer,
Schema.BYTES, topic);
+ msg.getDataBuffer().retain();
+ totalReadabled += msg.getDataBuffer().readableBytes();
+ totalUncompressedSize += msg.getUncompressedSize();
+ producer.sendAsync(msg, sendComplete);
+ }
+
+ Awaitility.await().untilTrue(result);
Review Comment:
We can use CountDownLatch watch here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
