Simplify PushSource (#1898)

### Motivation

Currently implementing push source requires users implement an additional interface called setConsumer and then using that consumer in their loop. This pr just exposes a consume method from the pushSource abstract class that derived functions can make use of.

pulsar-io/core/src/main/java/org/apache/pulsar/io/core/PushSource.java

@@ -41,16 +41,6 @@ public abstract class PushSource<T> implements Source<T> {
 
     public PushSource() {
         this.queue = new LinkedBlockingQueue<>(this.getQueueLength());
-        this.setConsumer(new Consumer<Record<T>>() {
-            @Override
-            public void accept(Record<T> record) {
-                try {
-                    queue.put(record);
-                } catch (InterruptedException e) {
-                    throw new RuntimeException(e);
-                }
-            }
-        });
     }
 
     @Override
@@ -71,7 +61,13 @@ public Record<T> read() throws Exception {
      * to pass messages whenever there is data to be pushed to Pulsar.
      * @param consumer
      */
-    abstract public void setConsumer(Consumer<Record<T>> consumer);
+    public void consume(Record<T> record) {
+        try {
+            queue.put(record);
+        } catch (InterruptedException e) {
+            throw new RuntimeException(e);
+        }
+    }
 
     /**
      * Get length of the queue that records are push onto

pulsar-io/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaSource.java

@@ -48,8 +48,6 @@ public abstract class KafkaSource<V> extends PushSource<V> {
     private KafkaSourceConfig kafkaSourceConfig;
     Thread runnerThread;
 
-    private java.util.function.Consumer<Record<V>> consumeFunction;
-
     @Override
     public void open(Map<String, Object> config) throws Exception {
         kafkaSourceConfig = KafkaSourceConfig.load(config);
@@ -78,11 +76,6 @@ public void open(Map<String, Object> config) throws Exception {
 
     }
 
-    @Override
-    public void setConsumer(java.util.function.Consumer<Record<V>> consumerFunction) {
-        this.consumeFunction = consumerFunction;
-    }
-
     @Override
     public void close() throws InterruptedException {
         LOG.info("Stopping kafka source");
@@ -112,7 +105,7 @@ public void start() {
                 for (ConsumerRecord<byte[], byte[]> consumerRecord : consumerRecords) {
                     LOG.debug("Record received from kafka, key: {}. value: {}", consumerRecord.key(), consumerRecord.value());
                     KafkaRecord<V> record = new KafkaRecord<>(consumerRecord, extractValue(consumerRecord));
-                    consumeFunction.accept(record);
+                    consume(record);
                     futures[index] = record.getCompletableFuture();
                     index++;
                 }

pulsar-io/rabbitmq/src/main/java/org/apache/pulsar/io/rabbitmq/RabbitMQSource.java

@@ -41,7 +41,6 @@ public class RabbitMQSource extends PushSource<byte[]> {
 
     private static Logger logger = LoggerFactory.getLogger(RabbitMQSource.class);
 
-    private Consumer<Record<byte[]>> consumer;
     private Connection rabbitMQConnection;
     private Channel rabbitMQChannel;
     private RabbitMQConfig rabbitMQConfig;
@@ -62,33 +61,28 @@ public void open(Map<String, Object> config) throws Exception {
         );
         rabbitMQChannel = rabbitMQConnection.createChannel();
         rabbitMQChannel.queueDeclare(rabbitMQConfig.getQueueName(), false, false, false, null);
-        com.rabbitmq.client.Consumer consumer = new RabbitMQConsumer(this.consumer, rabbitMQChannel);
+        com.rabbitmq.client.Consumer consumer = new RabbitMQConsumer(this, rabbitMQChannel);
         rabbitMQChannel.basicConsume(rabbitMQConfig.getQueueName(), consumer);
         logger.info("A consumer for queue {} has been successfully started.", rabbitMQConfig.getQueueName());
     }
 
-    @Override
-    public void setConsumer(Consumer<Record<byte[]>> consumer) {
-        this.consumer = consumer;
-    }
-
     @Override
     public void close() throws Exception {
         rabbitMQChannel.close();
         rabbitMQConnection.close();
     }
 
     private class RabbitMQConsumer extends DefaultConsumer {
-        private Consumer<Record<byte[]>> consumeFunction;
+        private RabbitMQSource source;
 
-        public RabbitMQConsumer(Consumer<Record<byte[]>> consumeFunction, Channel channel) {
+        public RabbitMQConsumer(RabbitMQSource source, Channel channel) {
             super(channel);
-            this.consumeFunction = consumeFunction;
+            this.source = source;
         }
 
         @Override
         public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
-            consumeFunction.accept(new RabbitMQRecord(body));
+            source.consume(new RabbitMQRecord(body));
         }
     }
 

pulsar-io/twitter/src/main/java/org/apache/pulsar/io/twitter/TwitterFireHose.java

@@ -50,7 +50,6 @@ public class TwitterFireHose extends PushSource<String> {
 
     // ----- Runtime fields
     private Object waitObject;
-    private Consumer<Record<String>> consumeFunction;
 
     @Override
     public void open(Map<String, Object> config) throws IOException {
@@ -65,11 +64,6 @@ public void open(Map<String, Object> config) throws IOException {
         startThread(hoseConfig);
     }
 
-    @Override
-    public void setConsumer(Consumer<Record<String>> consumeFunction) {
-        this.consumeFunction = consumeFunction;
-    }
-
     @Override
     public void close() throws Exception {
         stopThread();
@@ -125,7 +119,7 @@ public boolean process() throws IOException, InterruptedException {
                             // We don't really care if the record succeeds or not.
                             // However might be in the future to count failures
                             // TODO:- Figure out the metrics story for connectors
-                            consumeFunction.accept(new TwitterRecord(line));
+                            consume(new TwitterRecord(line));
                         } catch (Exception e) {
                             LOG.error("Exception thrown");
                         }