Skip to content

Hotfix/kafka 3 #78

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
maldwg merged 3 commits into from
Aug 14, 2026
Merged

Hotfix/kafka 3 #78

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
d1afe25
Merge pull request #77 from ASTRAOS-de/dev
maldwg Aug 14, 2026
bd4f2f5
chore: release v2.3.3 [skip ci]
github-actions[bot] Aug 14, 2026
97a1ca7
Fix small kafka argument error
maldwg Aug 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion VERSION
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1 +1 @@
v2.3.3-dev
v2.3.3
9 changes: 5 additions & 4 deletions src/base/kafka/producer.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -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(
Expand All @@ -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:
Expand Down
15 changes: 14 additions & 1 deletion tests/kafka/test_exactly_once_kafka_produce_handler.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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(
Expand Down
18 changes: 9 additions & 9 deletions tests/kafka/test_kafka_eos_batch.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand All @@ -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(
Expand All @@ -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()
Expand Down Expand Up @@ -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()

Expand Down Expand Up @@ -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)

Expand All @@ -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()
Expand Down
Loading