Python Instance should resort to default serializer if nothing is spe…

…cified (apache#180)

pulsar-functions/runtime/src/main/python/python_instance.py

@@ -59,6 +59,7 @@
 # This is the message that the consumers put on the queue for the function thread to process
 InternalMessage = namedtuple('InternalMessage', 'message topic serde consumer')
 InternalQuitMessage = namedtuple('InternalQuitMessage', 'quit')
+DEFAULT_SERIALIZER = "pulsarfunction.serde.IdentitySerDe"
 
 # We keep track of the following metrics
 class Stats(object):
@@ -128,7 +129,8 @@ def run(self):
       )
 
     for topic in self.instance_config.function_config.inputs:
-      serde_kclass = util.import_class(os.path.dirname(self.user_code), "pulsarfunction.IdentitySerDe", try_internal=True)
+      global DEFAULT_SERIALIZER
+      serde_kclass = util.import_class(os.path.dirname(self.user_code), DEFAULT_SERIALIZER, try_internal=True)
       self.input_serdes[topic] = serde_kclass()
       Log.info("Setting up consumer for topic %s with subname %s" % (topic, subscription_name))
       self.consumers[topic] = self.pulsar_client.subscribe(
@@ -137,11 +139,7 @@ def run(self):
         message_listener=partial(self.message_listener, topic, self.input_serdes[topic])
       )
 
-    # See if we need to setup output producers/output serializers
-    if self.instance_config.function_config.outputSerdeClassName != None and \
-      len(self.instance_config.function_config.outputSerdeClassName) > 0:
-      serde_kclass = util.import_class(os.path.dirname(self.user_code), self.instance_config.function_config.outputSerdeClassName, try_internal=True)
-      self.output_serde = serde_kclass()
+    # See if we need to setup output producers
     function_kclass = util.import_class(os.path.dirname(self.user_code), self.instance_config.function_config.className, try_internal=True)
     self.function_class = function_kclass()
     if self.instance_config.function_config.sinkTopic != None and \
@@ -199,6 +197,8 @@ def done_producing(self, consumer, orig_message, result, sent_message):
   def process_result(self, output, msg):
     if output is not None and self.producer is not None:
       output_bytes = None
+      if self.output_serde is None:
+        self.setup_output_serde()
       try:
         output_bytes = self.output_serde.serialize(output)
       except:
@@ -213,6 +213,16 @@ def process_result(self, output, msg):
     elif self.auto_ack and self.atleast_once:
       msg.consumer.acknowledge(msg.message)
 
+  def setup_output_serde(self):
+    if self.instance_config.function_config.outputSerdeClassName != None and \
+            len(self.instance_config.function_config.outputSerdeClassName) > 0:
+      serde_kclass = util.import_class(os.path.dirname(self.user_code), self.instance_config.function_config.outputSerdeClassName, try_internal=True)
+      self.output_serde = serde_kclass()
+    else:
+      global DEFAULT_SERIALIZER
+      serde_kclass = util.import_class(os.path.dirname(self.user_code), DEFAULT_SERIALIZER, try_internal=True)
+      self.output_serde = serde_kclass()
+
   def message_listener(self, topic, serde, consumer, message):
     item = InternalMessage(message, topic, serde, consumer)
     self.queue.put(item, True)