feat(transactions): Transaction added to garantee only once paradigm.… by marcosschroh · Pull Request #297 · kpn/kstreams

marcosschroh · 2025-03-06T13:30:23Z

… Related to #265

github-actions · 2025-03-06T13:31:11Z

PR Preview Action v1.6.0
🚀 View preview at https://kpn.github.io/kstreams/pr-preview/pr-297/
Built to branch `gh-pages` at 2025-03-06 13:31 UTC. Preview will be ready when the GitHub Pages deployment is complete.

woile · 2025-03-10T21:57:03Z

@@ -0,0 +1,241 @@
+`Kafka 0.11.0` includes support for `idempotent` and `transactional` capabilities in the `producer`. Idempotent delivery ensures that messages are delivered `exactly once`
+to a particular topic partition during the lifetime of a `single producer`. Transactional delivery allows producers to send data to multiple partitions such that either
+all messages are successfully delivered, or none of them are. Together, these capabilities enable `*exactly once semantics*` in Kafka.


I think transactions lack some explanation here. I would add that "exactly once semantics" are for only one scenario:

consume-process-produce under the same kafka cluster.

woile · 2025-03-10T22:03:15Z

    ) -> typing.Awaitable[RecordMetadata]: ...


+class Transaction(typing.Protocol):


Add a docstring to this class please. It's not clear why we have a protocol Transaction and then we have transaction.Transaction. An explanation on the docstring would clarify.

consider also renaming this to TransactionProtocol.

woile · 2025-03-10T22:03:45Z

+from .clients import Producer
+
 if typing.TYPE_CHECKING:
+    from . import transaction  #  pragma: no cover


is there a way to avoid this circular import? could transaction.Transaction also be a protocol?

woile · 2025-03-10T22:09:17Z

+- Transaction state is stored in a new internal topic `__transaction_state`. This topic is not created until the the first attempt to use a transactional request API. There are several settings to control the topic's configuration.
+- `Topics` which are included in transactions should be configured for durability. In particular, the `replication.factor` should be at least `3`, and the `min.insync.replicas` for these topics should be set to `2`
+- `Transactions` always add overhead, meaning that more effort is needed to produce events and the consumer have to apply filters
+For example, `transaction.state.log.min.isr` controls the minimum ISR for this topic.


For discoverability I would add links to the kafka settings.

Suggested change

For example, `transaction.state.log.min.isr` controls the minimum ISR for this topic.

