[fix][fn] Forward source message properties in Python runtime

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

@@ -337,7 +337,10 @@ def process_result(self, output, msg):
         output_object = self.output_serde.serialize(output)
 
       if output_object is not None:
-        props = {"__pfn_input_topic__" : str(msg.topic), "__pfn_input_msg_id__" : base64ify(msg.message.message_id().serialize())}
+        props = {}
+        if self.instance_config.function_details.sink.forwardSourceMessageProperty:
+          props = msg.message.properties()
+        props.update({"__pfn_input_topic__" : str(msg.topic), "__pfn_input_msg_id__" : base64ify(msg.message.message_id().serialize())})
         if self.effectively_once:
           self.producer.send_async(output_object,
                                    partial(self.done_producing, msg.consumer, msg.message, self.producer.topic()),

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

@@ -105,8 +105,8 @@ def __init__(self, t, hFunction, name="timer-thread"):
         self.t = t
         self.hFunction = hFunction
         self.thread = Timer(self.t, self.handle_function)
-        self.thread.setName(name)
-        self.thread.setDaemon(True)
+        self.thread.name = name
+        self.thread.daemon = True
 
     def handle_function(self):
         self.hFunction()

pulsar-functions/instance/src/scripts/run_python_instance_tests.sh

@@ -21,11 +21,11 @@
 
 # Make sure dependencies are installed
 pip3 install mock --user
-pip3 install protobuf==3.20.1 --user
+pip3 install protobuf==6.31.1 --user
 pip3 install fastavro --user
 
 CUR_DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" >/dev/null && pwd )"
 PULSAR_HOME="$( cd "$CUR_DIR/../../../../" >/dev/null && pwd )"
 
 # run instance tests
-PULSAR_HOME=${PULSAR_HOME} PYTHONPATH=${PULSAR_HOME}/pulsar-functions/instance/target/python-instance python3 -m unittest discover -v ${PULSAR_HOME}/pulsar-functions/instance/target/python-instance/tests
+PULSAR_HOME=${PULSAR_HOME} PYTHONPATH=${PULSAR_HOME}/pulsar-functions/instance/src/main/python python3 -m unittest discover -v -s ${PULSAR_HOME}/pulsar-functions/instance/src/test/python

pulsar-functions/instance/src/test/python/test_python_instance.py

@@ -23,6 +23,12 @@
 from mock import Mock
 import sys
 sys.modules['prometheus_client'] = Mock()
+sys.modules['bookkeeper'] = Mock()
+sys.modules['bookkeeper.types'] = Mock()
+sys.modules['bookkeeper.common'] = Mock()
+sys.modules['bookkeeper.common.exceptions'] = Mock()
+sys.modules['bookkeeper.proto'] = Mock()
+sys.modules['bookkeeper.proto.stream_pb2'] = Mock()
 
 from contextimpl import ContextImpl
 from python_instance import PythonInstance, InstanceConfig
@@ -42,7 +48,8 @@ def __eq__(self, other):
     return Any()
 
   def setUp(self):
-    log.init_logger("INFO", "foo", os.environ.get("PULSAR_HOME") + "/conf/functions-logging/console_logging_config.ini")
+    if not hasattr(sys.stdout, 'logger'):
+      log.init_logger("INFO", "foo", os.environ.get("PULSAR_HOME") + "/conf/functions-logging/console_logging_config.ini")
 
   def test_context_publish(self):
     instance_id = 'test_instance_id'
@@ -90,3 +97,55 @@ def test_context_ack_partitionedtopic(self):
 
     args, kwargs = consumer.acknowledge.call_args
     self.assertEqual(args[0], "test_message_id")
+
+class TestPropertiesForwarding(unittest.TestCase):
+
+  def _setup_mock_instance(self, forward_property):
+    function_details = Function_pb2.FunctionDetails()
+    function_details.sink.topic = "test_sink_topic"
+    function_details.sink.forwardSourceMessageProperty = forward_property
+
+    mock_pulsar_client = Mock()
+    mock_producer = Mock()
+    mock_pulsar_client.create_producer.return_value = mock_producer
+
+    instance = PythonInstance('test_instance', 'test_func', '1.0', function_details, 100, 30, 'user_code', mock_pulsar_client, Mock(), 'test_cluster', 'test_url', None)
+    instance.producer = mock_producer
+    instance.contextimpl = Mock()
+    instance.contextimpl.get_message_partition_index.return_value = None
+    instance.output_schema = "DEFAULT_SCHEMA"
+    instance.output_serde = Mock()
+    instance.output_serde.serialize.return_value = b'serialized_output'
+    instance.effectively_once = False
+
+    return instance, mock_producer
+
+  def test_forwards_properties(self):
+    instance, mock_producer = self._setup_mock_instance(forward_property=True)
+
+    mock_msg = Mock()
+    mock_msg.topic = "source-topic"
+    mock_msg.message.message_id().serialize.return_value = b'msg-id'
+    mock_msg.message.properties.return_value = {"custom-key": "custom-value"}
+
+    instance.process_result("output-data", mock_msg)
+
+    args, kwargs = mock_producer.send_async.call_args
+    self.assertIn("custom-key", kwargs['properties'])
+    self.assertEqual(kwargs['properties']["custom-key"], "custom-value")
+    self.assertIn("__pfn_input_topic__", kwargs['properties'])
+
+  def test_do_not_forward_properties(self):
+    instance, mock_producer = self._setup_mock_instance(forward_property=False)
+
+    mock_msg = Mock()
+    mock_msg.topic = "source-topic"
+    mock_msg.message.message_id().serialize.return_value = b'msg-id'
+    mock_msg.message.properties.return_value = {"custom-key": "custom-value"}
+
+    instance.process_result("output-data", mock_msg)
+
+    args, kwargs = mock_producer.send_async.call_args
+    self.assertNotIn("custom-key", kwargs['properties'])
+    self.assertIn("__pfn_input_topic__", kwargs['properties'])
+