diff --git a/VERSION b/VERSION index fac9fc0..84b787e 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -v2.3.3 +v2.3.4-dev diff --git a/src/base/kafka/producer.py b/src/base/kafka/producer.py index e298d73..edd3849 100644 --- a/src/base/kafka/producer.py +++ b/src/base/kafka/producer.py @@ -217,10 +217,11 @@ def _reset_producer(self) -> None: def _init_transactions_with_retry(self) -> None: retry_forever( lambda: self.producer.init_transactions( - timeout=kafka_config.KAFKA_TRANSACTION_API_TIMEOUT_SECONDS + kafka_config.KAFKA_TRANSACTION_API_TIMEOUT_SECONDS ), "Kafka transactional producer initialization", kafka_config.RETRY_SETTINGS, + retryable=(KafkaException, RuntimeError, OSError), ) def produce(self, topic: str, data: str, key: None | str = None) -> None: @@ -299,14 +300,14 @@ def operation(): self.producer.send_offsets_to_transaction( consumer.offsets_for(consumed_messages), consumer.group_metadata(), - timeout=kafka_config.KAFKA_TRANSACTION_API_TIMEOUT_SECONDS, + kafka_config.KAFKA_TRANSACTION_API_TIMEOUT_SECONDS, ) self.commit_transaction_with_retry() except Exception as exception: logger.info("Aborting Kafka transaction.") try: self.producer.abort_transaction( - timeout=kafka_config.KAFKA_TRANSACTION_API_TIMEOUT_SECONDS + kafka_config.KAFKA_TRANSACTION_API_TIMEOUT_SECONDS ) except Exception as abort_exception: logger.warning( @@ -330,7 +331,7 @@ def commit_transaction_with_retry( while not committed and retry_count < max_retries: try: self.producer.commit_transaction( - timeout=kafka_config.KAFKA_TRANSACTION_API_TIMEOUT_SECONDS + kafka_config.KAFKA_TRANSACTION_API_TIMEOUT_SECONDS ) committed = True except KafkaException as exception: diff --git a/tests/kafka/test_exactly_once_kafka_produce_handler.py b/tests/kafka/test_exactly_once_kafka_produce_handler.py index cd7af0a..7eebf58 100644 --- a/tests/kafka/test_exactly_once_kafka_produce_handler.py +++ b/tests/kafka/test_exactly_once_kafka_produce_handler.py @@ -51,7 +51,7 @@ def test_init(self, mock_producer, mock_uuid): mock_producer.assert_called_once_with(expected_conf) mock_producer_instance.init_transactions.assert_called_once_with( - timeout=KAFKA_TRANSACTION_API_TIMEOUT_SECONDS + KAFKA_TRANSACTION_API_TIMEOUT_SECONDS ) @patch("src.base.retry.time.sleep", return_value=None) @@ -104,6 +104,19 @@ def test_init_retries_until_transactions_initialize( self.assertEqual(2, mock_producer_instance.init_transactions.call_count) mock_sleep.assert_called() + @patch("src.base.kafka.producer.Producer") + def test_init_does_not_retry_programming_errors(self, mock_producer): + mock_producer.return_value.init_transactions.side_effect = TypeError( + "unsupported call signature" + ) + + with self.assertRaisesRegex(TypeError, "unsupported call signature"): + ExactlyOnceKafkaProduceHandler() + + mock_producer.return_value.init_transactions.assert_called_once_with( + KAFKA_TRANSACTION_API_TIMEOUT_SECONDS + ) + class TestSend(unittest.TestCase): @patch( diff --git a/tests/kafka/test_kafka_eos_batch.py b/tests/kafka/test_kafka_eos_batch.py index 47b6ea8..5f00c04 100644 --- a/tests/kafka/test_kafka_eos_batch.py +++ b/tests/kafka/test_kafka_eos_batch.py @@ -54,10 +54,10 @@ def test_outputs_and_source_offsets_are_committed_in_one_transaction( producer.send_offsets_to_transaction.assert_called_once_with( offsets, group_metadata, - timeout=kafka_config.KAFKA_TRANSACTION_API_TIMEOUT_SECONDS, + kafka_config.KAFKA_TRANSACTION_API_TIMEOUT_SECONDS, ) producer.commit_transaction.assert_called_once_with( - timeout=kafka_config.KAFKA_TRANSACTION_API_TIMEOUT_SECONDS + kafka_config.KAFKA_TRANSACTION_API_TIMEOUT_SECONDS ) @patch("src.base.kafka.producer.Producer") @@ -79,10 +79,10 @@ def test_source_offsets_are_committed_when_processing_has_no_output( producer.send_offsets_to_transaction.assert_called_once_with( consumer.offsets_for.return_value, consumer.group_metadata.return_value, - timeout=kafka_config.KAFKA_TRANSACTION_API_TIMEOUT_SECONDS, + kafka_config.KAFKA_TRANSACTION_API_TIMEOUT_SECONDS, ) producer.commit_transaction.assert_called_once_with( - timeout=kafka_config.KAFKA_TRANSACTION_API_TIMEOUT_SECONDS + kafka_config.KAFKA_TRANSACTION_API_TIMEOUT_SECONDS ) @patch( @@ -107,7 +107,7 @@ def test_stale_consumer_membership_aborts_and_recovers_without_producer_retry( handler.produce("output", "result", key="source-key") producer.abort_transaction.assert_called_once_with( - timeout=kafka_config.KAFKA_TRANSACTION_API_TIMEOUT_SECONDS + kafka_config.KAFKA_TRANSACTION_API_TIMEOUT_SECONDS ) producer.commit_transaction.assert_not_called() consumer.recover_group_membership.assert_called_once() @@ -144,7 +144,7 @@ def test_all_stale_group_errors_take_the_consumer_recovery_path( handler.produce("output", "result") producer.abort_transaction.assert_called_once_with( - timeout=kafka_config.KAFKA_TRANSACTION_API_TIMEOUT_SECONDS + kafka_config.KAFKA_TRANSACTION_API_TIMEOUT_SECONDS ) consumer.recover_group_membership.assert_called_once() @@ -173,10 +173,10 @@ def test_transaction_can_succeed_after_membership_recovery( consumer.recover_group_membership.assert_called_once() producer.abort_transaction.assert_called_once_with( - timeout=kafka_config.KAFKA_TRANSACTION_API_TIMEOUT_SECONDS + kafka_config.KAFKA_TRANSACTION_API_TIMEOUT_SECONDS ) producer.commit_transaction.assert_called_once_with( - timeout=kafka_config.KAFKA_TRANSACTION_API_TIMEOUT_SECONDS + kafka_config.KAFKA_TRANSACTION_API_TIMEOUT_SECONDS ) self.assertEqual(2, producer.begin_transaction.call_count) @@ -202,7 +202,7 @@ def test_transport_timeout_resets_producer_and_consumer_before_replay( handler.produce("output", "result") producer.abort_transaction.assert_called_once_with( - timeout=kafka_config.KAFKA_TRANSACTION_API_TIMEOUT_SECONDS + kafka_config.KAFKA_TRANSACTION_API_TIMEOUT_SECONDS ) consumer.disconnect_for_recovery.assert_called_once() consumer.reconnect_after_recovery.assert_called_once_with()