[FLINK-26378][sdk][py] Set caller to None for ingress messages.

This commit makes sure that the caller's address would be None for
messages that are coming from an ingress.

Author: igalshilman

statefun-sdk-python/statefun/request_reply_v3.py

@@ -113,7 +113,7 @@ def send_egress(self, message: EgressMessage):
 # -------------------------------------------------------------------------------------------------------------------
 
 def sdk_address_from_pb(addr: Address) -> typing.Optional[SdkAddress]:
-    if not addr:
+    if not addr or (not addr.namespace and not addr.type and not addr.id):
         return None
     return SdkAddress(namespace=addr.namespace,
                       name=addr.type,

statefun-sdk-python/tests/request_reply_test.py

@@ -229,3 +229,16 @@ def fun(context, message):
         self.assertEqual(missing_state_2_spec['state_name'], 'missing_state_2')
         self.assertEqual(missing_state_2_spec['expiration_spec']['mode'], 'AFTER_WRITE')
         self.assertEqual(missing_state_2_spec['expiration_spec']['expire_after_millis'], '2000')
+
+    def test_caller_not_set(self):
+        functions = StatefulFunctions()
+
+        @functions.bind(typename='org.foo/greeter')
+        def fun(context: Context, message: Message):
+            self.assertIsNone(context.caller)
+
+        builder = InvocationBuilder()
+        builder.with_target("org.foo", "greeter", "0")
+        builder.with_invocation("Hello", StringType)
+
+        round_trip(functions, builder)