Do not create any producer if the output type of a function is void (apache#2756)

srkukarni · web-flow · commit 18c26e07202e · 2018-10-10T06:11:00.000-07:00
* Do not create any producer if the output type of a function is void

* Do not write if the record is null

* Revert the check for null
diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java
@@ -218,6 +218,10 @@ public void open(Map<String, Object> config, SinkContext sinkContext) throws Exc
         log.info("Opening pulsar sink with config: {}", pulsarSinkConfig);
 
         Schema<T> schema = initializeSchema();
+        if (schema == null) {
+            log.info("Since output type is null, not creating any real sink");
+            return;
+        }
 
         FunctionConfig.ProcessingGuarantees processingGuarantees = this.pulsarSinkConfig.getProcessingGuarantees();
         switch (processingGuarantees) {
@@ -283,7 +287,7 @@ Schema<T> initializeSchema() throws ClassNotFoundException {
 
         if (Void.class.equals(typeArg)) {
             // return type is 'void', so there's no schema to check
-            return (Schema<T>) Schema.BYTES;
+            return null;
         }
 
         if (!StringUtils.isEmpty(pulsarSinkConfig.getSchemaType())) {
diff --git a/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/sink/PulsarSinkTest.java b/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/sink/PulsarSinkTest.java
@@ -42,14 +42,7 @@
 import lombok.Setter;
 import lombok.extern.slf4j.Slf4j;
 
-import org.apache.pulsar.client.api.Consumer;
-import org.apache.pulsar.client.api.ConsumerBuilder;
-import org.apache.pulsar.client.api.MessageId;
-import org.apache.pulsar.client.api.Producer;
-import org.apache.pulsar.client.api.ProducerBuilder;
-import org.apache.pulsar.client.api.PulsarClient;
-import org.apache.pulsar.client.api.PulsarClientException;
-import org.apache.pulsar.client.api.TypedMessageBuilder;
+import org.apache.pulsar.client.api.*;
 import org.apache.pulsar.client.impl.PulsarClientImpl;
 import org.apache.pulsar.functions.api.Record;
 import org.apache.pulsar.functions.api.SerDe;
@@ -169,7 +162,8 @@ public void testVoidOutputClasses() throws Exception {
         PulsarSink pulsarSink = new PulsarSink(getPulsarClient(), pulsarConfig, "test");
 
         try {
-            pulsarSink.initializeSchema();
+            Schema schema = pulsarSink.initializeSchema();
+            assertEquals(schema, (Schema)null);
         } catch (Exception ex) {
             ex.printStackTrace();
             assertEquals(ex, null);