For example, [`transaction.state.log.min.isr`](https://kafka.apache.org/documentation/#brokerconfigs_transaction.state.log.min.isr) controls the minimum ISR for this topic.

woile · 2025-03-10T22:13:42Z

+
+## Usage
+
+From the `kstreams` point of view, the `transaction pattern` is a `context manager` that will `start` a transaction and then `commit` or `aboort` it. Always a `transaction` starts when an event is send. Once that we have the `context` we can send events in two ways:


Suggested change

From the `kstreams` point of view, the `transaction pattern` is a `context manager` that will `start` a transaction and then `commit` or `aboort` it. Always a `transaction` starts when an event is send. Once that we have the `context` we can send events in two ways:

From the `kstreams` point of view, the "transaction pattern" is a `context manager` that will `start` a transaction and then `commit` or `abort` it. A `transaction` always starts when an event is sent. Once we have the `context` we can send events in two ways:

woile · 2025-03-10T22:14:51Z

+
+From the `kstreams` point of view, the `transaction pattern` is a `context manager` that will `start` a transaction and then `commit` or `aboort` it. Always a `transaction` starts when an event is send. Once that we have the `context` we can send events in two ways:
+
+Using the `StreamEngine` directly:


This should be a side-note, the recommended way should be with the type annotation IMO.

woile · 2025-03-10T22:15:39Z

Excellent work! 💪🏻

marcosschroh · 2025-03-11T14:29:16Z

I have hit an issue in aiokafka. Currently I am blocked by aio-libs/aiokafka#1098

Masqueey

What an insane commit! Incredible that you can manage to implement such a large feature and manage to adapt the testing client to work and test your changes accordingly.

I left some spelling corrections and some questions, but mostly very impressed with the work you did.

Masqueey · 2025-03-10T14:14:50Z

@@ -0,0 +1,241 @@
+`Kafka 0.11.0` includes support for `idempotent` and `transactional` capabilities in the `producer`. Idempotent delivery ensures that messages are delivered `exactly once`


Although we call it Kafka, formally it is called Apache Kafka. I think this first mention would be good to use the full name. Further references can stay Kafka.

Masqueey · 2025-03-10T14:17:03Z

+It is important to notice that:
+
+- `Transaction` always start from a `send` (producer)
+- Events sent to one or more topics will only be visible on consumers after the transaction is committed.


"Events or (offset) commits"

Masqueey · 2025-03-10T14:33:09Z

+- To use transactions, a `transaction id` (unique id per transaction) must be set prior an event is sent. If you do not provide one `kstreams` will auto generate one for you.
+- Transaction state is stored in a new internal topic `__transaction_state`. This topic is not created until the the first attempt to use a transactional request API. There are several settings to control the topic's configuration.
+- `Topics` which are included in transactions should be configured for durability. In particular, the `replication.factor` should be at least `3`, and the `min.insync.replicas` for these topics should be set to `2`
+- `Transactions` always add overhead, meaning that more effort is needed to produce events and the consumer have to apply filters


Not all sentences in this itemisation end in a period (.), I would change them to make sure all of them do. (Sorry for nitpicking.)

Masqueey · 2025-03-10T15:17:37Z

+- `Topics` which are included in transactions should be configured for durability. In particular, the `replication.factor` should be at least `3`, and the `min.insync.replicas` for these topics should be set to `2`
+- `Transactions` always add overhead, meaning that more effort is needed to produce events and the consumer have to apply filters
+For example, `transaction.state.log.min.isr` controls the minimum ISR for this topic.
+- `Streams` (consumers) will have to filter or not transactional events


Maybe more: "Consumer will have to set per Stream whether to filter out events based on their transaction header." (Or something like that, I might not understand the process well enough yet.)

Masqueey · 2025-03-10T15:21:18Z

+queue "Topic A" as topic_a
+
+producer -> topic_a: "produce with a transaction"
+topic_a --> stream_a: "consume only transactional events"


It's more about the success of the transaction, more than whether or not it was part of a transaction, right? So more "filter out aborted transactional events" than "only transactional"?

Masqueey · 2025-03-14T16:37:25Z

-        self._producer: typing.Optional[typing.Type[Producer]] = None
+        self._producer: typing.Optional[Producer] = None
        self._streams: typing.List[Stream] = []
+        self._transaction_manager = TransactionManager(


How come this TransactionManager no longer needs to be imported?

Masqueey · 2025-03-14T16:40:46Z

@@ -74,7 +132,10 @@ def __init__(self, group_id: Optional[str] = None, **kwargs) -> None:

        # Called to make sure that has all the kafka attributes like _coordinator


"that it has"

Masqueey · 2025-03-14T16:40:56Z

@@ -74,7 +132,10 @@ def __init__(self, group_id: Optional[str] = None, **kwargs) -> None:

        # Called to make sure that has all the kafka attributes like _coordinator
        # so it will behave like an real Kafka Consumer


"a" real Kafka

Masqueey · 2025-03-14T16:52:36Z

+
+
+@pytest.mark.asyncio
+async def test_producer_with_transaction_no_consumer(


I do not entirely get what this test is supposed to test. Not sure what this scenario is representing.

Masqueey · 2025-03-14T16:57:32Z

+            await asyncio.wait_for(stream.start(), timeout=0.1)
+
+        await stream.stop()
+


The only test I'm still missing here is the produce-then-commit transaction test. (Although I'm aware that that one doesn't work because of the bug in aiokafka.)

feat(transactions): Transaction added to garantee only once paradigm.…

b79b267

… Related to #265

marcosschroh requested a review from woile March 6, 2025 13:36

woile reviewed Mar 10, 2025

View reviewed changes

Masqueey approved these changes Mar 14, 2025

View reviewed changes

		) -> typing.Awaitable[RecordMetadata]: ...


		class Transaction(typing.Protocol):

	For example, `transaction.state.log.min.isr` controls the minimum ISR for this topic.
	For example, [`transaction.state.log.min.isr`](https://kafka.apache.org/documentation/#brokerconfigs_transaction.state.log.min.isr) controls the minimum ISR for this topic.


		## Usage

		From the `kstreams` point of view, the `transaction pattern` is a `context manager` that will `start` a transaction and then `commit` or `aboort` it. Always a `transaction` starts when an event is send. Once that we have the `context` we can send events in two ways:

	From the `kstreams` point of view, the `transaction pattern` is a `context manager` that will `start` a transaction and then `commit` or `aboort` it. Always a `transaction` starts when an event is send. Once that we have the `context` we can send events in two ways:
	From the `kstreams` point of view, the "transaction pattern" is a `context manager` that will `start` a transaction and then `commit` or `abort` it. A `transaction` always starts when an event is sent. Once we have the `context` we can send events in two ways:


		From the `kstreams` point of view, the `transaction pattern` is a `context manager` that will `start` a transaction and then `commit` or `aboort` it. Always a `transaction` starts when an event is send. Once that we have the `context` we can send events in two ways:

		Using the `StreamEngine` directly:

		@@ -0,0 +1,241 @@
		`Kafka 0.11.0` includes support for `idempotent` and `transactional` capabilities in the `producer`. Idempotent delivery ensures that messages are delivered `exactly once`

		@@ -74,7 +132,10 @@ def __init__(self, group_id: Optional[str] = None, **kwargs) -> None:

		# Called to make sure that has all the kafka attributes like _coordinator

		@@ -74,7 +132,10 @@ def __init__(self, group_id: Optional[str] = None, **kwargs) -> None:

		# Called to make sure that has all the kafka attributes like _coordinator
		# so it will behave like an real Kafka Consumer



		@pytest.mark.asyncio
		async def test_producer_with_transaction_no_consumer(

		await asyncio.wait_for(stream.start(), timeout=0.1)

		await stream.stop()

Uh oh!

Conversation

marcosschroh commented Mar 6, 2025

Uh oh!

github-actions Bot commented Mar 6, 2025

Built to branch gh-pages at 2025-03-06 13:31 UTC. Preview will be ready when the GitHub Pages deployment is complete.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

woile commented Mar 10, 2025

Uh oh!

marcosschroh commented Mar 11, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Masqueey left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Built to branch `gh-pages` at 2025-03-06 13:31 UTC.
Preview will be ready when the GitHub Pages deployment is complete.

marcosschroh commented Mar 11, 2025 •

edited

Loading