[functions][python] support both python 2 and 3 (#2742)

*Motivation*

base64 is non compatible between python2 and 3

*Changes*

Provides a python 2/3 compatible method

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

@@ -52,6 +52,20 @@ InternalMessage = namedtuple('InternalMessage', 'message topic serde consumer')
 InternalQuitMessage = namedtuple('InternalQuitMessage', 'quit')
 DEFAULT_SERIALIZER = "serde.IdentitySerDe"
 
+PY3 = sys.version_info[0] >= 3
+
+def base64ify(bytes_or_str):
+    if PY3 and isinstance(bytes_or_str, str):
+        input_bytes = bytes_or_str.encode('utf8')
+    else:
+        input_bytes = bytes_or_str
+
+    output_bytes = base64.urlsafe_b64encode(input_bytes)
+    if PY3:
+        return output_bytes.decode('ascii')
+    else:
+        return output_bytes
+
 # We keep track of the following metrics
 class Stats(object):
   def __init__(self):
@@ -284,7 +298,7 @@ class PythonInstance(object):
         self.current_stats.nserialization_exceptions += 1
         self.total_stats.nserialization_exceptions += 1
       if output_bytes is not None:
-        props = {"__pfn_input_topic__" : str(msg.topic), "__pfn_input_msg_id__" : base64.b64encode(msg.message.message_id().serialize())}
+        props = {"__pfn_input_topic__" : str(msg.topic), "__pfn_input_msg_id__" : base64ify(msg.message.message_id().serialize())}
         try:
           self.producer.send_async(output_bytes, partial(self.done_producing, msg.consumer, msg.message), properties=props)
         except Exception as e: