[GitHub] [pulsar] srkukarni commented on a change in pull request #7573: Allow null consume in BatchPushSource

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

srkukarni commented on a change in pull request #7573:
URL: https://github.com/apache/pulsar/pull/7573#discussion_r456584747



##########
File path: pulsar-io/core/src/main/java/org/apache/pulsar/io/core/BatchPushSource.java
##########
@@ -41,17 +46,26 @@ public BatchPushSource() {
 
     @Override
     public Record<T> readNext() throws Exception {
-        return queue.take();
+        Record<T> record = queue.take();
+        if (record instanceof NullRecord) {
+            return null;
+        } else {
+            return record;
+        }
     }
 
     /**
      * Send this message to be written to Pulsar.
-     *
+     * Pass null if you you are done with this task
      * @param record next message from source which should be sent to a Pulsar topic
      */
     public void consume(Record<T> record) {
         try {
-            queue.put(record);
+            if (record != null) {
+                queue.put(record);
+            } else {
+                queue.put(new NullRecord());

Review comment:
       done




----------------------------------------------------------------
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.

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