[fix][fn] Honour consumerCryptoFailureAction in the Python function runtime

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

@@ -211,6 +211,7 @@ def run(self):
         "crypto_key_reader": crypto_key_reader
       }
       consumer_args.update(nack_args)
+      consumer_args.update(self.get_crypto_failure_action_args(consumer_conf))
       if consumer_conf.HasField("receiverQueueSize"):
         consumer_args["receiver_queue_size"] = consumer_conf.receiverQueueSize.value
 
@@ -614,3 +615,33 @@ def get_crypto_reader(self, crypto_spec):
       except Exception as e:
         Log.error("Failed to load the crypto key reader from spec: %s, error: %s" % (crypto_spec, e))
     return crypto_key_reader
+
+  def get_crypto_failure_action_args(self, consumer_conf):
+    """Build the crypto failure action argument for Client.subscribe().
+
+    Returns a dict to splat into the subscribe() call: either empty, or carrying
+    crypto_failure_action.
+
+    ConsumerSpec.cryptoSpec.consumerCryptoFailureAction tells the client what to do when an
+    input message cannot be decrypted. The Java runtime applies it in PulsarSource
+    (cb.cryptoFailureAction(...)); the Python runtime previously dropped it, so the client
+    default (FAIL) always applied and a configured DISCARD or CONSUME was silently ignored.
+
+    The argument is omitted when no cryptoSpec is configured rather than passed as None,
+    because subscribe() validates it with _check_type(ConsumerCryptoFailureAction, ...) rather
+    than _check_type_or_none, so None would fail for every function without crypto - the same
+    constraint get_negative_ack_args documents for its delay. When a cryptoSpec is present the
+    action is always forwarded: the proto3 enum default (FAIL) equals the client default, so an
+    unset action keeps today's behavior while DISCARD and CONSUME now reach the client. The
+    producer-only value (SEND) is not meaningful for a consumer and falls back to FAIL, like
+    the compression mapping treats values without a consumer equivalent.
+    """
+    if not consumer_conf.HasField("cryptoSpec"):
+      return {}
+
+    if consumer_conf.cryptoSpec.consumerCryptoFailureAction == Function_pb2.CryptoSpec.FailureAction.Value("DISCARD"):
+      return {"crypto_failure_action": pulsar.ConsumerCryptoFailureAction.DISCARD}
+    if consumer_conf.cryptoSpec.consumerCryptoFailureAction == Function_pb2.CryptoSpec.FailureAction.Value("CONSUME"):
+      return {"crypto_failure_action": pulsar.ConsumerCryptoFailureAction.CONSUME}
+
+    return {"crypto_failure_action": pulsar.ConsumerCryptoFailureAction.FAIL}

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

@@ -396,3 +396,70 @@ def test_result_is_splattable_into_subscribe_kwargs(self):
     args = self._instance(delay_ms=250).get_negative_ack_args()
     self.assertIsInstance(args, dict)
     self.assertEqual(["negative_ack_redelivery_delay_ms"], list(args.keys()))
+
+
+class TestConsumerCryptoFailureAction(unittest.TestCase):
+  """Covers ConsumerSpec.cryptoSpec.consumerCryptoFailureAction reaching the consumer.
+
+  The Java runtime applies the configured action in PulsarSource (cb.cryptoFailureAction(...));
+  the Python runtime dropped it, so the client default (FAIL) always applied and a configured
+  DISCARD or CONSUME was silently ignored.
+  """
+
+  def _instance_and_consumer_conf(self, failure_action=None, with_crypto_spec=False):
+    function_details = Function_pb2.FunctionDetails()
+    function_details.sink.topic = "test_sink_topic"
+    consumer_spec = function_details.source.inputSpecs["test_input_topic"]
+    if failure_action is not None:
+      consumer_spec.cryptoSpec.consumerCryptoFailureAction = failure_action
+    elif with_crypto_spec:
+      # a cryptoSpec present with the action left unset reads as the proto3 default (FAIL)
+      consumer_spec.cryptoSpec.cryptoKeyReaderClassName = "test.CryptoKeyReader"
+
+    instance = PythonInstance('test_instance', 'test_func', '1.0', function_details, 100, 30,
+                              'user_code', Mock(), Mock(), 'test_cluster', 'test_url', None)
+    consumer_conf = instance.instance_config.function_details.source.inputSpecs["test_input_topic"]
+    return instance, consumer_conf
+
+  def test_action_is_omitted_without_a_crypto_spec(self):
+    # subscribe() validates crypto_failure_action with _check_type rather than
+    # _check_type_or_none, so it must be omitted rather than passed as None - passing None
+    # would fail for every function that does not configure crypto at all.
+    instance, consumer_conf = self._instance_and_consumer_conf()
+    self.assertEqual({}, instance.get_crypto_failure_action_args(consumer_conf))
+
+  def test_unset_action_in_a_present_crypto_spec_keeps_the_client_default(self):
+    # proto3 enum default (FAIL) equals the client default, so a cryptoSpec that does not set
+    # the action leaves the pre-change behavior in place.
+    instance, consumer_conf = self._instance_and_consumer_conf(with_crypto_spec=True)
+    args = instance.get_crypto_failure_action_args(consumer_conf)
+    self.assertEqual(pulsar.ConsumerCryptoFailureAction.FAIL, args["crypto_failure_action"])
+
+  def test_discard_is_forwarded(self):
+    instance, consumer_conf = self._instance_and_consumer_conf(
+        failure_action=Function_pb2.CryptoSpec.FailureAction.Value("DISCARD"))
+    args = instance.get_crypto_failure_action_args(consumer_conf)
+    self.assertEqual(pulsar.ConsumerCryptoFailureAction.DISCARD, args["crypto_failure_action"])
+
+  def test_consume_is_forwarded(self):
+    instance, consumer_conf = self._instance_and_consumer_conf(
+        failure_action=Function_pb2.CryptoSpec.FailureAction.Value("CONSUME"))
+    args = instance.get_crypto_failure_action_args(consumer_conf)
+    self.assertEqual(pulsar.ConsumerCryptoFailureAction.CONSUME, args["crypto_failure_action"])
+
+  def test_producer_only_send_falls_back_to_fail(self):
+    # SEND only exists for producers; a consumer cannot honor it, so fall back to the client
+    # default rather than crash the instance at startup.
+    instance, consumer_conf = self._instance_and_consumer_conf(
+        failure_action=Function_pb2.CryptoSpec.FailureAction.Value("SEND"))
+    args = instance.get_crypto_failure_action_args(consumer_conf)
+    self.assertEqual(pulsar.ConsumerCryptoFailureAction.FAIL, args["crypto_failure_action"])
+
+  def test_result_is_splattable_into_subscribe_kwargs(self):
+    # The value is consumed via consumer_args.update(...), so it must be a dict with exactly the
+    # keyword subscribe() expects.
+    instance, consumer_conf = self._instance_and_consumer_conf(
+        failure_action=Function_pb2.CryptoSpec.FailureAction.Value("DISCARD"))
+    args = instance.get_crypto_failure_action_args(consumer_conf)
+    self.assertIsInstance(args, dict)
+    self.assertEqual(["crypto_failure_action"], list(args.keys()))