[GitHub] [pulsar] merlimat commented on a change in pull request #14185: Java client: only schedule batch when there are pending messages

[GitBox <gi...@apache.org>]

merlimat commented on a change in pull request #14185:
URL: https://github.com/apache/pulsar/pull/14185#discussion_r803026555



##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
##########
@@ -1947,8 +1933,34 @@ protected void triggerFlush() {
         }
     }
 
+    // must acquire semaphore before calling
+    private void maybeTriggerBatchTimerTask() {
+        if (this.batchTimerTask == null) {
+            this.batchTimerTask = cnx().ctx().executor()
+                    .schedule(catchingAndLoggingThrowables(() -> {
+                        if (log.isTraceEnabled()) {
+                            log.trace("[{}] [{}] Batching the messages from the batch container from timer thread",
+                                    topic, producerName);
+                        }
+                        // No need to prematurely cut batches if we're not ready to send them.
+                        if (getState() != State.Ready) {
+                            return;

Review comment:
       I think we need to ensure that we schedule or trigger `batchMessageAndSend()` once the producer gets reconnected, otherwise it might not get re-triggered until new `send()` are issued. It could leave the existing incomplete batch for indefinite time.

##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
##########
@@ -1947,8 +1933,34 @@ protected void triggerFlush() {
         }
     }
 
+    // must acquire semaphore before calling
+    private void maybeTriggerBatchTimerTask() {
+        if (this.batchTimerTask == null) {
+            this.batchTimerTask = cnx().ctx().executor()
+                    .schedule(catchingAndLoggingThrowables(() -> {

Review comment:
       Also returning early could help: 
   
   
   ```java
      if (this.batchTimerTask != null) {
          return;
      }
   ```

##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
##########
@@ -1947,8 +1933,34 @@ protected void triggerFlush() {
         }
     }
 
+    // must acquire semaphore before calling
+    private void maybeTriggerBatchTimerTask() {
+        if (this.batchTimerTask == null) {
+            this.batchTimerTask = cnx().ctx().executor()
+                    .schedule(catchingAndLoggingThrowables(() -> {
+                        if (log.isTraceEnabled()) {
+                            log.trace("[{}] [{}] Batching the messages from the batch container from timer thread",
+                                    topic, producerName);
+                        }
+                        // No need to prematurely cut batches if we're not ready to send them.
+                        if (getState() != State.Ready) {
+                            return;
+                        }
+                        // semaphore acquired when message was enqueued to container
+                        synchronized (ProducerImpl.this) {
+                            batchMessageAndSend();
+                        }
+                    }), conf.getBatchingMaxPublishDelayMicros(), TimeUnit.MICROSECONDS);
+        }
+    }
+
     // must acquire semaphore before enqueuing
     private void batchMessageAndSend() {
+        if (this.batchTimerTask != null) {
+            // Cancel batch timer task since we're sending the batch now. (Don't interrupt; it could be this future.)
+            this.batchTimerTask.cancel(false);
+            this.batchTimerTask = null;
+        }

Review comment:
       Yes, there should only be 1 batch being packed at once, because we're reusing some of the containers 




-- 
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: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org