diff --git a/pulsar-functions/instance/src/main/python/python_instance.py b/pulsar-functions/instance/src/main/python/python_instance.py index 672134c0689cc..2de6f8e40d90a 100755 --- a/pulsar-functions/instance/src/main/python/python_instance.py +++ b/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} diff --git a/pulsar-functions/instance/src/test/python/test_python_instance.py b/pulsar-functions/instance/src/test/python/test_python_instance.py index 8667f13964b04..8af90f78a775b 100644 --- a/pulsar-functions/instance/src/test/python/test_python_instance.py +++ b/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()))