Fix IdentitySerde (apache#2762)

* Apply bytes only if the output of serializer is string

* Change IdentitySerde

* Fixed deser logic

* Changed the name of IdentitySerDe to StringSerDe to better reflect what it is

* Change doc comments

* Reverted all changes

* Made IdentitySerde mimic Json Schema

* Reverted any changes

* IdentitySerde will only work with primitive types

pulsar-client-cpp/python/pulsar/functions/serde.py

@@ -68,9 +68,19 @@ def deserialize(self, input_bytes):
       return pickle.loads(input_bytes)
 
 class IdentitySerDe(SerDe):
-  """Pickle based serializer"""
+  """Simple Serde that just conversion to string and back"""
+  def __init__(self):
+    self._types = [int, float, complex, str]
+
   def serialize(self, input):
-    return input
+    if type(input) in self._types:
+      return str(input).encode('utf-8')
+    raise TypeError
 
   def deserialize(self, input_bytes):
-    return input_bytes
+    for typ in self._types:
+      try:
+        return typ(input_bytes.decode('utf-8'))
+      except:
+        pass
+    raise TypeError

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

@@ -294,7 +294,7 @@ def process_result(self, output, msg):
       if self.producer is None:
         self.setup_producer()
       try:
-        output_bytes = bytes(self.output_serde.serialize(output))
+        output_bytes = self.output_serde.serialize(output)
       except:
         self.current_stats.nserialization_exceptions += 1
         self.total_stats.nserialization_exceptions += 1