You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "AndrewJSchofield (via GitHub)" <gi...@apache.org> on 2024/10/10 15:28:29 UTC
[PR] KAFKA-17671: Create better documentation for transactions [kafka]
AndrewJSchofield opened a new pull request, #17454:
URL: https://github.com/apache/kafka/pull/17454
https://issues.apache.org/jira/browse/KAFKA-17671
This PR adds better documentation for using transactions using the Apache Kafka producer and consumer clients.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17671: Create better documentation for transactions [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #17454:
URL: https://github.com/apache/kafka/pull/17454#discussion_r1795961052
##########
docs/design.html:
##########
@@ -160,7 +160,7 @@ <h4 class="anchor-heading"><a id="design_asyncsend" class="anchor-link"></a><a h
to accumulate no more than a fixed number of messages and to wait no longer than some fixed latency bound (say 64k or 10 ms). This allows the accumulation of more bytes to send, and few larger I/O operations on the
servers. This buffering is configurable and gives a mechanism to trade off a small amount of additional latency for better throughput.
<p>
- Details on <a href="#producerconfigs">configuration</a> and the <a href="https://kafka.apache.org/{{version}}/javadoc/org/apache/kafka/clients/producer/KafkaProducer.html">api</a> for the producer can be found
+ Details on <a href="#producerconfigs">configuration</a> and the <a href="https://kafka.apache.org/082/javadoc/index.html?org/apache/kafka/clients/producer/KafkaProducer.html">api</a> for the producer can be found
Review Comment:
why did we make this change? I don't think we want to link to the 0.8.2 docs (they also seem to be broken)
We want the latest ones right?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17671: Create better documentation for transactions [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #17454:
URL: https://github.com/apache/kafka/pull/17454#discussion_r1907665770
##########
docs/design.html:
##########
@@ -290,24 +290,56 @@ <h3 class="anchor-heading"><a id="semantics" class="anchor-link"></a><a href="#s
messages have a primary key and so the updates are idempotent (receiving the same message twice just overwrites a record with another copy of itself).
</ol>
<p>
- So what about exactly once semantics (i.e. the thing you actually want)? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>
- application), we can leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in a topic, so we can write the offset to Kafka in the
- same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's position will revert to its old value and the produced data on the output topics will not be visible
- to other consumers, depending on their "isolation level." In the default "read_uncommitted" isolation level, all messages are visible to consumers even if they were part of an aborted transaction,
- but in "read_committed," the consumer will only return messages from transactions which were committed (and any messages which were not part of a transaction).
+ So what about exactly-once semantics? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a> application), we can
+ leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in an internal topic, so we can write the offset to Kafka in the
+ same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's stored position will revert to its old value and the produced data on the output topics will not
+ be visible to other consumers, depending on their "isolation level." In the default "read_uncommitted" isolation level, all messages are visible to consumers even if they were part of an aborted transaction,
+ but in "read_committed" isolation level, the consumer will only return messages from transactions which were committed (and any messages which were not part of a transaction).
<p>
When writing to an external system, the limitation is in the need to coordinate the consumer's position with what is actually stored as output. The classic way of achieving this would be to introduce a two-phase
- commit between the storage of the consumer position and the storage of the consumers output. But this can be handled more simply and generally by letting the consumer store its offset in the same place as
+ commit between the storage of the consumer position and the storage of the consumers output. This can be handled more simply and generally by letting the consumer store its offset in the same place as
its output. This is better because many of the output systems a consumer might want to write to will not support a two-phase commit. As an example of this, consider a
<a href="https://kafka.apache.org/documentation/#connect">Kafka Connect</a> connector which populates data in HDFS along with the offsets of the data it reads so that it is guaranteed that either data and
offsets are both updated or neither is. We follow similar patterns for many other data systems which require these stronger semantics and for which the messages do not have a primary key to allow for deduplication.
<p>
- So effectively Kafka supports exactly-once delivery in <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>, and the transactional producer/consumer can be used generally to provide
+ As a result, Kafka supports exactly-once delivery in <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>, and the transactional producer/consumer can be used generally to provide
exactly-once delivery when transferring and processing data between Kafka topics. Exactly-once delivery for other destination systems generally requires cooperation with such systems, but Kafka provides the
offset which makes implementing this feasible (see also <a href="https://kafka.apache.org/documentation/#connect">Kafka Connect</a>). Otherwise, Kafka guarantees at-least-once delivery by default, and allows
the user to implement at-most-once delivery by disabling retries on the producer and committing offsets in the consumer prior to processing a batch of messages.
- <h3 class="anchor-heading"><a id="replication" class="anchor-link"></a><a href="#replication">4.7 Replication</a></h3>
+ <h3 class="anchor-heading"><a id="usingtransactions" class="anchor-link"></a><a href="#usingtransactions">4.7 Using Transactions</a></h3>
+ <p>
+ As mentioned above, the simplest way to get exactly-once semantics from Kafka is to use <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>. However, it is also possible to achieve
+ the same transactional guarantees using the Kafka producer and consumer directly by using them in the same way as Kafka Streams does.
+ <p>
+ Kafka transactions are a bit different than transactions in other messaging systems. In Kafka, the consumer and producer are separate and it is only the producer which is transactional. It is however able to
+ make transactional updates to the consumer's position (confusingly called the "committed offset"), and it is this which gives the overall exactly-once behavior.
+ <p>
+ There are three key aspects to exactly-once processing using the producer and consumer, which match how Kafka Streams works.
+ <ol>
+ <li>The consumer uses partition assignment to ensure that it is the only consumer in the consumer group currently processing each partition.</li>
+ <li>The consumer uses read-committed isolation level to ensure that it does not consume records produced by transactions which aborted.</li>
+ <li>The producer uses transactions so that all of the records it produces, and any offsets it updates on behalf of the consumer, are performed atomically.</li>
+ </ol>
+ <p>
+ The consumer configuration must include <code>isolation.level=read_committed</code> and <code>enable.auto.commit=false</code>. The producer configuration must set <code>transactional.id</code>
+ to the name of the transactional ID to be used, which configures the producer for transactional delivery and also makes sure that a restarted application causes any in-flight transaction from
+ the previous instance to abort. Only the producer has the <code>transactional.id</code> configuration.
+ <p>
+ Here's an example of a <a href="https://github.com/apache/kafka/blob/trunk/tools/src/main/java/org/apache/kafka/tools/TransactionalMessageCopier.java">transactional message copier</a>
+ which uses these principles. It uses a <code>KafkaConsumer</code> to consume records from one topic and a <code>KafkaProducer</code> to produce records to another topic. It uses transactions to ensure
+ that there is no duplication or loss of records as they are copied.
+ <p>
+ It is important to handle exceptions and aborted transactions correctly. Any records written by the transational producer will be marked as being part of the transactions, and then when the
+ transaction commits or aborts, transaction marker records are written to indicate the outcome of the transaction. This is how the read-committed consumer does not see records from aborted
+ transactions. However, in the event of a transaction abort, the application's in-memory state and in particular the current position of the consumer must be reset explicitly so that it can
+ reprocess the records processed by the aborted transaction.
+ <p>
+ A simple policy for handling exceptions and aborted transactions is to discard and recreate the Kafka producer and consumer objects and start afresh. As part of recreating the consumer, the consumer
Review Comment:
KIP is now approved, but not yet implemented. :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17671: Create better documentation for transactions [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #17454:
URL: https://github.com/apache/kafka/pull/17454#discussion_r1810885718
##########
docs/design.html:
##########
@@ -254,32 +254,32 @@ <h3 class="anchor-heading"><a id="semantics" class="anchor-link"></a><a href="#s
<i>At least once</i>—Messages are never lost but may be redelivered.
</li>
<li>
- <i>Exactly once</i>—this is what people actually want, each message is delivered once and only once.
+ <i>Exactly once</i>—Each message is delivered once and only once.
Review Comment:
I will try for "processed" then, but really I think of it as delivered to the application exactly once.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17671: Create better documentation for transactions [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield merged PR #17454:
URL: https://github.com/apache/kafka/pull/17454
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17671: Create better documentation for transactions [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #17454:
URL: https://github.com/apache/kafka/pull/17454#discussion_r1795963676
##########
docs/design.html:
##########
@@ -290,24 +290,56 @@ <h3 class="anchor-heading"><a id="semantics" class="anchor-link"></a><a href="#s
messages have a primary key and so the updates are idempotent (receiving the same message twice just overwrites a record with another copy of itself).
</ol>
<p>
- So what about exactly once semantics (i.e. the thing you actually want)? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>
- application), we can leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in a topic, so we can write the offset to Kafka in the
- same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's position will revert to its old value and the produced data on the output topics will not be visible
- to other consumers, depending on their "isolation level." In the default "read_uncommitted" isolation level, all messages are visible to consumers even if they were part of an aborted transaction,
- but in "read_committed," the consumer will only return messages from transactions which were committed (and any messages which were not part of a transaction).
+ So what about exactly-once semantics? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a> application), we can
+ leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in an internal topic, so we can write the offset to Kafka in the
+ same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's stored position will revert to its old value and the produced data on the output topics will not
Review Comment:
Did we want to change this line because it is misleading -- the stored positition doesn't revert unless the offsets are fetched again.
##########
docs/design.html:
##########
@@ -290,24 +290,56 @@ <h3 class="anchor-heading"><a id="semantics" class="anchor-link"></a><a href="#s
messages have a primary key and so the updates are idempotent (receiving the same message twice just overwrites a record with another copy of itself).
</ol>
<p>
- So what about exactly once semantics (i.e. the thing you actually want)? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>
- application), we can leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in a topic, so we can write the offset to Kafka in the
- same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's position will revert to its old value and the produced data on the output topics will not be visible
- to other consumers, depending on their "isolation level." In the default "read_uncommitted" isolation level, all messages are visible to consumers even if they were part of an aborted transaction,
- but in "read_committed," the consumer will only return messages from transactions which were committed (and any messages which were not part of a transaction).
+ So what about exactly-once semantics? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a> application), we can
+ leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in an internal topic, so we can write the offset to Kafka in the
+ same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's stored position will revert to its old value and the produced data on the output topics will not
Review Comment:
Did we want to change this line because it is misleading -- the stored position doesn't revert unless the offsets are fetched again.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17671: Create better documentation for transactions [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #17454:
URL: https://github.com/apache/kafka/pull/17454#discussion_r1907876806
##########
docs/design.html:
##########
@@ -290,24 +290,58 @@ <h3 class="anchor-heading"><a id="semantics" class="anchor-link"></a><a href="#s
messages have a primary key and so the updates are idempotent (receiving the same message twice just overwrites a record with another copy of itself).
</ol>
<p>
- So what about exactly once semantics (i.e. the thing you actually want)? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>
- application), we can leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in a topic, so we can write the offset to Kafka in the
- same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's position will revert to its old value and the produced data on the output topics will not be visible
- to other consumers, depending on their "isolation level." In the default "read_uncommitted" isolation level, all messages are visible to consumers even if they were part of an aborted transaction,
- but in "read_committed," the consumer will only return messages from transactions which were committed (and any messages which were not part of a transaction).
+ So what about exactly-once semantics? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a> application), we can
+ leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in an internal topic, so we can write the offset to Kafka in the
+ same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's stored position will revert to its old value (although the consumer has to refetch the
+ committed offset because it does not automatically rewind) and the produced data on the output topics will not be visible to other consumers, depending on their "isolation level". In the default
+ "read_uncommitted" isolation level, all messages are visible to consumers even if they were part of an aborted transaction, but in "read_committed" isolation level, the consumer will only return messages
+ from transactions which were committed (and any messages which were not part of a transaction).
<p>
When writing to an external system, the limitation is in the need to coordinate the consumer's position with what is actually stored as output. The classic way of achieving this would be to introduce a two-phase
- commit between the storage of the consumer position and the storage of the consumers output. But this can be handled more simply and generally by letting the consumer store its offset in the same place as
+ commit between the storage of the consumer position and the storage of the consumers output. This can be handled more simply and generally by letting the consumer store its offset in the same place as
its output. This is better because many of the output systems a consumer might want to write to will not support a two-phase commit. As an example of this, consider a
<a href="https://kafka.apache.org/documentation/#connect">Kafka Connect</a> connector which populates data in HDFS along with the offsets of the data it reads so that it is guaranteed that either data and
offsets are both updated or neither is. We follow similar patterns for many other data systems which require these stronger semantics and for which the messages do not have a primary key to allow for deduplication.
<p>
- So effectively Kafka supports exactly-once delivery in <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>, and the transactional producer/consumer can be used generally to provide
+ As a result, Kafka supports exactly-once delivery in <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>, and the transactional producer/consumer can be used generally to provide
exactly-once delivery when transferring and processing data between Kafka topics. Exactly-once delivery for other destination systems generally requires cooperation with such systems, but Kafka provides the
Review Comment:
I've reworded that part.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17671: Create better documentation for transactions [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #17454:
URL: https://github.com/apache/kafka/pull/17454#discussion_r1795970930
##########
docs/design.html:
##########
@@ -290,24 +290,56 @@ <h3 class="anchor-heading"><a id="semantics" class="anchor-link"></a><a href="#s
messages have a primary key and so the updates are idempotent (receiving the same message twice just overwrites a record with another copy of itself).
</ol>
<p>
- So what about exactly once semantics (i.e. the thing you actually want)? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>
- application), we can leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in a topic, so we can write the offset to Kafka in the
- same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's position will revert to its old value and the produced data on the output topics will not be visible
- to other consumers, depending on their "isolation level." In the default "read_uncommitted" isolation level, all messages are visible to consumers even if they were part of an aborted transaction,
- but in "read_committed," the consumer will only return messages from transactions which were committed (and any messages which were not part of a transaction).
+ So what about exactly-once semantics? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a> application), we can
+ leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in an internal topic, so we can write the offset to Kafka in the
+ same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's stored position will revert to its old value and the produced data on the output topics will not
+ be visible to other consumers, depending on their "isolation level." In the default "read_uncommitted" isolation level, all messages are visible to consumers even if they were part of an aborted transaction,
+ but in "read_committed" isolation level, the consumer will only return messages from transactions which were committed (and any messages which were not part of a transaction).
<p>
When writing to an external system, the limitation is in the need to coordinate the consumer's position with what is actually stored as output. The classic way of achieving this would be to introduce a two-phase
- commit between the storage of the consumer position and the storage of the consumers output. But this can be handled more simply and generally by letting the consumer store its offset in the same place as
+ commit between the storage of the consumer position and the storage of the consumers output. This can be handled more simply and generally by letting the consumer store its offset in the same place as
its output. This is better because many of the output systems a consumer might want to write to will not support a two-phase commit. As an example of this, consider a
<a href="https://kafka.apache.org/documentation/#connect">Kafka Connect</a> connector which populates data in HDFS along with the offsets of the data it reads so that it is guaranteed that either data and
offsets are both updated or neither is. We follow similar patterns for many other data systems which require these stronger semantics and for which the messages do not have a primary key to allow for deduplication.
<p>
- So effectively Kafka supports exactly-once delivery in <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>, and the transactional producer/consumer can be used generally to provide
+ As a result, Kafka supports exactly-once delivery in <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>, and the transactional producer/consumer can be used generally to provide
exactly-once delivery when transferring and processing data between Kafka topics. Exactly-once delivery for other destination systems generally requires cooperation with such systems, but Kafka provides the
offset which makes implementing this feasible (see also <a href="https://kafka.apache.org/documentation/#connect">Kafka Connect</a>). Otherwise, Kafka guarantees at-least-once delivery by default, and allows
the user to implement at-most-once delivery by disabling retries on the producer and committing offsets in the consumer prior to processing a batch of messages.
- <h3 class="anchor-heading"><a id="replication" class="anchor-link"></a><a href="#replication">4.7 Replication</a></h3>
+ <h3 class="anchor-heading"><a id="usingtransactions" class="anchor-link"></a><a href="#usingtransactions">4.7 Using Transactions</a></h3>
+ <p>
+ As mentioned above, the simplest way to get exactly-once semantics from Kafka is to use <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>. However, it is also possible to achieve
+ the same transactional guarantees using the Kafka producer and consumer directly by using them in the same way as Kafka Streams does.
+ <p>
+ Kafka transactions are a bit different than transactions in other messaging systems. In Kafka, the consumer and producer are separate and it is only the producer which is transactional. It is however able to
+ make transactional updates to the consumer's position (confusingly called the "committed offset"), and it is this which gives the overall exactly-once behavior.
+ <p>
+ There are three key aspects to exactly-once processing using the producer and consumer, which match how Kafka Streams works.
+ <ol>
+ <li>The consumer uses partition assignment to ensure that it is the only consumer in the consumer group currently processing each partition.</li>
+ <li>The consumer uses read-committed isolation level to ensure that it does not consume records produced by transactions which aborted.</li>
+ <li>The producer uses transactions so that all of the records it produces, and any offsets it updates on behalf of the consumer, are performed atomically.</li>
+ </ol>
+ <p>
+ The consumer configuration must include <code>isolation.level=read_committed</code> and <code>enable.auto.commit=false</code>. The producer configuration must set <code>transactional.id</code>
+ to the name of the transactional ID to be used, which configures the producer for transactional delivery and also makes sure that a restarted application causes any in-flight transaction from
+ the previous instance to abort. Only the producer has the <code>transactional.id</code> configuration.
+ <p>
+ Here's an example of a <a href="https://github.com/apache/kafka/blob/trunk/tools/src/main/java/org/apache/kafka/tools/TransactionalMessageCopier.java">transactional message copier</a>
+ which uses these principles. It uses a <code>KafkaConsumer</code> to consume records from one topic and a <code>KafkaProducer</code> to produce records to another topic. It uses transactions to ensure
+ that there is no duplication or loss of records as they are copied.
+ <p>
+ It is important to handle exceptions and aborted transactions correctly. Any records written by the transational producer will be marked as being part of the transactions, and then when the
+ transaction commits or aborts, transaction marker records are written to indicate the outcome of the transaction. This is how the read-committed consumer does not see records from aborted
+ transactions. However, in the event of a transaction abort, the application's in-memory state and in particular the current position of the consumer must be reset explicitly so that it can
+ reprocess the records processed by the aborted transaction.
+ <p>
+ A simple policy for handling exceptions and aborted transactions is to discard and recreate the Kafka producer and consumer objects and start afresh. As part of recreating the consumer, the consumer
Review Comment:
I'm not sure we want to always encourage this. There are cases where we can safely abort without discarding the producer. KIP-1050 will help make the cases clearer.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17671: Create better documentation for transactions [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #17454:
URL: https://github.com/apache/kafka/pull/17454#discussion_r1811157856
##########
docs/design.html:
##########
@@ -254,32 +254,32 @@ <h3 class="anchor-heading"><a id="semantics" class="anchor-link"></a><a href="#s
<i>At least once</i>—Messages are never lost but may be redelivered.
</li>
<li>
- <i>Exactly once</i>—this is what people actually want, each message is delivered once and only once.
+ <i>Exactly once</i>—Each message is delivered once and only once.
</li>
</ul>
It's worth noting that this breaks down into two problems: the durability guarantees for publishing a message and the guarantees when consuming a message.
<p>
- Many systems claim to provide "exactly once" delivery semantics, but it is important to read the fine print, most of these claims are misleading (i.e. they don't translate to the case where consumers or producers
+ Many systems claim to provide "exactly-once" delivery semantics, but it is important to read the fine print, because sometimes these claims are misleading (i.e. they don't translate to the case where consumers or producers
can fail, cases where there are multiple consumer processes, or cases where data written to disk can be lost).
<p>
- Kafka's semantics are straight-forward. When publishing a message we have a notion of the message being "committed" to the log. Once a published message is committed it will not be lost as long as one broker that
- replicates the partition to which this message was written remains "alive". The definition of committed message, alive partition as well as a description of which types of failures we attempt to handle will be
+ Kafka's semantics are straightforward. When publishing a message we have a notion of the message being "committed" to the log. Once a published message is committed, it will not be lost as long as one broker that
+ replicates the partition to which this message was written remains "alive". The definition of committed message and alive partition as well as a description of which types of failures we attempt to handle will be
described in more detail in the next section. For now let's assume a perfect, lossless broker and try to understand the guarantees to the producer and consumer. If a producer attempts to publish a message and
- experiences a network error it cannot be sure if this error happened before or after the message was committed. This is similar to the semantics of inserting into a database table with an autogenerated key.
+ experiences a network error, it cannot be sure if this error happened before or after the message was committed. This is similar to the semantics of inserting into a database table with an autogenerated key.
<p>
Prior to 0.11.0.0, if a producer failed to receive a response indicating that a message was committed, it had little choice but to resend the message. This provides at-least-once delivery semantics since the
message may be written to the log again during resending if the original request had in fact succeeded. Since 0.11.0.0, the Kafka producer also supports an idempotent delivery option which guarantees that resending
will not result in duplicate entries in the log. To achieve this, the broker assigns each producer an ID and deduplicates messages using a sequence number that is sent by the producer along with every message.
- Also beginning with 0.11.0.0, the producer supports the ability to send messages to multiple topic partitions using transaction-like semantics: i.e. either all messages are successfully written or none of them are.
+ Also beginning with 0.11.0.0, the producer supports the ability to send messages to multiple topic partitions using transactional semantics, so that either all messages are successfully written or none of them are.
The main use case for this is exactly-once processing between Kafka topics (described below).
<p>
- Not all use cases require such strong guarantees. For uses which are latency sensitive we allow the producer to specify the durability level it desires. If the producer specifies that it wants to wait on the message
- being committed this can take on the order of 10 ms. However the producer can also specify that it wants to perform the send completely asynchronously or that it wants to wait only until the leader (but not
+ Not all use cases require such strong guarantees. For uses which are latency-sensitive, we allow the producer to specify the durability level it desires. If the producer specifies that it wants to wait on the message
+ being committed, this can take on the order of 10 ms. However the producer can also specify that it wants to perform the send completely asynchronously or that it wants to wait only until the leader (but not
necessarily the followers) have the message.
Review Comment:
You are probably correct but you are reviewing the existing text, not the changes. It could do with a fairly comprehensive re-write I think. I wasn't expecting to do that now. For example, it's hardly relevant nowadays what the 0.11.0.0 release introduced :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17671: Create better documentation for transactions [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17454:
URL: https://github.com/apache/kafka/pull/17454#discussion_r1807083423
##########
docs/design.html:
##########
@@ -254,32 +254,32 @@ <h3 class="anchor-heading"><a id="semantics" class="anchor-link"></a><a href="#s
<i>At least once</i>—Messages are never lost but may be redelivered.
</li>
<li>
- <i>Exactly once</i>—this is what people actually want, each message is delivered once and only once.
+ <i>Exactly once</i>—Each message is delivered once and only once.
Review Comment:
We should not say "delivered" (also further above) -- this term is too close to "delivery semantics at the network layer", -- we still retry internally, and _deliver_ a message more than once (holds true for idempotent producer, as well as aborted TX which are retried, and aborted stuff is just filtered later).
Btw: Further above we says:
> let's discuss the semantic guarantees Kafka provides between producer and consumer
This is also misleading, because EOS works for read-process-write, but not for upstream producer to downstream consumer. While the consumer is guarded to not return aborted messages to the app, it's still at-least-once as there is not guarantee that a committed messages is only read exactly-once.
##########
docs/design.html:
##########
@@ -254,32 +254,32 @@ <h3 class="anchor-heading"><a id="semantics" class="anchor-link"></a><a href="#s
<i>At least once</i>—Messages are never lost but may be redelivered.
</li>
<li>
- <i>Exactly once</i>—this is what people actually want, each message is delivered once and only once.
+ <i>Exactly once</i>—Each message is delivered once and only once.
</li>
</ul>
It's worth noting that this breaks down into two problems: the durability guarantees for publishing a message and the guarantees when consuming a message.
<p>
- Many systems claim to provide "exactly once" delivery semantics, but it is important to read the fine print, most of these claims are misleading (i.e. they don't translate to the case where consumers or producers
+ Many systems claim to provide "exactly-once" delivery semantics, but it is important to read the fine print, because sometimes these claims are misleading (i.e. they don't translate to the case where consumers or producers
can fail, cases where there are multiple consumer processes, or cases where data written to disk can be lost).
<p>
- Kafka's semantics are straight-forward. When publishing a message we have a notion of the message being "committed" to the log. Once a published message is committed it will not be lost as long as one broker that
- replicates the partition to which this message was written remains "alive". The definition of committed message, alive partition as well as a description of which types of failures we attempt to handle will be
+ Kafka's semantics are straightforward. When publishing a message we have a notion of the message being "committed" to the log. Once a published message is committed, it will not be lost as long as one broker that
+ replicates the partition to which this message was written remains "alive". The definition of committed message and alive partition as well as a description of which types of failures we attempt to handle will be
described in more detail in the next section. For now let's assume a perfect, lossless broker and try to understand the guarantees to the producer and consumer. If a producer attempts to publish a message and
- experiences a network error it cannot be sure if this error happened before or after the message was committed. This is similar to the semantics of inserting into a database table with an autogenerated key.
+ experiences a network error, it cannot be sure if this error happened before or after the message was committed. This is similar to the semantics of inserting into a database table with an autogenerated key.
<p>
Prior to 0.11.0.0, if a producer failed to receive a response indicating that a message was committed, it had little choice but to resend the message. This provides at-least-once delivery semantics since the
message may be written to the log again during resending if the original request had in fact succeeded. Since 0.11.0.0, the Kafka producer also supports an idempotent delivery option which guarantees that resending
Review Comment:
> resending
Might need to clarify the distinction between producer retries, vs app retires?
##########
docs/design.html:
##########
@@ -254,32 +254,32 @@ <h3 class="anchor-heading"><a id="semantics" class="anchor-link"></a><a href="#s
<i>At least once</i>—Messages are never lost but may be redelivered.
</li>
<li>
- <i>Exactly once</i>—this is what people actually want, each message is delivered once and only once.
+ <i>Exactly once</i>—Each message is delivered once and only once.
</li>
</ul>
It's worth noting that this breaks down into two problems: the durability guarantees for publishing a message and the guarantees when consuming a message.
<p>
- Many systems claim to provide "exactly once" delivery semantics, but it is important to read the fine print, most of these claims are misleading (i.e. they don't translate to the case where consumers or producers
+ Many systems claim to provide "exactly-once" delivery semantics, but it is important to read the fine print, because sometimes these claims are misleading (i.e. they don't translate to the case where consumers or producers
can fail, cases where there are multiple consumer processes, or cases where data written to disk can be lost).
<p>
- Kafka's semantics are straight-forward. When publishing a message we have a notion of the message being "committed" to the log. Once a published message is committed it will not be lost as long as one broker that
- replicates the partition to which this message was written remains "alive". The definition of committed message, alive partition as well as a description of which types of failures we attempt to handle will be
+ Kafka's semantics are straightforward. When publishing a message we have a notion of the message being "committed" to the log. Once a published message is committed, it will not be lost as long as one broker that
+ replicates the partition to which this message was written remains "alive". The definition of committed message and alive partition as well as a description of which types of failures we attempt to handle will be
described in more detail in the next section. For now let's assume a perfect, lossless broker and try to understand the guarantees to the producer and consumer. If a producer attempts to publish a message and
- experiences a network error it cannot be sure if this error happened before or after the message was committed. This is similar to the semantics of inserting into a database table with an autogenerated key.
+ experiences a network error, it cannot be sure if this error happened before or after the message was committed. This is similar to the semantics of inserting into a database table with an autogenerated key.
<p>
Prior to 0.11.0.0, if a producer failed to receive a response indicating that a message was committed, it had little choice but to resend the message. This provides at-least-once delivery semantics since the
message may be written to the log again during resending if the original request had in fact succeeded. Since 0.11.0.0, the Kafka producer also supports an idempotent delivery option which guarantees that resending
will not result in duplicate entries in the log. To achieve this, the broker assigns each producer an ID and deduplicates messages using a sequence number that is sent by the producer along with every message.
- Also beginning with 0.11.0.0, the producer supports the ability to send messages to multiple topic partitions using transaction-like semantics: i.e. either all messages are successfully written or none of them are.
+ Also beginning with 0.11.0.0, the producer supports the ability to send messages to multiple topic partitions using transactional semantics, so that either all messages are successfully written or none of them are.
The main use case for this is exactly-once processing between Kafka topics (described below).
Review Comment:
I would call it "exactly-once processing" -- it's just an "atomic write".
##########
docs/design.html:
##########
@@ -290,24 +290,58 @@ <h3 class="anchor-heading"><a id="semantics" class="anchor-link"></a><a href="#s
messages have a primary key and so the updates are idempotent (receiving the same message twice just overwrites a record with another copy of itself).
</ol>
<p>
- So what about exactly once semantics (i.e. the thing you actually want)? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>
- application), we can leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in a topic, so we can write the offset to Kafka in the
- same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's position will revert to its old value and the produced data on the output topics will not be visible
- to other consumers, depending on their "isolation level." In the default "read_uncommitted" isolation level, all messages are visible to consumers even if they were part of an aborted transaction,
- but in "read_committed," the consumer will only return messages from transactions which were committed (and any messages which were not part of a transaction).
+ So what about exactly-once semantics? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a> application), we can
Review Comment:
We are (correctly) switching from producer -> topic -> consumer to read-process-write here, but that's very subtle. I think we need to call the difference out at the very top and make it explicit.
##########
docs/design.html:
##########
@@ -254,32 +254,32 @@ <h3 class="anchor-heading"><a id="semantics" class="anchor-link"></a><a href="#s
<i>At least once</i>—Messages are never lost but may be redelivered.
</li>
<li>
- <i>Exactly once</i>—this is what people actually want, each message is delivered once and only once.
+ <i>Exactly once</i>—Each message is delivered once and only once.
</li>
</ul>
It's worth noting that this breaks down into two problems: the durability guarantees for publishing a message and the guarantees when consuming a message.
<p>
- Many systems claim to provide "exactly once" delivery semantics, but it is important to read the fine print, most of these claims are misleading (i.e. they don't translate to the case where consumers or producers
+ Many systems claim to provide "exactly-once" delivery semantics, but it is important to read the fine print, because sometimes these claims are misleading (i.e. they don't translate to the case where consumers or producers
can fail, cases where there are multiple consumer processes, or cases where data written to disk can be lost).
<p>
- Kafka's semantics are straight-forward. When publishing a message we have a notion of the message being "committed" to the log. Once a published message is committed it will not be lost as long as one broker that
- replicates the partition to which this message was written remains "alive". The definition of committed message, alive partition as well as a description of which types of failures we attempt to handle will be
+ Kafka's semantics are straightforward. When publishing a message we have a notion of the message being "committed" to the log. Once a published message is committed, it will not be lost as long as one broker that
Review Comment:
> straightforward
Sounds like a huge simplification...
Also, for non-transaction, there no such thing as "committed" -- seems we should clarify this better? Why not use "acknowledge" which would be the proper Kafka term?
##########
docs/design.html:
##########
@@ -290,24 +290,58 @@ <h3 class="anchor-heading"><a id="semantics" class="anchor-link"></a><a href="#s
messages have a primary key and so the updates are idempotent (receiving the same message twice just overwrites a record with another copy of itself).
</ol>
<p>
- So what about exactly once semantics (i.e. the thing you actually want)? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>
- application), we can leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in a topic, so we can write the offset to Kafka in the
- same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's position will revert to its old value and the produced data on the output topics will not be visible
- to other consumers, depending on their "isolation level." In the default "read_uncommitted" isolation level, all messages are visible to consumers even if they were part of an aborted transaction,
- but in "read_committed," the consumer will only return messages from transactions which were committed (and any messages which were not part of a transaction).
+ So what about exactly-once semantics? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a> application), we can
+ leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in an internal topic, so we can write the offset to Kafka in the
+ same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's stored position will revert to its old value (although the consumer has to refetch the
+ committed offset because it does not automatically rewind) and the produced data on the output topics will not be visible to other consumers, depending on their "isolation level". In the default
+ "read_uncommitted" isolation level, all messages are visible to consumers even if they were part of an aborted transaction, but in "read_committed" isolation level, the consumer will only return messages
+ from transactions which were committed (and any messages which were not part of a transaction).
<p>
When writing to an external system, the limitation is in the need to coordinate the consumer's position with what is actually stored as output. The classic way of achieving this would be to introduce a two-phase
- commit between the storage of the consumer position and the storage of the consumers output. But this can be handled more simply and generally by letting the consumer store its offset in the same place as
+ commit between the storage of the consumer position and the storage of the consumers output. This can be handled more simply and generally by letting the consumer store its offset in the same place as
its output. This is better because many of the output systems a consumer might want to write to will not support a two-phase commit. As an example of this, consider a
<a href="https://kafka.apache.org/documentation/#connect">Kafka Connect</a> connector which populates data in HDFS along with the offsets of the data it reads so that it is guaranteed that either data and
offsets are both updated or neither is. We follow similar patterns for many other data systems which require these stronger semantics and for which the messages do not have a primary key to allow for deduplication.
<p>
- So effectively Kafka supports exactly-once delivery in <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>, and the transactional producer/consumer can be used generally to provide
+ As a result, Kafka supports exactly-once delivery in <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>, and the transactional producer/consumer can be used generally to provide
exactly-once delivery when transferring and processing data between Kafka topics. Exactly-once delivery for other destination systems generally requires cooperation with such systems, but Kafka provides the
offset which makes implementing this feasible (see also <a href="https://kafka.apache.org/documentation/#connect">Kafka Connect</a>). Otherwise, Kafka guarantees at-least-once delivery by default, and allows
Review Comment:
```suggestion
primitives which makes implementing this feasible (see also <a href="https://kafka.apache.org/documentation/#connect">Kafka Connect</a>). Otherwise, Kafka guarantees at-least-once delivery by default, and allows
```
##########
docs/design.html:
##########
@@ -254,32 +254,32 @@ <h3 class="anchor-heading"><a id="semantics" class="anchor-link"></a><a href="#s
<i>At least once</i>—Messages are never lost but may be redelivered.
</li>
<li>
- <i>Exactly once</i>—this is what people actually want, each message is delivered once and only once.
+ <i>Exactly once</i>—Each message is delivered once and only once.
</li>
</ul>
It's worth noting that this breaks down into two problems: the durability guarantees for publishing a message and the guarantees when consuming a message.
<p>
- Many systems claim to provide "exactly once" delivery semantics, but it is important to read the fine print, most of these claims are misleading (i.e. they don't translate to the case where consumers or producers
+ Many systems claim to provide "exactly-once" delivery semantics, but it is important to read the fine print, because sometimes these claims are misleading (i.e. they don't translate to the case where consumers or producers
can fail, cases where there are multiple consumer processes, or cases where data written to disk can be lost).
<p>
- Kafka's semantics are straight-forward. When publishing a message we have a notion of the message being "committed" to the log. Once a published message is committed it will not be lost as long as one broker that
- replicates the partition to which this message was written remains "alive". The definition of committed message, alive partition as well as a description of which types of failures we attempt to handle will be
+ Kafka's semantics are straightforward. When publishing a message we have a notion of the message being "committed" to the log. Once a published message is committed, it will not be lost as long as one broker that
+ replicates the partition to which this message was written remains "alive". The definition of committed message and alive partition as well as a description of which types of failures we attempt to handle will be
described in more detail in the next section. For now let's assume a perfect, lossless broker and try to understand the guarantees to the producer and consumer. If a producer attempts to publish a message and
- experiences a network error it cannot be sure if this error happened before or after the message was committed. This is similar to the semantics of inserting into a database table with an autogenerated key.
+ experiences a network error, it cannot be sure if this error happened before or after the message was committed. This is similar to the semantics of inserting into a database table with an autogenerated key.
<p>
Prior to 0.11.0.0, if a producer failed to receive a response indicating that a message was committed, it had little choice but to resend the message. This provides at-least-once delivery semantics since the
message may be written to the log again during resending if the original request had in fact succeeded. Since 0.11.0.0, the Kafka producer also supports an idempotent delivery option which guarantees that resending
will not result in duplicate entries in the log. To achieve this, the broker assigns each producer an ID and deduplicates messages using a sequence number that is sent by the producer along with every message.
- Also beginning with 0.11.0.0, the producer supports the ability to send messages to multiple topic partitions using transaction-like semantics: i.e. either all messages are successfully written or none of them are.
+ Also beginning with 0.11.0.0, the producer supports the ability to send messages to multiple topic partitions using transactional semantics, so that either all messages are successfully written or none of them are.
The main use case for this is exactly-once processing between Kafka topics (described below).
<p>
- Not all use cases require such strong guarantees. For uses which are latency sensitive we allow the producer to specify the durability level it desires. If the producer specifies that it wants to wait on the message
- being committed this can take on the order of 10 ms. However the producer can also specify that it wants to perform the send completely asynchronously or that it wants to wait only until the leader (but not
+ Not all use cases require such strong guarantees. For uses which are latency-sensitive, we allow the producer to specify the durability level it desires. If the producer specifies that it wants to wait on the message
+ being committed, this can take on the order of 10 ms. However the producer can also specify that it wants to perform the send completely asynchronously or that it wants to wait only until the leader (but not
necessarily the followers) have the message.
Review Comment:
Seems we are jumping back and forth between different things here.
##########
docs/design.html:
##########
@@ -290,24 +290,58 @@ <h3 class="anchor-heading"><a id="semantics" class="anchor-link"></a><a href="#s
messages have a primary key and so the updates are idempotent (receiving the same message twice just overwrites a record with another copy of itself).
</ol>
<p>
- So what about exactly once semantics (i.e. the thing you actually want)? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>
- application), we can leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in a topic, so we can write the offset to Kafka in the
- same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's position will revert to its old value and the produced data on the output topics will not be visible
- to other consumers, depending on their "isolation level." In the default "read_uncommitted" isolation level, all messages are visible to consumers even if they were part of an aborted transaction,
- but in "read_committed," the consumer will only return messages from transactions which were committed (and any messages which were not part of a transaction).
+ So what about exactly-once semantics? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a> application), we can
+ leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in an internal topic, so we can write the offset to Kafka in the
+ same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's stored position will revert to its old value (although the consumer has to refetch the
+ committed offset because it does not automatically rewind) and the produced data on the output topics will not be visible to other consumers, depending on their "isolation level". In the default
+ "read_uncommitted" isolation level, all messages are visible to consumers even if they were part of an aborted transaction, but in "read_committed" isolation level, the consumer will only return messages
+ from transactions which were committed (and any messages which were not part of a transaction).
Review Comment:
If we consider a single read-process-write pattern, the consumer's isolation level does not really matter too much? Of course, if the input topic is transactional, having the consumer in read-committed mode is desirable, but from the POV how EOS work, it seems somewhat orthogonal? (Or do I start to split hairs...?)
##########
docs/design.html:
##########
@@ -254,32 +254,32 @@ <h3 class="anchor-heading"><a id="semantics" class="anchor-link"></a><a href="#s
<i>At least once</i>—Messages are never lost but may be redelivered.
</li>
<li>
- <i>Exactly once</i>—this is what people actually want, each message is delivered once and only once.
+ <i>Exactly once</i>—Each message is delivered once and only once.
</li>
</ul>
It's worth noting that this breaks down into two problems: the durability guarantees for publishing a message and the guarantees when consuming a message.
<p>
- Many systems claim to provide "exactly once" delivery semantics, but it is important to read the fine print, most of these claims are misleading (i.e. they don't translate to the case where consumers or producers
+ Many systems claim to provide "exactly-once" delivery semantics, but it is important to read the fine print, because sometimes these claims are misleading (i.e. they don't translate to the case where consumers or producers
can fail, cases where there are multiple consumer processes, or cases where data written to disk can be lost).
<p>
- Kafka's semantics are straight-forward. When publishing a message we have a notion of the message being "committed" to the log. Once a published message is committed it will not be lost as long as one broker that
- replicates the partition to which this message was written remains "alive". The definition of committed message, alive partition as well as a description of which types of failures we attempt to handle will be
+ Kafka's semantics are straightforward. When publishing a message we have a notion of the message being "committed" to the log. Once a published message is committed, it will not be lost as long as one broker that
+ replicates the partition to which this message was written remains "alive". The definition of committed message and alive partition as well as a description of which types of failures we attempt to handle will be
described in more detail in the next section. For now let's assume a perfect, lossless broker and try to understand the guarantees to the producer and consumer. If a producer attempts to publish a message and
- experiences a network error it cannot be sure if this error happened before or after the message was committed. This is similar to the semantics of inserting into a database table with an autogenerated key.
+ experiences a network error, it cannot be sure if this error happened before or after the message was committed. This is similar to the semantics of inserting into a database table with an autogenerated key.
<p>
Prior to 0.11.0.0, if a producer failed to receive a response indicating that a message was committed, it had little choice but to resend the message. This provides at-least-once delivery semantics since the
message may be written to the log again during resending if the original request had in fact succeeded. Since 0.11.0.0, the Kafka producer also supports an idempotent delivery option which guarantees that resending
will not result in duplicate entries in the log. To achieve this, the broker assigns each producer an ID and deduplicates messages using a sequence number that is sent by the producer along with every message.
- Also beginning with 0.11.0.0, the producer supports the ability to send messages to multiple topic partitions using transaction-like semantics: i.e. either all messages are successfully written or none of them are.
+ Also beginning with 0.11.0.0, the producer supports the ability to send messages to multiple topic partitions using transactional semantics, so that either all messages are successfully written or none of them are.
The main use case for this is exactly-once processing between Kafka topics (described below).
<p>
- Not all use cases require such strong guarantees. For uses which are latency sensitive we allow the producer to specify the durability level it desires. If the producer specifies that it wants to wait on the message
- being committed this can take on the order of 10 ms. However the producer can also specify that it wants to perform the send completely asynchronously or that it wants to wait only until the leader (but not
+ Not all use cases require such strong guarantees. For uses which are latency-sensitive, we allow the producer to specify the durability level it desires. If the producer specifies that it wants to wait on the message
Review Comment:
```suggestion
Not all use cases require such strong guarantees. For use cases which are latency-sensitive, we allow the producer to specify the durability level it desires. If the producer specifies that it wants to wait on the message
```
##########
docs/design.html:
##########
@@ -290,24 +290,58 @@ <h3 class="anchor-heading"><a id="semantics" class="anchor-link"></a><a href="#s
messages have a primary key and so the updates are idempotent (receiving the same message twice just overwrites a record with another copy of itself).
</ol>
<p>
- So what about exactly once semantics (i.e. the thing you actually want)? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>
- application), we can leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in a topic, so we can write the offset to Kafka in the
- same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's position will revert to its old value and the produced data on the output topics will not be visible
- to other consumers, depending on their "isolation level." In the default "read_uncommitted" isolation level, all messages are visible to consumers even if they were part of an aborted transaction,
- but in "read_committed," the consumer will only return messages from transactions which were committed (and any messages which were not part of a transaction).
+ So what about exactly-once semantics? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a> application), we can
+ leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in an internal topic, so we can write the offset to Kafka in the
+ same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's stored position will revert to its old value (although the consumer has to refetch the
+ committed offset because it does not automatically rewind) and the produced data on the output topics will not be visible to other consumers, depending on their "isolation level". In the default
+ "read_uncommitted" isolation level, all messages are visible to consumers even if they were part of an aborted transaction, but in "read_committed" isolation level, the consumer will only return messages
+ from transactions which were committed (and any messages which were not part of a transaction).
<p>
When writing to an external system, the limitation is in the need to coordinate the consumer's position with what is actually stored as output. The classic way of achieving this would be to introduce a two-phase
- commit between the storage of the consumer position and the storage of the consumers output. But this can be handled more simply and generally by letting the consumer store its offset in the same place as
+ commit between the storage of the consumer position and the storage of the consumers output. This can be handled more simply and generally by letting the consumer store its offset in the same place as
its output. This is better because many of the output systems a consumer might want to write to will not support a two-phase commit. As an example of this, consider a
<a href="https://kafka.apache.org/documentation/#connect">Kafka Connect</a> connector which populates data in HDFS along with the offsets of the data it reads so that it is guaranteed that either data and
offsets are both updated or neither is. We follow similar patterns for many other data systems which require these stronger semantics and for which the messages do not have a primary key to allow for deduplication.
<p>
- So effectively Kafka supports exactly-once delivery in <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>, and the transactional producer/consumer can be used generally to provide
+ As a result, Kafka supports exactly-once delivery in <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>, and the transactional producer/consumer can be used generally to provide
exactly-once delivery when transferring and processing data between Kafka topics. Exactly-once delivery for other destination systems generally requires cooperation with such systems, but Kafka provides the
offset which makes implementing this feasible (see also <a href="https://kafka.apache.org/documentation/#connect">Kafka Connect</a>). Otherwise, Kafka guarantees at-least-once delivery by default, and allows
the user to implement at-most-once delivery by disabling retries on the producer and committing offsets in the consumer prior to processing a batch of messages.
- <h3 class="anchor-heading"><a id="replication" class="anchor-link"></a><a href="#replication">4.7 Replication</a></h3>
+ <h3 class="anchor-heading"><a id="usingtransactions" class="anchor-link"></a><a href="#usingtransactions">4.7 Using Transactions</a></h3>
+ <p>
+ As mentioned above, the simplest way to get exactly-once semantics from Kafka is to use <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>. However, it is also possible to achieve
+ the same transactional guarantees using the Kafka producer and consumer directly by using them in the same way as Kafka Streams does.
+ <p>
+ Kafka transactions are a bit different than transactions in other messaging systems. In Kafka, the consumer and producer are separate and it is only the producer which is transactional. It is however able to
+ make transactional updates to the consumer's position (confusingly called the "committed offset"), and it is this which gives the overall exactly-once behavior.
+ <p>
+ There are three key aspects to exactly-once processing using the producer and consumer, which match how Kafka Streams works.
+ <ol>
+ <li>The consumer uses partition assignment to ensure that it is the only consumer in the consumer group currently processing each partition.</li>
+ <li>The consumer uses read-committed isolation level to ensure that it does not consume records produced by transactions which aborted.</li>
+ <li>The producer uses transactions so that all of the records it produces, and any offsets it updates on behalf of the consumer, are performed atomically.</li>
+ </ol>
+ <p>
+ The consumer configuration must include <code>isolation.level=read_committed</code> and <code>enable.auto.commit=false</code>. The producer configuration must set <code>transactional.id</code>
+ to the name of the transactional ID to be used, which configures the producer for transactional delivery and also makes sure that a restarted application causes any in-flight transaction from
+ the previous instance to abort. Only the producer has the <code>transactional.id</code> configuration. Strictly speaking, the consumer doesn't have to use read_committed isolation level, but if
+ it does not, it will see records from aborted transactions and also open transactions which have not yet completed, which seems undesirable if trying to achieve exactly-once.
+ <p>
+ Here's an example of a <a href="https://github.com/apache/kafka/blob/trunk/tools/src/main/java/org/apache/kafka/tools/TransactionalMessageCopier.java">transactional message copier</a>
+ which uses these principles. It uses a <code>KafkaConsumer</code> to consume records from one topic and a <code>KafkaProducer</code> to produce records to another topic. It uses transactions to ensure
+ that there is no duplication or loss of records as they are copied.
+ <p>
+ It is important to handle exceptions and aborted transactions correctly. Any records written by the transational producer will be marked as being part of the transactions, and then when the
+ transaction commits or aborts, transaction marker records are written to indicate the outcome of the transaction. This is how the read-committed consumer does not see records from aborted
+ transactions. However, in the event of a transaction abort, the application's in-memory state and in particular the current position of the consumer must be reset explicitly so that it can
Review Comment:
```suggestion
transactions. However, in the event of a transaction abort, the application's state and in particular the current position of the consumer must be reset explicitly so that it can
```
##########
docs/design.html:
##########
@@ -290,24 +290,58 @@ <h3 class="anchor-heading"><a id="semantics" class="anchor-link"></a><a href="#s
messages have a primary key and so the updates are idempotent (receiving the same message twice just overwrites a record with another copy of itself).
</ol>
<p>
- So what about exactly once semantics (i.e. the thing you actually want)? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>
- application), we can leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in a topic, so we can write the offset to Kafka in the
- same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's position will revert to its old value and the produced data on the output topics will not be visible
- to other consumers, depending on their "isolation level." In the default "read_uncommitted" isolation level, all messages are visible to consumers even if they were part of an aborted transaction,
- but in "read_committed," the consumer will only return messages from transactions which were committed (and any messages which were not part of a transaction).
+ So what about exactly-once semantics? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a> application), we can
+ leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in an internal topic, so we can write the offset to Kafka in the
+ same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's stored position will revert to its old value (although the consumer has to refetch the
+ committed offset because it does not automatically rewind) and the produced data on the output topics will not be visible to other consumers, depending on their "isolation level". In the default
+ "read_uncommitted" isolation level, all messages are visible to consumers even if they were part of an aborted transaction, but in "read_committed" isolation level, the consumer will only return messages
+ from transactions which were committed (and any messages which were not part of a transaction).
<p>
When writing to an external system, the limitation is in the need to coordinate the consumer's position with what is actually stored as output. The classic way of achieving this would be to introduce a two-phase
- commit between the storage of the consumer position and the storage of the consumers output. But this can be handled more simply and generally by letting the consumer store its offset in the same place as
+ commit between the storage of the consumer position and the storage of the consumers output. This can be handled more simply and generally by letting the consumer store its offset in the same place as
its output. This is better because many of the output systems a consumer might want to write to will not support a two-phase commit. As an example of this, consider a
<a href="https://kafka.apache.org/documentation/#connect">Kafka Connect</a> connector which populates data in HDFS along with the offsets of the data it reads so that it is guaranteed that either data and
offsets are both updated or neither is. We follow similar patterns for many other data systems which require these stronger semantics and for which the messages do not have a primary key to allow for deduplication.
<p>
- So effectively Kafka supports exactly-once delivery in <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>, and the transactional producer/consumer can be used generally to provide
+ As a result, Kafka supports exactly-once delivery in <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>, and the transactional producer/consumer can be used generally to provide
Review Comment:
```suggestion
As a result, Kafka supports exactly-once processing semantics in <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>, and the transactional producer can be used generally to provide
```
##########
docs/design.html:
##########
@@ -254,32 +254,32 @@ <h3 class="anchor-heading"><a id="semantics" class="anchor-link"></a><a href="#s
<i>At least once</i>—Messages are never lost but may be redelivered.
</li>
<li>
- <i>Exactly once</i>—this is what people actually want, each message is delivered once and only once.
+ <i>Exactly once</i>—Each message is delivered once and only once.
</li>
</ul>
It's worth noting that this breaks down into two problems: the durability guarantees for publishing a message and the guarantees when consuming a message.
<p>
- Many systems claim to provide "exactly once" delivery semantics, but it is important to read the fine print, most of these claims are misleading (i.e. they don't translate to the case where consumers or producers
+ Many systems claim to provide "exactly-once" delivery semantics, but it is important to read the fine print, because sometimes these claims are misleading (i.e. they don't translate to the case where consumers or producers
can fail, cases where there are multiple consumer processes, or cases where data written to disk can be lost).
<p>
- Kafka's semantics are straight-forward. When publishing a message we have a notion of the message being "committed" to the log. Once a published message is committed it will not be lost as long as one broker that
- replicates the partition to which this message was written remains "alive". The definition of committed message, alive partition as well as a description of which types of failures we attempt to handle will be
+ Kafka's semantics are straightforward. When publishing a message we have a notion of the message being "committed" to the log. Once a published message is committed, it will not be lost as long as one broker that
+ replicates the partition to which this message was written remains "alive". The definition of committed message and alive partition as well as a description of which types of failures we attempt to handle will be
described in more detail in the next section. For now let's assume a perfect, lossless broker and try to understand the guarantees to the producer and consumer. If a producer attempts to publish a message and
- experiences a network error it cannot be sure if this error happened before or after the message was committed. This is similar to the semantics of inserting into a database table with an autogenerated key.
+ experiences a network error, it cannot be sure if this error happened before or after the message was committed. This is similar to the semantics of inserting into a database table with an autogenerated key.
<p>
Prior to 0.11.0.0, if a producer failed to receive a response indicating that a message was committed, it had little choice but to resend the message. This provides at-least-once delivery semantics since the
message may be written to the log again during resending if the original request had in fact succeeded. Since 0.11.0.0, the Kafka producer also supports an idempotent delivery option which guarantees that resending
will not result in duplicate entries in the log. To achieve this, the broker assigns each producer an ID and deduplicates messages using a sequence number that is sent by the producer along with every message.
- Also beginning with 0.11.0.0, the producer supports the ability to send messages to multiple topic partitions using transaction-like semantics: i.e. either all messages are successfully written or none of them are.
+ Also beginning with 0.11.0.0, the producer supports the ability to send messages to multiple topic partitions using transactional semantics, so that either all messages are successfully written or none of them are.
The main use case for this is exactly-once processing between Kafka topics (described below).
<p>
- Not all use cases require such strong guarantees. For uses which are latency sensitive we allow the producer to specify the durability level it desires. If the producer specifies that it wants to wait on the message
- being committed this can take on the order of 10 ms. However the producer can also specify that it wants to perform the send completely asynchronously or that it wants to wait only until the leader (but not
+ Not all use cases require such strong guarantees. For uses which are latency-sensitive, we allow the producer to specify the durability level it desires. If the producer specifies that it wants to wait on the message
Review Comment:
What is a "durability level" -- are we talking about `acks=...` here?
##########
docs/design.html:
##########
@@ -290,24 +290,56 @@ <h3 class="anchor-heading"><a id="semantics" class="anchor-link"></a><a href="#s
messages have a primary key and so the updates are idempotent (receiving the same message twice just overwrites a record with another copy of itself).
</ol>
<p>
- So what about exactly once semantics (i.e. the thing you actually want)? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>
- application), we can leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in a topic, so we can write the offset to Kafka in the
- same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's position will revert to its old value and the produced data on the output topics will not be visible
- to other consumers, depending on their "isolation level." In the default "read_uncommitted" isolation level, all messages are visible to consumers even if they were part of an aborted transaction,
- but in "read_committed," the consumer will only return messages from transactions which were committed (and any messages which were not part of a transaction).
+ So what about exactly-once semantics? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a> application), we can
+ leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in an internal topic, so we can write the offset to Kafka in the
+ same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's stored position will revert to its old value and the produced data on the output topics will not
+ be visible to other consumers, depending on their "isolation level." In the default "read_uncommitted" isolation level, all messages are visible to consumers even if they were part of an aborted transaction,
+ but in "read_committed" isolation level, the consumer will only return messages from transactions which were committed (and any messages which were not part of a transaction).
<p>
When writing to an external system, the limitation is in the need to coordinate the consumer's position with what is actually stored as output. The classic way of achieving this would be to introduce a two-phase
- commit between the storage of the consumer position and the storage of the consumers output. But this can be handled more simply and generally by letting the consumer store its offset in the same place as
+ commit between the storage of the consumer position and the storage of the consumers output. This can be handled more simply and generally by letting the consumer store its offset in the same place as
its output. This is better because many of the output systems a consumer might want to write to will not support a two-phase commit. As an example of this, consider a
<a href="https://kafka.apache.org/documentation/#connect">Kafka Connect</a> connector which populates data in HDFS along with the offsets of the data it reads so that it is guaranteed that either data and
offsets are both updated or neither is. We follow similar patterns for many other data systems which require these stronger semantics and for which the messages do not have a primary key to allow for deduplication.
<p>
- So effectively Kafka supports exactly-once delivery in <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>, and the transactional producer/consumer can be used generally to provide
+ As a result, Kafka supports exactly-once delivery in <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>, and the transactional producer/consumer can be used generally to provide
exactly-once delivery when transferring and processing data between Kafka topics. Exactly-once delivery for other destination systems generally requires cooperation with such systems, but Kafka provides the
offset which makes implementing this feasible (see also <a href="https://kafka.apache.org/documentation/#connect">Kafka Connect</a>). Otherwise, Kafka guarantees at-least-once delivery by default, and allows
the user to implement at-most-once delivery by disabling retries on the producer and committing offsets in the consumer prior to processing a batch of messages.
- <h3 class="anchor-heading"><a id="replication" class="anchor-link"></a><a href="#replication">4.7 Replication</a></h3>
+ <h3 class="anchor-heading"><a id="usingtransactions" class="anchor-link"></a><a href="#usingtransactions">4.7 Using Transactions</a></h3>
+ <p>
+ As mentioned above, the simplest way to get exactly-once semantics from Kafka is to use <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>. However, it is also possible to achieve
+ the same transactional guarantees using the Kafka producer and consumer directly by using them in the same way as Kafka Streams does.
+ <p>
+ Kafka transactions are a bit different than transactions in other messaging systems. In Kafka, the consumer and producer are separate and it is only the producer which is transactional. It is however able to
+ make transactional updates to the consumer's position (confusingly called the "committed offset"), and it is this which gives the overall exactly-once behavior.
+ <p>
+ There are three key aspects to exactly-once processing using the producer and consumer, which match how Kafka Streams works.
+ <ol>
+ <li>The consumer uses partition assignment to ensure that it is the only consumer in the consumer group currently processing each partition.</li>
+ <li>The consumer uses read-committed isolation level to ensure that it does not consume records produced by transactions which aborted.</li>
+ <li>The producer uses transactions so that all of the records it produces, and any offsets it updates on behalf of the consumer, are performed atomically.</li>
+ </ol>
+ <p>
+ The consumer configuration must include <code>isolation.level=read_committed</code> and <code>enable.auto.commit=false</code>. The producer configuration must set <code>transactional.id</code>
Review Comment:
Seems, I did not split hairs above :)
##########
docs/design.html:
##########
@@ -290,24 +290,58 @@ <h3 class="anchor-heading"><a id="semantics" class="anchor-link"></a><a href="#s
messages have a primary key and so the updates are idempotent (receiving the same message twice just overwrites a record with another copy of itself).
</ol>
<p>
- So what about exactly once semantics (i.e. the thing you actually want)? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>
- application), we can leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in a topic, so we can write the offset to Kafka in the
- same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's position will revert to its old value and the produced data on the output topics will not be visible
- to other consumers, depending on their "isolation level." In the default "read_uncommitted" isolation level, all messages are visible to consumers even if they were part of an aborted transaction,
- but in "read_committed," the consumer will only return messages from transactions which were committed (and any messages which were not part of a transaction).
+ So what about exactly-once semantics? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a> application), we can
+ leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in an internal topic, so we can write the offset to Kafka in the
+ same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's stored position will revert to its old value (although the consumer has to refetch the
+ committed offset because it does not automatically rewind) and the produced data on the output topics will not be visible to other consumers, depending on their "isolation level". In the default
+ "read_uncommitted" isolation level, all messages are visible to consumers even if they were part of an aborted transaction, but in "read_committed" isolation level, the consumer will only return messages
+ from transactions which were committed (and any messages which were not part of a transaction).
<p>
When writing to an external system, the limitation is in the need to coordinate the consumer's position with what is actually stored as output. The classic way of achieving this would be to introduce a two-phase
- commit between the storage of the consumer position and the storage of the consumers output. But this can be handled more simply and generally by letting the consumer store its offset in the same place as
+ commit between the storage of the consumer position and the storage of the consumers output. This can be handled more simply and generally by letting the consumer store its offset in the same place as
its output. This is better because many of the output systems a consumer might want to write to will not support a two-phase commit. As an example of this, consider a
<a href="https://kafka.apache.org/documentation/#connect">Kafka Connect</a> connector which populates data in HDFS along with the offsets of the data it reads so that it is guaranteed that either data and
offsets are both updated or neither is. We follow similar patterns for many other data systems which require these stronger semantics and for which the messages do not have a primary key to allow for deduplication.
<p>
- So effectively Kafka supports exactly-once delivery in <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>, and the transactional producer/consumer can be used generally to provide
+ As a result, Kafka supports exactly-once delivery in <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>, and the transactional producer/consumer can be used generally to provide
Review Comment:
There is no such thing as a "transactional consumer"? Or do you mean a consumer in "read-committed" mode?
##########
docs/design.html:
##########
@@ -290,24 +290,58 @@ <h3 class="anchor-heading"><a id="semantics" class="anchor-link"></a><a href="#s
messages have a primary key and so the updates are idempotent (receiving the same message twice just overwrites a record with another copy of itself).
</ol>
<p>
- So what about exactly once semantics (i.e. the thing you actually want)? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>
- application), we can leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in a topic, so we can write the offset to Kafka in the
- same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's position will revert to its old value and the produced data on the output topics will not be visible
- to other consumers, depending on their "isolation level." In the default "read_uncommitted" isolation level, all messages are visible to consumers even if they were part of an aborted transaction,
- but in "read_committed," the consumer will only return messages from transactions which were committed (and any messages which were not part of a transaction).
+ So what about exactly-once semantics? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a> application), we can
+ leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in an internal topic, so we can write the offset to Kafka in the
+ same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's stored position will revert to its old value (although the consumer has to refetch the
+ committed offset because it does not automatically rewind) and the produced data on the output topics will not be visible to other consumers, depending on their "isolation level". In the default
+ "read_uncommitted" isolation level, all messages are visible to consumers even if they were part of an aborted transaction, but in "read_committed" isolation level, the consumer will only return messages
+ from transactions which were committed (and any messages which were not part of a transaction).
<p>
When writing to an external system, the limitation is in the need to coordinate the consumer's position with what is actually stored as output. The classic way of achieving this would be to introduce a two-phase
- commit between the storage of the consumer position and the storage of the consumers output. But this can be handled more simply and generally by letting the consumer store its offset in the same place as
+ commit between the storage of the consumer position and the storage of the consumers output. This can be handled more simply and generally by letting the consumer store its offset in the same place as
its output. This is better because many of the output systems a consumer might want to write to will not support a two-phase commit. As an example of this, consider a
<a href="https://kafka.apache.org/documentation/#connect">Kafka Connect</a> connector which populates data in HDFS along with the offsets of the data it reads so that it is guaranteed that either data and
offsets are both updated or neither is. We follow similar patterns for many other data systems which require these stronger semantics and for which the messages do not have a primary key to allow for deduplication.
<p>
- So effectively Kafka supports exactly-once delivery in <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>, and the transactional producer/consumer can be used generally to provide
+ As a result, Kafka supports exactly-once delivery in <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>, and the transactional producer/consumer can be used generally to provide
exactly-once delivery when transferring and processing data between Kafka topics. Exactly-once delivery for other destination systems generally requires cooperation with such systems, but Kafka provides the
offset which makes implementing this feasible (see also <a href="https://kafka.apache.org/documentation/#connect">Kafka Connect</a>). Otherwise, Kafka guarantees at-least-once delivery by default, and allows
the user to implement at-most-once delivery by disabling retries on the producer and committing offsets in the consumer prior to processing a batch of messages.
- <h3 class="anchor-heading"><a id="replication" class="anchor-link"></a><a href="#replication">4.7 Replication</a></h3>
+ <h3 class="anchor-heading"><a id="usingtransactions" class="anchor-link"></a><a href="#usingtransactions">4.7 Using Transactions</a></h3>
+ <p>
+ As mentioned above, the simplest way to get exactly-once semantics from Kafka is to use <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>. However, it is also possible to achieve
+ the same transactional guarantees using the Kafka producer and consumer directly by using them in the same way as Kafka Streams does.
+ <p>
+ Kafka transactions are a bit different than transactions in other messaging systems. In Kafka, the consumer and producer are separate and it is only the producer which is transactional. It is however able to
+ make transactional updates to the consumer's position (confusingly called the "committed offset"), and it is this which gives the overall exactly-once behavior.
+ <p>
+ There are three key aspects to exactly-once processing using the producer and consumer, which match how Kafka Streams works.
+ <ol>
+ <li>The consumer uses partition assignment to ensure that it is the only consumer in the consumer group currently processing each partition.</li>
+ <li>The consumer uses read-committed isolation level to ensure that it does not consume records produced by transactions which aborted.</li>
+ <li>The producer uses transactions so that all of the records it produces, and any offsets it updates on behalf of the consumer, are performed atomically.</li>
+ </ol>
+ <p>
+ The consumer configuration must include <code>isolation.level=read_committed</code> and <code>enable.auto.commit=false</code>. The producer configuration must set <code>transactional.id</code>
+ to the name of the transactional ID to be used, which configures the producer for transactional delivery and also makes sure that a restarted application causes any in-flight transaction from
+ the previous instance to abort. Only the producer has the <code>transactional.id</code> configuration. Strictly speaking, the consumer doesn't have to use read_committed isolation level, but if
+ it does not, it will see records from aborted transactions and also open transactions which have not yet completed, which seems undesirable if trying to achieve exactly-once.
+ <p>
+ Here's an example of a <a href="https://github.com/apache/kafka/blob/trunk/tools/src/main/java/org/apache/kafka/tools/TransactionalMessageCopier.java">transactional message copier</a>
Review Comment:
This example seems only to work correctly, if `useGroupMetadata` option is enables... w/o it, EOS cannot be guaranteed.
##########
docs/design.html:
##########
@@ -290,24 +290,58 @@ <h3 class="anchor-heading"><a id="semantics" class="anchor-link"></a><a href="#s
messages have a primary key and so the updates are idempotent (receiving the same message twice just overwrites a record with another copy of itself).
</ol>
<p>
- So what about exactly once semantics (i.e. the thing you actually want)? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>
- application), we can leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in a topic, so we can write the offset to Kafka in the
- same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's position will revert to its old value and the produced data on the output topics will not be visible
- to other consumers, depending on their "isolation level." In the default "read_uncommitted" isolation level, all messages are visible to consumers even if they were part of an aborted transaction,
- but in "read_committed," the consumer will only return messages from transactions which were committed (and any messages which were not part of a transaction).
+ So what about exactly-once semantics? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a> application), we can
+ leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in an internal topic, so we can write the offset to Kafka in the
+ same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's stored position will revert to its old value (although the consumer has to refetch the
+ committed offset because it does not automatically rewind) and the produced data on the output topics will not be visible to other consumers, depending on their "isolation level". In the default
+ "read_uncommitted" isolation level, all messages are visible to consumers even if they were part of an aborted transaction, but in "read_committed" isolation level, the consumer will only return messages
+ from transactions which were committed (and any messages which were not part of a transaction).
<p>
When writing to an external system, the limitation is in the need to coordinate the consumer's position with what is actually stored as output. The classic way of achieving this would be to introduce a two-phase
- commit between the storage of the consumer position and the storage of the consumers output. But this can be handled more simply and generally by letting the consumer store its offset in the same place as
+ commit between the storage of the consumer position and the storage of the consumers output. This can be handled more simply and generally by letting the consumer store its offset in the same place as
its output. This is better because many of the output systems a consumer might want to write to will not support a two-phase commit. As an example of this, consider a
<a href="https://kafka.apache.org/documentation/#connect">Kafka Connect</a> connector which populates data in HDFS along with the offsets of the data it reads so that it is guaranteed that either data and
offsets are both updated or neither is. We follow similar patterns for many other data systems which require these stronger semantics and for which the messages do not have a primary key to allow for deduplication.
<p>
- So effectively Kafka supports exactly-once delivery in <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>, and the transactional producer/consumer can be used generally to provide
+ As a result, Kafka supports exactly-once delivery in <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>, and the transactional producer/consumer can be used generally to provide
exactly-once delivery when transferring and processing data between Kafka topics. Exactly-once delivery for other destination systems generally requires cooperation with such systems, but Kafka provides the
Review Comment:
> when transferring and processing data between Kafka topics
What does this mean? Sounds like read-process-write, ie, what KS does.
##########
docs/design.html:
##########
@@ -254,32 +254,32 @@ <h3 class="anchor-heading"><a id="semantics" class="anchor-link"></a><a href="#s
<i>At least once</i>—Messages are never lost but may be redelivered.
</li>
<li>
- <i>Exactly once</i>—this is what people actually want, each message is delivered once and only once.
+ <i>Exactly once</i>—Each message is delivered once and only once.
</li>
</ul>
It's worth noting that this breaks down into two problems: the durability guarantees for publishing a message and the guarantees when consuming a message.
<p>
- Many systems claim to provide "exactly once" delivery semantics, but it is important to read the fine print, most of these claims are misleading (i.e. they don't translate to the case where consumers or producers
+ Many systems claim to provide "exactly-once" delivery semantics, but it is important to read the fine print, because sometimes these claims are misleading (i.e. they don't translate to the case where consumers or producers
can fail, cases where there are multiple consumer processes, or cases where data written to disk can be lost).
<p>
- Kafka's semantics are straight-forward. When publishing a message we have a notion of the message being "committed" to the log. Once a published message is committed it will not be lost as long as one broker that
- replicates the partition to which this message was written remains "alive". The definition of committed message, alive partition as well as a description of which types of failures we attempt to handle will be
+ Kafka's semantics are straightforward. When publishing a message we have a notion of the message being "committed" to the log. Once a published message is committed, it will not be lost as long as one broker that
+ replicates the partition to which this message was written remains "alive". The definition of committed message and alive partition as well as a description of which types of failures we attempt to handle will be
described in more detail in the next section. For now let's assume a perfect, lossless broker and try to understand the guarantees to the producer and consumer. If a producer attempts to publish a message and
- experiences a network error it cannot be sure if this error happened before or after the message was committed. This is similar to the semantics of inserting into a database table with an autogenerated key.
+ experiences a network error, it cannot be sure if this error happened before or after the message was committed. This is similar to the semantics of inserting into a database table with an autogenerated key.
<p>
Prior to 0.11.0.0, if a producer failed to receive a response indicating that a message was committed, it had little choice but to resend the message. This provides at-least-once delivery semantics since the
message may be written to the log again during resending if the original request had in fact succeeded. Since 0.11.0.0, the Kafka producer also supports an idempotent delivery option which guarantees that resending
will not result in duplicate entries in the log. To achieve this, the broker assigns each producer an ID and deduplicates messages using a sequence number that is sent by the producer along with every message.
- Also beginning with 0.11.0.0, the producer supports the ability to send messages to multiple topic partitions using transaction-like semantics: i.e. either all messages are successfully written or none of them are.
+ Also beginning with 0.11.0.0, the producer supports the ability to send messages to multiple topic partitions using transactional semantics, so that either all messages are successfully written or none of them are.
Review Comment:
I like to describe TX as "multi-message multi-topic/partition atomic write". Should we add something like this?
##########
docs/design.html:
##########
@@ -290,24 +290,58 @@ <h3 class="anchor-heading"><a id="semantics" class="anchor-link"></a><a href="#s
messages have a primary key and so the updates are idempotent (receiving the same message twice just overwrites a record with another copy of itself).
</ol>
<p>
- So what about exactly once semantics (i.e. the thing you actually want)? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>
- application), we can leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in a topic, so we can write the offset to Kafka in the
- same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's position will revert to its old value and the produced data on the output topics will not be visible
- to other consumers, depending on their "isolation level." In the default "read_uncommitted" isolation level, all messages are visible to consumers even if they were part of an aborted transaction,
- but in "read_committed," the consumer will only return messages from transactions which were committed (and any messages which were not part of a transaction).
+ So what about exactly-once semantics? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a> application), we can
+ leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in an internal topic, so we can write the offset to Kafka in the
+ same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's stored position will revert to its old value (although the consumer has to refetch the
+ committed offset because it does not automatically rewind) and the produced data on the output topics will not be visible to other consumers, depending on their "isolation level". In the default
+ "read_uncommitted" isolation level, all messages are visible to consumers even if they were part of an aborted transaction, but in "read_committed" isolation level, the consumer will only return messages
+ from transactions which were committed (and any messages which were not part of a transaction).
<p>
When writing to an external system, the limitation is in the need to coordinate the consumer's position with what is actually stored as output. The classic way of achieving this would be to introduce a two-phase
- commit between the storage of the consumer position and the storage of the consumers output. But this can be handled more simply and generally by letting the consumer store its offset in the same place as
+ commit between the storage of the consumer position and the storage of the consumers output. This can be handled more simply and generally by letting the consumer store its offset in the same place as
its output. This is better because many of the output systems a consumer might want to write to will not support a two-phase commit. As an example of this, consider a
<a href="https://kafka.apache.org/documentation/#connect">Kafka Connect</a> connector which populates data in HDFS along with the offsets of the data it reads so that it is guaranteed that either data and
offsets are both updated or neither is. We follow similar patterns for many other data systems which require these stronger semantics and for which the messages do not have a primary key to allow for deduplication.
<p>
- So effectively Kafka supports exactly-once delivery in <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>, and the transactional producer/consumer can be used generally to provide
+ As a result, Kafka supports exactly-once delivery in <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>, and the transactional producer/consumer can be used generally to provide
exactly-once delivery when transferring and processing data between Kafka topics. Exactly-once delivery for other destination systems generally requires cooperation with such systems, but Kafka provides the
offset which makes implementing this feasible (see also <a href="https://kafka.apache.org/documentation/#connect">Kafka Connect</a>). Otherwise, Kafka guarantees at-least-once delivery by default, and allows
the user to implement at-most-once delivery by disabling retries on the producer and committing offsets in the consumer prior to processing a batch of messages.
- <h3 class="anchor-heading"><a id="replication" class="anchor-link"></a><a href="#replication">4.7 Replication</a></h3>
+ <h3 class="anchor-heading"><a id="usingtransactions" class="anchor-link"></a><a href="#usingtransactions">4.7 Using Transactions</a></h3>
+ <p>
+ As mentioned above, the simplest way to get exactly-once semantics from Kafka is to use <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>. However, it is also possible to achieve
+ the same transactional guarantees using the Kafka producer and consumer directly by using them in the same way as Kafka Streams does.
+ <p>
+ Kafka transactions are a bit different than transactions in other messaging systems. In Kafka, the consumer and producer are separate and it is only the producer which is transactional. It is however able to
+ make transactional updates to the consumer's position (confusingly called the "committed offset"), and it is this which gives the overall exactly-once behavior.
+ <p>
+ There are three key aspects to exactly-once processing using the producer and consumer, which match how Kafka Streams works.
+ <ol>
+ <li>The consumer uses partition assignment to ensure that it is the only consumer in the consumer group currently processing each partition.</li>
+ <li>The consumer uses read-committed isolation level to ensure that it does not consume records produced by transactions which aborted.</li>
+ <li>The producer uses transactions so that all of the records it produces, and any offsets it updates on behalf of the consumer, are performed atomically.</li>
+ </ol>
+ <p>
+ The consumer configuration must include <code>isolation.level=read_committed</code> and <code>enable.auto.commit=false</code>. The producer configuration must set <code>transactional.id</code>
+ to the name of the transactional ID to be used, which configures the producer for transactional delivery and also makes sure that a restarted application causes any in-flight transaction from
+ the previous instance to abort. Only the producer has the <code>transactional.id</code> configuration. Strictly speaking, the consumer doesn't have to use read_committed isolation level, but if
Review Comment:
There is actually much more to it... For KS, to do proper fencing in combination with rebalancing, we needed to use a producer-per-task. This is very expensive.
For EOSv2, we move to producer-per-thread model (which is basically described here) for which fencing does not happen on the producer side any longer (it might, be we don't rely on it), but we actually general random `transactional.id`s and rely on a build it "fencing" mechanism inside the ConsumerGroupCoordinator (in combination with broker side transaction timeouts...)
Not sure how detailed we want to be here, but bottom line is: this description might be over-simplified.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17671: Create better documentation for transactions [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #17454:
URL: https://github.com/apache/kafka/pull/17454#discussion_r1810885718
##########
docs/design.html:
##########
@@ -254,32 +254,32 @@ <h3 class="anchor-heading"><a id="semantics" class="anchor-link"></a><a href="#s
<i>At least once</i>—Messages are never lost but may be redelivered.
</li>
<li>
- <i>Exactly once</i>—this is what people actually want, each message is delivered once and only once.
+ <i>Exactly once</i>—Each message is delivered once and only once.
Review Comment:
I will try for "processed" then.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17671: Create better documentation for transactions [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #17454:
URL: https://github.com/apache/kafka/pull/17454#discussion_r1907866806
##########
docs/design.html:
##########
@@ -290,24 +290,56 @@ <h3 class="anchor-heading"><a id="semantics" class="anchor-link"></a><a href="#s
messages have a primary key and so the updates are idempotent (receiving the same message twice just overwrites a record with another copy of itself).
</ol>
<p>
- So what about exactly once semantics (i.e. the thing you actually want)? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>
- application), we can leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in a topic, so we can write the offset to Kafka in the
- same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's position will revert to its old value and the produced data on the output topics will not be visible
- to other consumers, depending on their "isolation level." In the default "read_uncommitted" isolation level, all messages are visible to consumers even if they were part of an aborted transaction,
- but in "read_committed," the consumer will only return messages from transactions which were committed (and any messages which were not part of a transaction).
+ So what about exactly-once semantics? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a> application), we can
+ leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in an internal topic, so we can write the offset to Kafka in the
+ same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's stored position will revert to its old value and the produced data on the output topics will not
+ be visible to other consumers, depending on their "isolation level." In the default "read_uncommitted" isolation level, all messages are visible to consumers even if they were part of an aborted transaction,
+ but in "read_committed" isolation level, the consumer will only return messages from transactions which were committed (and any messages which were not part of a transaction).
<p>
When writing to an external system, the limitation is in the need to coordinate the consumer's position with what is actually stored as output. The classic way of achieving this would be to introduce a two-phase
- commit between the storage of the consumer position and the storage of the consumers output. But this can be handled more simply and generally by letting the consumer store its offset in the same place as
+ commit between the storage of the consumer position and the storage of the consumers output. This can be handled more simply and generally by letting the consumer store its offset in the same place as
its output. This is better because many of the output systems a consumer might want to write to will not support a two-phase commit. As an example of this, consider a
<a href="https://kafka.apache.org/documentation/#connect">Kafka Connect</a> connector which populates data in HDFS along with the offsets of the data it reads so that it is guaranteed that either data and
offsets are both updated or neither is. We follow similar patterns for many other data systems which require these stronger semantics and for which the messages do not have a primary key to allow for deduplication.
<p>
- So effectively Kafka supports exactly-once delivery in <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>, and the transactional producer/consumer can be used generally to provide
+ As a result, Kafka supports exactly-once delivery in <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>, and the transactional producer/consumer can be used generally to provide
exactly-once delivery when transferring and processing data between Kafka topics. Exactly-once delivery for other destination systems generally requires cooperation with such systems, but Kafka provides the
offset which makes implementing this feasible (see also <a href="https://kafka.apache.org/documentation/#connect">Kafka Connect</a>). Otherwise, Kafka guarantees at-least-once delivery by default, and allows
the user to implement at-most-once delivery by disabling retries on the producer and committing offsets in the consumer prior to processing a batch of messages.
- <h3 class="anchor-heading"><a id="replication" class="anchor-link"></a><a href="#replication">4.7 Replication</a></h3>
+ <h3 class="anchor-heading"><a id="usingtransactions" class="anchor-link"></a><a href="#usingtransactions">4.7 Using Transactions</a></h3>
+ <p>
+ As mentioned above, the simplest way to get exactly-once semantics from Kafka is to use <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>. However, it is also possible to achieve
+ the same transactional guarantees using the Kafka producer and consumer directly by using them in the same way as Kafka Streams does.
+ <p>
+ Kafka transactions are a bit different than transactions in other messaging systems. In Kafka, the consumer and producer are separate and it is only the producer which is transactional. It is however able to
+ make transactional updates to the consumer's position (confusingly called the "committed offset"), and it is this which gives the overall exactly-once behavior.
+ <p>
+ There are three key aspects to exactly-once processing using the producer and consumer, which match how Kafka Streams works.
+ <ol>
+ <li>The consumer uses partition assignment to ensure that it is the only consumer in the consumer group currently processing each partition.</li>
+ <li>The consumer uses read-committed isolation level to ensure that it does not consume records produced by transactions which aborted.</li>
+ <li>The producer uses transactions so that all of the records it produces, and any offsets it updates on behalf of the consumer, are performed atomically.</li>
+ </ol>
+ <p>
+ The consumer configuration must include <code>isolation.level=read_committed</code> and <code>enable.auto.commit=false</code>. The producer configuration must set <code>transactional.id</code>
+ to the name of the transactional ID to be used, which configures the producer for transactional delivery and also makes sure that a restarted application causes any in-flight transaction from
+ the previous instance to abort. Only the producer has the <code>transactional.id</code> configuration.
+ <p>
+ Here's an example of a <a href="https://github.com/apache/kafka/blob/trunk/tools/src/main/java/org/apache/kafka/tools/TransactionalMessageCopier.java">transactional message copier</a>
+ which uses these principles. It uses a <code>KafkaConsumer</code> to consume records from one topic and a <code>KafkaProducer</code> to produce records to another topic. It uses transactions to ensure
+ that there is no duplication or loss of records as they are copied.
+ <p>
+ It is important to handle exceptions and aborted transactions correctly. Any records written by the transational producer will be marked as being part of the transactions, and then when the
+ transaction commits or aborts, transaction marker records are written to indicate the outcome of the transaction. This is how the read-committed consumer does not see records from aborted
+ transactions. However, in the event of a transaction abort, the application's in-memory state and in particular the current position of the consumer must be reset explicitly so that it can
+ reprocess the records processed by the aborted transaction.
+ <p>
+ A simple policy for handling exceptions and aborted transactions is to discard and recreate the Kafka producer and consumer objects and start afresh. As part of recreating the consumer, the consumer
Review Comment:
Yup, I have a task to improve this further once we have KIP-1050 implemented.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17671: Create better documentation for transactions [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #17454:
URL: https://github.com/apache/kafka/pull/17454#discussion_r1795969978
##########
docs/design.html:
##########
@@ -290,24 +290,56 @@ <h3 class="anchor-heading"><a id="semantics" class="anchor-link"></a><a href="#s
messages have a primary key and so the updates are idempotent (receiving the same message twice just overwrites a record with another copy of itself).
</ol>
<p>
- So what about exactly once semantics (i.e. the thing you actually want)? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>
- application), we can leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in a topic, so we can write the offset to Kafka in the
- same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's position will revert to its old value and the produced data on the output topics will not be visible
- to other consumers, depending on their "isolation level." In the default "read_uncommitted" isolation level, all messages are visible to consumers even if they were part of an aborted transaction,
- but in "read_committed," the consumer will only return messages from transactions which were committed (and any messages which were not part of a transaction).
+ So what about exactly-once semantics? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a> application), we can
+ leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in an internal topic, so we can write the offset to Kafka in the
+ same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's stored position will revert to its old value and the produced data on the output topics will not
+ be visible to other consumers, depending on their "isolation level." In the default "read_uncommitted" isolation level, all messages are visible to consumers even if they were part of an aborted transaction,
+ but in "read_committed" isolation level, the consumer will only return messages from transactions which were committed (and any messages which were not part of a transaction).
<p>
When writing to an external system, the limitation is in the need to coordinate the consumer's position with what is actually stored as output. The classic way of achieving this would be to introduce a two-phase
- commit between the storage of the consumer position and the storage of the consumers output. But this can be handled more simply and generally by letting the consumer store its offset in the same place as
+ commit between the storage of the consumer position and the storage of the consumers output. This can be handled more simply and generally by letting the consumer store its offset in the same place as
its output. This is better because many of the output systems a consumer might want to write to will not support a two-phase commit. As an example of this, consider a
<a href="https://kafka.apache.org/documentation/#connect">Kafka Connect</a> connector which populates data in HDFS along with the offsets of the data it reads so that it is guaranteed that either data and
offsets are both updated or neither is. We follow similar patterns for many other data systems which require these stronger semantics and for which the messages do not have a primary key to allow for deduplication.
<p>
- So effectively Kafka supports exactly-once delivery in <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>, and the transactional producer/consumer can be used generally to provide
+ As a result, Kafka supports exactly-once delivery in <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>, and the transactional producer/consumer can be used generally to provide
exactly-once delivery when transferring and processing data between Kafka topics. Exactly-once delivery for other destination systems generally requires cooperation with such systems, but Kafka provides the
offset which makes implementing this feasible (see also <a href="https://kafka.apache.org/documentation/#connect">Kafka Connect</a>). Otherwise, Kafka guarantees at-least-once delivery by default, and allows
the user to implement at-most-once delivery by disabling retries on the producer and committing offsets in the consumer prior to processing a batch of messages.
- <h3 class="anchor-heading"><a id="replication" class="anchor-link"></a><a href="#replication">4.7 Replication</a></h3>
+ <h3 class="anchor-heading"><a id="usingtransactions" class="anchor-link"></a><a href="#usingtransactions">4.7 Using Transactions</a></h3>
+ <p>
+ As mentioned above, the simplest way to get exactly-once semantics from Kafka is to use <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>. However, it is also possible to achieve
+ the same transactional guarantees using the Kafka producer and consumer directly by using them in the same way as Kafka Streams does.
+ <p>
+ Kafka transactions are a bit different than transactions in other messaging systems. In Kafka, the consumer and producer are separate and it is only the producer which is transactional. It is however able to
+ make transactional updates to the consumer's position (confusingly called the "committed offset"), and it is this which gives the overall exactly-once behavior.
+ <p>
+ There are three key aspects to exactly-once processing using the producer and consumer, which match how Kafka Streams works.
+ <ol>
+ <li>The consumer uses partition assignment to ensure that it is the only consumer in the consumer group currently processing each partition.</li>
+ <li>The consumer uses read-committed isolation level to ensure that it does not consume records produced by transactions which aborted.</li>
+ <li>The producer uses transactions so that all of the records it produces, and any offsets it updates on behalf of the consumer, are performed atomically.</li>
+ </ol>
+ <p>
+ The consumer configuration must include <code>isolation.level=read_committed</code> and <code>enable.auto.commit=false</code>. The producer configuration must set <code>transactional.id</code>
Review Comment:
I don't think the consumer necessarily needs to be read_committed unless it is also reading from a topic written via transactions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17671: Create better documentation for transactions [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #17454:
URL: https://github.com/apache/kafka/pull/17454#discussion_r1811160874
##########
docs/design.html:
##########
@@ -290,24 +290,58 @@ <h3 class="anchor-heading"><a id="semantics" class="anchor-link"></a><a href="#s
messages have a primary key and so the updates are idempotent (receiving the same message twice just overwrites a record with another copy of itself).
</ol>
<p>
- So what about exactly once semantics (i.e. the thing you actually want)? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>
- application), we can leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in a topic, so we can write the offset to Kafka in the
- same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's position will revert to its old value and the produced data on the output topics will not be visible
- to other consumers, depending on their "isolation level." In the default "read_uncommitted" isolation level, all messages are visible to consumers even if they were part of an aborted transaction,
- but in "read_committed," the consumer will only return messages from transactions which were committed (and any messages which were not part of a transaction).
+ So what about exactly-once semantics? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a> application), we can
+ leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in an internal topic, so we can write the offset to Kafka in the
+ same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's stored position will revert to its old value (although the consumer has to refetch the
+ committed offset because it does not automatically rewind) and the produced data on the output topics will not be visible to other consumers, depending on their "isolation level". In the default
+ "read_uncommitted" isolation level, all messages are visible to consumers even if they were part of an aborted transaction, but in "read_committed" isolation level, the consumer will only return messages
+ from transactions which were committed (and any messages which were not part of a transaction).
<p>
When writing to an external system, the limitation is in the need to coordinate the consumer's position with what is actually stored as output. The classic way of achieving this would be to introduce a two-phase
- commit between the storage of the consumer position and the storage of the consumers output. But this can be handled more simply and generally by letting the consumer store its offset in the same place as
+ commit between the storage of the consumer position and the storage of the consumers output. This can be handled more simply and generally by letting the consumer store its offset in the same place as
its output. This is better because many of the output systems a consumer might want to write to will not support a two-phase commit. As an example of this, consider a
<a href="https://kafka.apache.org/documentation/#connect">Kafka Connect</a> connector which populates data in HDFS along with the offsets of the data it reads so that it is guaranteed that either data and
offsets are both updated or neither is. We follow similar patterns for many other data systems which require these stronger semantics and for which the messages do not have a primary key to allow for deduplication.
<p>
- So effectively Kafka supports exactly-once delivery in <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>, and the transactional producer/consumer can be used generally to provide
+ As a result, Kafka supports exactly-once delivery in <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>, and the transactional producer/consumer can be used generally to provide
exactly-once delivery when transferring and processing data between Kafka topics. Exactly-once delivery for other destination systems generally requires cooperation with such systems, but Kafka provides the
offset which makes implementing this feasible (see also <a href="https://kafka.apache.org/documentation/#connect">Kafka Connect</a>). Otherwise, Kafka guarantees at-least-once delivery by default, and allows
the user to implement at-most-once delivery by disabling retries on the producer and committing offsets in the consumer prior to processing a batch of messages.
- <h3 class="anchor-heading"><a id="replication" class="anchor-link"></a><a href="#replication">4.7 Replication</a></h3>
+ <h3 class="anchor-heading"><a id="usingtransactions" class="anchor-link"></a><a href="#usingtransactions">4.7 Using Transactions</a></h3>
+ <p>
+ As mentioned above, the simplest way to get exactly-once semantics from Kafka is to use <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>. However, it is also possible to achieve
+ the same transactional guarantees using the Kafka producer and consumer directly by using them in the same way as Kafka Streams does.
+ <p>
+ Kafka transactions are a bit different than transactions in other messaging systems. In Kafka, the consumer and producer are separate and it is only the producer which is transactional. It is however able to
+ make transactional updates to the consumer's position (confusingly called the "committed offset"), and it is this which gives the overall exactly-once behavior.
+ <p>
+ There are three key aspects to exactly-once processing using the producer and consumer, which match how Kafka Streams works.
+ <ol>
+ <li>The consumer uses partition assignment to ensure that it is the only consumer in the consumer group currently processing each partition.</li>
+ <li>The consumer uses read-committed isolation level to ensure that it does not consume records produced by transactions which aborted.</li>
+ <li>The producer uses transactions so that all of the records it produces, and any offsets it updates on behalf of the consumer, are performed atomically.</li>
+ </ol>
+ <p>
+ The consumer configuration must include <code>isolation.level=read_committed</code> and <code>enable.auto.commit=false</code>. The producer configuration must set <code>transactional.id</code>
+ to the name of the transactional ID to be used, which configures the producer for transactional delivery and also makes sure that a restarted application causes any in-flight transaction from
+ the previous instance to abort. Only the producer has the <code>transactional.id</code> configuration. Strictly speaking, the consumer doesn't have to use read_committed isolation level, but if
+ it does not, it will see records from aborted transactions and also open transactions which have not yet completed, which seems undesirable if trying to achieve exactly-once.
+ <p>
+ Here's an example of a <a href="https://github.com/apache/kafka/blob/trunk/tools/src/main/java/org/apache/kafka/tools/TransactionalMessageCopier.java">transactional message copier</a>
Review Comment:
Indeed. There's a better example which is in a draft PR as part of KIP-1050. I'd prefer to refer to that, once it's ready to go.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17671: Create better documentation for transactions [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17454:
URL: https://github.com/apache/kafka/pull/17454#discussion_r1813999867
##########
docs/design.html:
##########
@@ -254,32 +254,32 @@ <h3 class="anchor-heading"><a id="semantics" class="anchor-link"></a><a href="#s
<i>At least once</i>—Messages are never lost but may be redelivered.
</li>
<li>
- <i>Exactly once</i>—this is what people actually want, each message is delivered once and only once.
+ <i>Exactly once</i>—Each message is delivered once and only once.
</li>
</ul>
It's worth noting that this breaks down into two problems: the durability guarantees for publishing a message and the guarantees when consuming a message.
<p>
- Many systems claim to provide "exactly once" delivery semantics, but it is important to read the fine print, most of these claims are misleading (i.e. they don't translate to the case where consumers or producers
+ Many systems claim to provide "exactly-once" delivery semantics, but it is important to read the fine print, because sometimes these claims are misleading (i.e. they don't translate to the case where consumers or producers
can fail, cases where there are multiple consumer processes, or cases where data written to disk can be lost).
<p>
- Kafka's semantics are straight-forward. When publishing a message we have a notion of the message being "committed" to the log. Once a published message is committed it will not be lost as long as one broker that
- replicates the partition to which this message was written remains "alive". The definition of committed message, alive partition as well as a description of which types of failures we attempt to handle will be
+ Kafka's semantics are straightforward. When publishing a message we have a notion of the message being "committed" to the log. Once a published message is committed, it will not be lost as long as one broker that
+ replicates the partition to which this message was written remains "alive". The definition of committed message and alive partition as well as a description of which types of failures we attempt to handle will be
described in more detail in the next section. For now let's assume a perfect, lossless broker and try to understand the guarantees to the producer and consumer. If a producer attempts to publish a message and
- experiences a network error it cannot be sure if this error happened before or after the message was committed. This is similar to the semantics of inserting into a database table with an autogenerated key.
+ experiences a network error, it cannot be sure if this error happened before or after the message was committed. This is similar to the semantics of inserting into a database table with an autogenerated key.
<p>
Prior to 0.11.0.0, if a producer failed to receive a response indicating that a message was committed, it had little choice but to resend the message. This provides at-least-once delivery semantics since the
message may be written to the log again during resending if the original request had in fact succeeded. Since 0.11.0.0, the Kafka producer also supports an idempotent delivery option which guarantees that resending
will not result in duplicate entries in the log. To achieve this, the broker assigns each producer an ID and deduplicates messages using a sequence number that is sent by the producer along with every message.
- Also beginning with 0.11.0.0, the producer supports the ability to send messages to multiple topic partitions using transaction-like semantics: i.e. either all messages are successfully written or none of them are.
+ Also beginning with 0.11.0.0, the producer supports the ability to send messages to multiple topic partitions using transactional semantics, so that either all messages are successfully written or none of them are.
The main use case for this is exactly-once processing between Kafka topics (described below).
<p>
- Not all use cases require such strong guarantees. For uses which are latency sensitive we allow the producer to specify the durability level it desires. If the producer specifies that it wants to wait on the message
- being committed this can take on the order of 10 ms. However the producer can also specify that it wants to perform the send completely asynchronously or that it wants to wait only until the leader (but not
+ Not all use cases require such strong guarantees. For uses which are latency-sensitive, we allow the producer to specify the durability level it desires. If the producer specifies that it wants to wait on the message
Review Comment:
I think we should fix this typo? `For uses which are latency-sensitive` is not an english sentence from what I can tell.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17671: Create better documentation for transactions [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #17454:
URL: https://github.com/apache/kafka/pull/17454#discussion_r1812189202
##########
docs/design.html:
##########
@@ -290,24 +290,58 @@ <h3 class="anchor-heading"><a id="semantics" class="anchor-link"></a><a href="#s
messages have a primary key and so the updates are idempotent (receiving the same message twice just overwrites a record with another copy of itself).
</ol>
<p>
- So what about exactly once semantics (i.e. the thing you actually want)? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>
- application), we can leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in a topic, so we can write the offset to Kafka in the
- same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's position will revert to its old value and the produced data on the output topics will not be visible
- to other consumers, depending on their "isolation level." In the default "read_uncommitted" isolation level, all messages are visible to consumers even if they were part of an aborted transaction,
- but in "read_committed," the consumer will only return messages from transactions which were committed (and any messages which were not part of a transaction).
+ So what about exactly-once semantics? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a> application), we can
+ leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in an internal topic, so we can write the offset to Kafka in the
+ same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's stored position will revert to its old value (although the consumer has to refetch the
+ committed offset because it does not automatically rewind) and the produced data on the output topics will not be visible to other consumers, depending on their "isolation level". In the default
+ "read_uncommitted" isolation level, all messages are visible to consumers even if they were part of an aborted transaction, but in "read_committed" isolation level, the consumer will only return messages
+ from transactions which were committed (and any messages which were not part of a transaction).
<p>
When writing to an external system, the limitation is in the need to coordinate the consumer's position with what is actually stored as output. The classic way of achieving this would be to introduce a two-phase
- commit between the storage of the consumer position and the storage of the consumers output. But this can be handled more simply and generally by letting the consumer store its offset in the same place as
+ commit between the storage of the consumer position and the storage of the consumers output. This can be handled more simply and generally by letting the consumer store its offset in the same place as
its output. This is better because many of the output systems a consumer might want to write to will not support a two-phase commit. As an example of this, consider a
<a href="https://kafka.apache.org/documentation/#connect">Kafka Connect</a> connector which populates data in HDFS along with the offsets of the data it reads so that it is guaranteed that either data and
offsets are both updated or neither is. We follow similar patterns for many other data systems which require these stronger semantics and for which the messages do not have a primary key to allow for deduplication.
<p>
- So effectively Kafka supports exactly-once delivery in <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>, and the transactional producer/consumer can be used generally to provide
+ As a result, Kafka supports exactly-once delivery in <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>, and the transactional producer/consumer can be used generally to provide
exactly-once delivery when transferring and processing data between Kafka topics. Exactly-once delivery for other destination systems generally requires cooperation with such systems, but Kafka provides the
offset which makes implementing this feasible (see also <a href="https://kafka.apache.org/documentation/#connect">Kafka Connect</a>). Otherwise, Kafka guarantees at-least-once delivery by default, and allows
the user to implement at-most-once delivery by disabling retries on the producer and committing offsets in the consumer prior to processing a batch of messages.
- <h3 class="anchor-heading"><a id="replication" class="anchor-link"></a><a href="#replication">4.7 Replication</a></h3>
+ <h3 class="anchor-heading"><a id="usingtransactions" class="anchor-link"></a><a href="#usingtransactions">4.7 Using Transactions</a></h3>
+ <p>
+ As mentioned above, the simplest way to get exactly-once semantics from Kafka is to use <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>. However, it is also possible to achieve
+ the same transactional guarantees using the Kafka producer and consumer directly by using them in the same way as Kafka Streams does.
+ <p>
+ Kafka transactions are a bit different than transactions in other messaging systems. In Kafka, the consumer and producer are separate and it is only the producer which is transactional. It is however able to
+ make transactional updates to the consumer's position (confusingly called the "committed offset"), and it is this which gives the overall exactly-once behavior.
+ <p>
+ There are three key aspects to exactly-once processing using the producer and consumer, which match how Kafka Streams works.
+ <ol>
+ <li>The consumer uses partition assignment to ensure that it is the only consumer in the consumer group currently processing each partition.</li>
+ <li>The consumer uses read-committed isolation level to ensure that it does not consume records produced by transactions which aborted.</li>
+ <li>The producer uses transactions so that all of the records it produces, and any offsets it updates on behalf of the consumer, are performed atomically.</li>
+ </ol>
+ <p>
+ The consumer configuration must include <code>isolation.level=read_committed</code> and <code>enable.auto.commit=false</code>. The producer configuration must set <code>transactional.id</code>
+ to the name of the transactional ID to be used, which configures the producer for transactional delivery and also makes sure that a restarted application causes any in-flight transaction from
+ the previous instance to abort. Only the producer has the <code>transactional.id</code> configuration. Strictly speaking, the consumer doesn't have to use read_committed isolation level, but if
Review Comment:
OK, I've added another couple of sentences for your delectation. Let me know what you think.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17671: Create better documentation for transactions [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #17454:
URL: https://github.com/apache/kafka/pull/17454#discussion_r1795970930
##########
docs/design.html:
##########
@@ -290,24 +290,56 @@ <h3 class="anchor-heading"><a id="semantics" class="anchor-link"></a><a href="#s
messages have a primary key and so the updates are idempotent (receiving the same message twice just overwrites a record with another copy of itself).
</ol>
<p>
- So what about exactly once semantics (i.e. the thing you actually want)? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>
- application), we can leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in a topic, so we can write the offset to Kafka in the
- same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's position will revert to its old value and the produced data on the output topics will not be visible
- to other consumers, depending on their "isolation level." In the default "read_uncommitted" isolation level, all messages are visible to consumers even if they were part of an aborted transaction,
- but in "read_committed," the consumer will only return messages from transactions which were committed (and any messages which were not part of a transaction).
+ So what about exactly-once semantics? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a> application), we can
+ leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in an internal topic, so we can write the offset to Kafka in the
+ same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's stored position will revert to its old value and the produced data on the output topics will not
+ be visible to other consumers, depending on their "isolation level." In the default "read_uncommitted" isolation level, all messages are visible to consumers even if they were part of an aborted transaction,
+ but in "read_committed" isolation level, the consumer will only return messages from transactions which were committed (and any messages which were not part of a transaction).
<p>
When writing to an external system, the limitation is in the need to coordinate the consumer's position with what is actually stored as output. The classic way of achieving this would be to introduce a two-phase
- commit between the storage of the consumer position and the storage of the consumers output. But this can be handled more simply and generally by letting the consumer store its offset in the same place as
+ commit between the storage of the consumer position and the storage of the consumers output. This can be handled more simply and generally by letting the consumer store its offset in the same place as
its output. This is better because many of the output systems a consumer might want to write to will not support a two-phase commit. As an example of this, consider a
<a href="https://kafka.apache.org/documentation/#connect">Kafka Connect</a> connector which populates data in HDFS along with the offsets of the data it reads so that it is guaranteed that either data and
offsets are both updated or neither is. We follow similar patterns for many other data systems which require these stronger semantics and for which the messages do not have a primary key to allow for deduplication.
<p>
- So effectively Kafka supports exactly-once delivery in <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>, and the transactional producer/consumer can be used generally to provide
+ As a result, Kafka supports exactly-once delivery in <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>, and the transactional producer/consumer can be used generally to provide
exactly-once delivery when transferring and processing data between Kafka topics. Exactly-once delivery for other destination systems generally requires cooperation with such systems, but Kafka provides the
offset which makes implementing this feasible (see also <a href="https://kafka.apache.org/documentation/#connect">Kafka Connect</a>). Otherwise, Kafka guarantees at-least-once delivery by default, and allows
the user to implement at-most-once delivery by disabling retries on the producer and committing offsets in the consumer prior to processing a batch of messages.
- <h3 class="anchor-heading"><a id="replication" class="anchor-link"></a><a href="#replication">4.7 Replication</a></h3>
+ <h3 class="anchor-heading"><a id="usingtransactions" class="anchor-link"></a><a href="#usingtransactions">4.7 Using Transactions</a></h3>
+ <p>
+ As mentioned above, the simplest way to get exactly-once semantics from Kafka is to use <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>. However, it is also possible to achieve
+ the same transactional guarantees using the Kafka producer and consumer directly by using them in the same way as Kafka Streams does.
+ <p>
+ Kafka transactions are a bit different than transactions in other messaging systems. In Kafka, the consumer and producer are separate and it is only the producer which is transactional. It is however able to
+ make transactional updates to the consumer's position (confusingly called the "committed offset"), and it is this which gives the overall exactly-once behavior.
+ <p>
+ There are three key aspects to exactly-once processing using the producer and consumer, which match how Kafka Streams works.
+ <ol>
+ <li>The consumer uses partition assignment to ensure that it is the only consumer in the consumer group currently processing each partition.</li>
+ <li>The consumer uses read-committed isolation level to ensure that it does not consume records produced by transactions which aborted.</li>
+ <li>The producer uses transactions so that all of the records it produces, and any offsets it updates on behalf of the consumer, are performed atomically.</li>
+ </ol>
+ <p>
+ The consumer configuration must include <code>isolation.level=read_committed</code> and <code>enable.auto.commit=false</code>. The producer configuration must set <code>transactional.id</code>
+ to the name of the transactional ID to be used, which configures the producer for transactional delivery and also makes sure that a restarted application causes any in-flight transaction from
+ the previous instance to abort. Only the producer has the <code>transactional.id</code> configuration.
+ <p>
+ Here's an example of a <a href="https://github.com/apache/kafka/blob/trunk/tools/src/main/java/org/apache/kafka/tools/TransactionalMessageCopier.java">transactional message copier</a>
+ which uses these principles. It uses a <code>KafkaConsumer</code> to consume records from one topic and a <code>KafkaProducer</code> to produce records to another topic. It uses transactions to ensure
+ that there is no duplication or loss of records as they are copied.
+ <p>
+ It is important to handle exceptions and aborted transactions correctly. Any records written by the transational producer will be marked as being part of the transactions, and then when the
+ transaction commits or aborts, transaction marker records are written to indicate the outcome of the transaction. This is how the read-committed consumer does not see records from aborted
+ transactions. However, in the event of a transaction abort, the application's in-memory state and in particular the current position of the consumer must be reset explicitly so that it can
+ reprocess the records processed by the aborted transaction.
+ <p>
+ A simple policy for handling exceptions and aborted transactions is to discard and recreate the Kafka producer and consumer objects and start afresh. As part of recreating the consumer, the consumer
Review Comment:
There are cases where we can safely abort without discarding the producer. KIP-1050 will help make the cases clearer.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17671: Create better documentation for transactions [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #17454:
URL: https://github.com/apache/kafka/pull/17454#discussion_r1811157856
##########
docs/design.html:
##########
@@ -254,32 +254,32 @@ <h3 class="anchor-heading"><a id="semantics" class="anchor-link"></a><a href="#s
<i>At least once</i>—Messages are never lost but may be redelivered.
</li>
<li>
- <i>Exactly once</i>—this is what people actually want, each message is delivered once and only once.
+ <i>Exactly once</i>—Each message is delivered once and only once.
</li>
</ul>
It's worth noting that this breaks down into two problems: the durability guarantees for publishing a message and the guarantees when consuming a message.
<p>
- Many systems claim to provide "exactly once" delivery semantics, but it is important to read the fine print, most of these claims are misleading (i.e. they don't translate to the case where consumers or producers
+ Many systems claim to provide "exactly-once" delivery semantics, but it is important to read the fine print, because sometimes these claims are misleading (i.e. they don't translate to the case where consumers or producers
can fail, cases where there are multiple consumer processes, or cases where data written to disk can be lost).
<p>
- Kafka's semantics are straight-forward. When publishing a message we have a notion of the message being "committed" to the log. Once a published message is committed it will not be lost as long as one broker that
- replicates the partition to which this message was written remains "alive". The definition of committed message, alive partition as well as a description of which types of failures we attempt to handle will be
+ Kafka's semantics are straightforward. When publishing a message we have a notion of the message being "committed" to the log. Once a published message is committed, it will not be lost as long as one broker that
+ replicates the partition to which this message was written remains "alive". The definition of committed message and alive partition as well as a description of which types of failures we attempt to handle will be
described in more detail in the next section. For now let's assume a perfect, lossless broker and try to understand the guarantees to the producer and consumer. If a producer attempts to publish a message and
- experiences a network error it cannot be sure if this error happened before or after the message was committed. This is similar to the semantics of inserting into a database table with an autogenerated key.
+ experiences a network error, it cannot be sure if this error happened before or after the message was committed. This is similar to the semantics of inserting into a database table with an autogenerated key.
<p>
Prior to 0.11.0.0, if a producer failed to receive a response indicating that a message was committed, it had little choice but to resend the message. This provides at-least-once delivery semantics since the
message may be written to the log again during resending if the original request had in fact succeeded. Since 0.11.0.0, the Kafka producer also supports an idempotent delivery option which guarantees that resending
will not result in duplicate entries in the log. To achieve this, the broker assigns each producer an ID and deduplicates messages using a sequence number that is sent by the producer along with every message.
- Also beginning with 0.11.0.0, the producer supports the ability to send messages to multiple topic partitions using transaction-like semantics: i.e. either all messages are successfully written or none of them are.
+ Also beginning with 0.11.0.0, the producer supports the ability to send messages to multiple topic partitions using transactional semantics, so that either all messages are successfully written or none of them are.
The main use case for this is exactly-once processing between Kafka topics (described below).
<p>
- Not all use cases require such strong guarantees. For uses which are latency sensitive we allow the producer to specify the durability level it desires. If the producer specifies that it wants to wait on the message
- being committed this can take on the order of 10 ms. However the producer can also specify that it wants to perform the send completely asynchronously or that it wants to wait only until the leader (but not
+ Not all use cases require such strong guarantees. For uses which are latency-sensitive, we allow the producer to specify the durability level it desires. If the producer specifies that it wants to wait on the message
+ being committed, this can take on the order of 10 ms. However the producer can also specify that it wants to perform the send completely asynchronously or that it wants to wait only until the leader (but not
necessarily the followers) have the message.
Review Comment:
You are probably correct but you are reviewing the existing text, not the changes. It could do with a fair comprehensive re-write I think. I wasn't expecting to do that now. For example, it's hardly relevant nowadays what the 0.11.0.0 release introduced :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17671: Create better documentation for transactions [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #17454:
URL: https://github.com/apache/kafka/pull/17454#discussion_r1811146922
##########
docs/design.html:
##########
@@ -254,32 +254,32 @@ <h3 class="anchor-heading"><a id="semantics" class="anchor-link"></a><a href="#s
<i>At least once</i>—Messages are never lost but may be redelivered.
</li>
<li>
- <i>Exactly once</i>—this is what people actually want, each message is delivered once and only once.
+ <i>Exactly once</i>—Each message is delivered once and only once.
</li>
</ul>
It's worth noting that this breaks down into two problems: the durability guarantees for publishing a message and the guarantees when consuming a message.
<p>
- Many systems claim to provide "exactly once" delivery semantics, but it is important to read the fine print, most of these claims are misleading (i.e. they don't translate to the case where consumers or producers
+ Many systems claim to provide "exactly-once" delivery semantics, but it is important to read the fine print, because sometimes these claims are misleading (i.e. they don't translate to the case where consumers or producers
can fail, cases where there are multiple consumer processes, or cases where data written to disk can be lost).
<p>
- Kafka's semantics are straight-forward. When publishing a message we have a notion of the message being "committed" to the log. Once a published message is committed it will not be lost as long as one broker that
- replicates the partition to which this message was written remains "alive". The definition of committed message, alive partition as well as a description of which types of failures we attempt to handle will be
+ Kafka's semantics are straightforward. When publishing a message we have a notion of the message being "committed" to the log. Once a published message is committed, it will not be lost as long as one broker that
Review Comment:
This term "committed" is exactly the same as used in the reliability guarantees section of Kafka: The Definitive Guide. I'm not inclined to change it as a result.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17671: Create better documentation for transactions [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #17454:
URL: https://github.com/apache/kafka/pull/17454#discussion_r1796713363
##########
docs/design.html:
##########
@@ -290,24 +290,56 @@ <h3 class="anchor-heading"><a id="semantics" class="anchor-link"></a><a href="#s
messages have a primary key and so the updates are idempotent (receiving the same message twice just overwrites a record with another copy of itself).
</ol>
<p>
- So what about exactly once semantics (i.e. the thing you actually want)? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>
- application), we can leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in a topic, so we can write the offset to Kafka in the
- same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's position will revert to its old value and the produced data on the output topics will not be visible
- to other consumers, depending on their "isolation level." In the default "read_uncommitted" isolation level, all messages are visible to consumers even if they were part of an aborted transaction,
- but in "read_committed," the consumer will only return messages from transactions which were committed (and any messages which were not part of a transaction).
+ So what about exactly-once semantics? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a> application), we can
+ leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in an internal topic, so we can write the offset to Kafka in the
+ same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's stored position will revert to its old value and the produced data on the output topics will not
+ be visible to other consumers, depending on their "isolation level." In the default "read_uncommitted" isolation level, all messages are visible to consumers even if they were part of an aborted transaction,
+ but in "read_committed" isolation level, the consumer will only return messages from transactions which were committed (and any messages which were not part of a transaction).
<p>
When writing to an external system, the limitation is in the need to coordinate the consumer's position with what is actually stored as output. The classic way of achieving this would be to introduce a two-phase
- commit between the storage of the consumer position and the storage of the consumers output. But this can be handled more simply and generally by letting the consumer store its offset in the same place as
+ commit between the storage of the consumer position and the storage of the consumers output. This can be handled more simply and generally by letting the consumer store its offset in the same place as
its output. This is better because many of the output systems a consumer might want to write to will not support a two-phase commit. As an example of this, consider a
<a href="https://kafka.apache.org/documentation/#connect">Kafka Connect</a> connector which populates data in HDFS along with the offsets of the data it reads so that it is guaranteed that either data and
offsets are both updated or neither is. We follow similar patterns for many other data systems which require these stronger semantics and for which the messages do not have a primary key to allow for deduplication.
<p>
- So effectively Kafka supports exactly-once delivery in <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>, and the transactional producer/consumer can be used generally to provide
+ As a result, Kafka supports exactly-once delivery in <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>, and the transactional producer/consumer can be used generally to provide
exactly-once delivery when transferring and processing data between Kafka topics. Exactly-once delivery for other destination systems generally requires cooperation with such systems, but Kafka provides the
offset which makes implementing this feasible (see also <a href="https://kafka.apache.org/documentation/#connect">Kafka Connect</a>). Otherwise, Kafka guarantees at-least-once delivery by default, and allows
the user to implement at-most-once delivery by disabling retries on the producer and committing offsets in the consumer prior to processing a batch of messages.
- <h3 class="anchor-heading"><a id="replication" class="anchor-link"></a><a href="#replication">4.7 Replication</a></h3>
+ <h3 class="anchor-heading"><a id="usingtransactions" class="anchor-link"></a><a href="#usingtransactions">4.7 Using Transactions</a></h3>
+ <p>
+ As mentioned above, the simplest way to get exactly-once semantics from Kafka is to use <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>. However, it is also possible to achieve
+ the same transactional guarantees using the Kafka producer and consumer directly by using them in the same way as Kafka Streams does.
+ <p>
+ Kafka transactions are a bit different than transactions in other messaging systems. In Kafka, the consumer and producer are separate and it is only the producer which is transactional. It is however able to
+ make transactional updates to the consumer's position (confusingly called the "committed offset"), and it is this which gives the overall exactly-once behavior.
+ <p>
+ There are three key aspects to exactly-once processing using the producer and consumer, which match how Kafka Streams works.
+ <ol>
+ <li>The consumer uses partition assignment to ensure that it is the only consumer in the consumer group currently processing each partition.</li>
+ <li>The consumer uses read-committed isolation level to ensure that it does not consume records produced by transactions which aborted.</li>
+ <li>The producer uses transactions so that all of the records it produces, and any offsets it updates on behalf of the consumer, are performed atomically.</li>
+ </ol>
+ <p>
+ The consumer configuration must include <code>isolation.level=read_committed</code> and <code>enable.auto.commit=false</code>. The producer configuration must set <code>transactional.id</code>
+ to the name of the transactional ID to be used, which configures the producer for transactional delivery and also makes sure that a restarted application causes any in-flight transaction from
+ the previous instance to abort. Only the producer has the <code>transactional.id</code> configuration.
+ <p>
+ Here's an example of a <a href="https://github.com/apache/kafka/blob/trunk/tools/src/main/java/org/apache/kafka/tools/TransactionalMessageCopier.java">transactional message copier</a>
+ which uses these principles. It uses a <code>KafkaConsumer</code> to consume records from one topic and a <code>KafkaProducer</code> to produce records to another topic. It uses transactions to ensure
+ that there is no duplication or loss of records as they are copied.
+ <p>
+ It is important to handle exceptions and aborted transactions correctly. Any records written by the transational producer will be marked as being part of the transactions, and then when the
+ transaction commits or aborts, transaction marker records are written to indicate the outcome of the transaction. This is how the read-committed consumer does not see records from aborted
+ transactions. However, in the event of a transaction abort, the application's in-memory state and in particular the current position of the consumer must be reset explicitly so that it can
+ reprocess the records processed by the aborted transaction.
+ <p>
+ A simple policy for handling exceptions and aborted transactions is to discard and recreate the Kafka producer and consumer objects and start afresh. As part of recreating the consumer, the consumer
Review Comment:
Indeed. That KIP is still not adopted, and I think we would update this documentation once it's there. It certainly helps.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17671: Create better documentation for transactions [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #17454:
URL: https://github.com/apache/kafka/pull/17454#discussion_r1795963676
##########
docs/design.html:
##########
@@ -290,24 +290,56 @@ <h3 class="anchor-heading"><a id="semantics" class="anchor-link"></a><a href="#s
messages have a primary key and so the updates are idempotent (receiving the same message twice just overwrites a record with another copy of itself).
</ol>
<p>
- So what about exactly once semantics (i.e. the thing you actually want)? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>
- application), we can leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in a topic, so we can write the offset to Kafka in the
- same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's position will revert to its old value and the produced data on the output topics will not be visible
- to other consumers, depending on their "isolation level." In the default "read_uncommitted" isolation level, all messages are visible to consumers even if they were part of an aborted transaction,
- but in "read_committed," the consumer will only return messages from transactions which were committed (and any messages which were not part of a transaction).
+ So what about exactly-once semantics? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a> application), we can
+ leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in an internal topic, so we can write the offset to Kafka in the
+ same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's stored position will revert to its old value and the produced data on the output topics will not
Review Comment:
Did we want to change this line because it is misleading -- the position on the consumer doesn't revert unless the offsets are fetched again.
It is also a little misleading since the offset doesn't really revert in the coordinator -- it is just pending until the transaction is completed (at least as of KIP-447, which may not have been implemented when this was written)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17671: Create better documentation for transactions [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on PR #17454:
URL: https://github.com/apache/kafka/pull/17454#issuecomment-2429996766
https://issues.apache.org/jira/browse/KAFKA-17855 is the continuation of this work which depends upon KIP-1050.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17671: Create better documentation for transactions [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #17454:
URL: https://github.com/apache/kafka/pull/17454#discussion_r1907662987
##########
docs/design.html:
##########
@@ -290,24 +290,58 @@ <h3 class="anchor-heading"><a id="semantics" class="anchor-link"></a><a href="#s
messages have a primary key and so the updates are idempotent (receiving the same message twice just overwrites a record with another copy of itself).
</ol>
<p>
- So what about exactly once semantics (i.e. the thing you actually want)? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>
- application), we can leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in a topic, so we can write the offset to Kafka in the
- same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's position will revert to its old value and the produced data on the output topics will not be visible
- to other consumers, depending on their "isolation level." In the default "read_uncommitted" isolation level, all messages are visible to consumers even if they were part of an aborted transaction,
- but in "read_committed," the consumer will only return messages from transactions which were committed (and any messages which were not part of a transaction).
+ So what about exactly-once semantics? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a> application), we can
+ leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in an internal topic, so we can write the offset to Kafka in the
+ same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's stored position will revert to its old value (although the consumer has to refetch the
+ committed offset because it does not automatically rewind) and the produced data on the output topics will not be visible to other consumers, depending on their "isolation level". In the default
+ "read_uncommitted" isolation level, all messages are visible to consumers even if they were part of an aborted transaction, but in "read_committed" isolation level, the consumer will only return messages
+ from transactions which were committed (and any messages which were not part of a transaction).
<p>
When writing to an external system, the limitation is in the need to coordinate the consumer's position with what is actually stored as output. The classic way of achieving this would be to introduce a two-phase
- commit between the storage of the consumer position and the storage of the consumers output. But this can be handled more simply and generally by letting the consumer store its offset in the same place as
+ commit between the storage of the consumer position and the storage of the consumers output. This can be handled more simply and generally by letting the consumer store its offset in the same place as
its output. This is better because many of the output systems a consumer might want to write to will not support a two-phase commit. As an example of this, consider a
<a href="https://kafka.apache.org/documentation/#connect">Kafka Connect</a> connector which populates data in HDFS along with the offsets of the data it reads so that it is guaranteed that either data and
offsets are both updated or neither is. We follow similar patterns for many other data systems which require these stronger semantics and for which the messages do not have a primary key to allow for deduplication.
<p>
- So effectively Kafka supports exactly-once delivery in <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>, and the transactional producer/consumer can be used generally to provide
+ As a result, Kafka supports exactly-once delivery in <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>, and the transactional producer/consumer can be used generally to provide
Review Comment:
+1 to the suggestion
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17671: Create better documentation for transactions [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #17454:
URL: https://github.com/apache/kafka/pull/17454#discussion_r1796735577
##########
docs/design.html:
##########
@@ -290,24 +290,56 @@ <h3 class="anchor-heading"><a id="semantics" class="anchor-link"></a><a href="#s
messages have a primary key and so the updates are idempotent (receiving the same message twice just overwrites a record with another copy of itself).
</ol>
<p>
- So what about exactly once semantics (i.e. the thing you actually want)? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>
- application), we can leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in a topic, so we can write the offset to Kafka in the
- same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's position will revert to its old value and the produced data on the output topics will not be visible
- to other consumers, depending on their "isolation level." In the default "read_uncommitted" isolation level, all messages are visible to consumers even if they were part of an aborted transaction,
- but in "read_committed," the consumer will only return messages from transactions which were committed (and any messages which were not part of a transaction).
+ So what about exactly-once semantics? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a> application), we can
+ leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in an internal topic, so we can write the offset to Kafka in the
+ same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's stored position will revert to its old value and the produced data on the output topics will not
Review Comment:
I really made very little change to this in the first draft. I'll try to tighten it up and see what you think.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17671: Create better documentation for transactions [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #17454:
URL: https://github.com/apache/kafka/pull/17454#discussion_r1796727208
##########
docs/design.html:
##########
@@ -160,7 +160,7 @@ <h4 class="anchor-heading"><a id="design_asyncsend" class="anchor-link"></a><a h
to accumulate no more than a fixed number of messages and to wait no longer than some fixed latency bound (say 64k or 10 ms). This allows the accumulation of more bytes to send, and few larger I/O operations on the
servers. This buffering is configurable and gives a mechanism to trade off a small amount of additional latency for better throughput.
<p>
- Details on <a href="#producerconfigs">configuration</a> and the <a href="https://kafka.apache.org/{{version}}/javadoc/org/apache/kafka/clients/producer/KafkaProducer.html">api</a> for the producer can be found
+ Details on <a href="#producerconfigs">configuration</a> and the <a href="https://kafka.apache.org/082/javadoc/index.html?org/apache/kafka/clients/producer/KafkaProducer.html">api</a> for the producer can be found
Review Comment:
You are entirely correct. It's a merge mistake, that I'll fix immediately.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17671: Create better documentation for transactions [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #17454:
URL: https://github.com/apache/kafka/pull/17454#discussion_r1796742994
##########
docs/design.html:
##########
@@ -290,24 +290,56 @@ <h3 class="anchor-heading"><a id="semantics" class="anchor-link"></a><a href="#s
messages have a primary key and so the updates are idempotent (receiving the same message twice just overwrites a record with another copy of itself).
</ol>
<p>
- So what about exactly once semantics (i.e. the thing you actually want)? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>
- application), we can leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in a topic, so we can write the offset to Kafka in the
- same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's position will revert to its old value and the produced data on the output topics will not be visible
- to other consumers, depending on their "isolation level." In the default "read_uncommitted" isolation level, all messages are visible to consumers even if they were part of an aborted transaction,
- but in "read_committed," the consumer will only return messages from transactions which were committed (and any messages which were not part of a transaction).
+ So what about exactly-once semantics? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a> application), we can
+ leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in an internal topic, so we can write the offset to Kafka in the
+ same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's stored position will revert to its old value and the produced data on the output topics will not
+ be visible to other consumers, depending on their "isolation level." In the default "read_uncommitted" isolation level, all messages are visible to consumers even if they were part of an aborted transaction,
+ but in "read_committed" isolation level, the consumer will only return messages from transactions which were committed (and any messages which were not part of a transaction).
<p>
When writing to an external system, the limitation is in the need to coordinate the consumer's position with what is actually stored as output. The classic way of achieving this would be to introduce a two-phase
- commit between the storage of the consumer position and the storage of the consumers output. But this can be handled more simply and generally by letting the consumer store its offset in the same place as
+ commit between the storage of the consumer position and the storage of the consumers output. This can be handled more simply and generally by letting the consumer store its offset in the same place as
its output. This is better because many of the output systems a consumer might want to write to will not support a two-phase commit. As an example of this, consider a
<a href="https://kafka.apache.org/documentation/#connect">Kafka Connect</a> connector which populates data in HDFS along with the offsets of the data it reads so that it is guaranteed that either data and
offsets are both updated or neither is. We follow similar patterns for many other data systems which require these stronger semantics and for which the messages do not have a primary key to allow for deduplication.
<p>
- So effectively Kafka supports exactly-once delivery in <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>, and the transactional producer/consumer can be used generally to provide
+ As a result, Kafka supports exactly-once delivery in <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>, and the transactional producer/consumer can be used generally to provide
exactly-once delivery when transferring and processing data between Kafka topics. Exactly-once delivery for other destination systems generally requires cooperation with such systems, but Kafka provides the
offset which makes implementing this feasible (see also <a href="https://kafka.apache.org/documentation/#connect">Kafka Connect</a>). Otherwise, Kafka guarantees at-least-once delivery by default, and allows
the user to implement at-most-once delivery by disabling retries on the producer and committing offsets in the consumer prior to processing a batch of messages.
- <h3 class="anchor-heading"><a id="replication" class="anchor-link"></a><a href="#replication">4.7 Replication</a></h3>
+ <h3 class="anchor-heading"><a id="usingtransactions" class="anchor-link"></a><a href="#usingtransactions">4.7 Using Transactions</a></h3>
+ <p>
+ As mentioned above, the simplest way to get exactly-once semantics from Kafka is to use <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>. However, it is also possible to achieve
+ the same transactional guarantees using the Kafka producer and consumer directly by using them in the same way as Kafka Streams does.
+ <p>
+ Kafka transactions are a bit different than transactions in other messaging systems. In Kafka, the consumer and producer are separate and it is only the producer which is transactional. It is however able to
+ make transactional updates to the consumer's position (confusingly called the "committed offset"), and it is this which gives the overall exactly-once behavior.
+ <p>
+ There are three key aspects to exactly-once processing using the producer and consumer, which match how Kafka Streams works.
+ <ol>
+ <li>The consumer uses partition assignment to ensure that it is the only consumer in the consumer group currently processing each partition.</li>
+ <li>The consumer uses read-committed isolation level to ensure that it does not consume records produced by transactions which aborted.</li>
+ <li>The producer uses transactions so that all of the records it produces, and any offsets it updates on behalf of the consumer, are performed atomically.</li>
+ </ol>
+ <p>
+ The consumer configuration must include <code>isolation.level=read_committed</code> and <code>enable.auto.commit=false</code>. The producer configuration must set <code>transactional.id</code>
Review Comment:
Yes, that's true but it's a weird transactional application that wants to process aborted data.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17671: Create better documentation for transactions [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on PR #17454:
URL: https://github.com/apache/kafka/pull/17454#issuecomment-2415364502
I think most of the documentation makes sense to me. I think my only remaining open question is whether we want to enumerate read committed consumers as a requirement for EOS as per the comment here: https://github.com/apache/kafka/pull/17454#discussion_r1801587174
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17671: Create better documentation for transactions [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #17454:
URL: https://github.com/apache/kafka/pull/17454#discussion_r1801587174
##########
docs/design.html:
##########
@@ -290,24 +290,56 @@ <h3 class="anchor-heading"><a id="semantics" class="anchor-link"></a><a href="#s
messages have a primary key and so the updates are idempotent (receiving the same message twice just overwrites a record with another copy of itself).
</ol>
<p>
- So what about exactly once semantics (i.e. the thing you actually want)? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>
- application), we can leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in a topic, so we can write the offset to Kafka in the
- same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's position will revert to its old value and the produced data on the output topics will not be visible
- to other consumers, depending on their "isolation level." In the default "read_uncommitted" isolation level, all messages are visible to consumers even if they were part of an aborted transaction,
- but in "read_committed," the consumer will only return messages from transactions which were committed (and any messages which were not part of a transaction).
+ So what about exactly-once semantics? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a> application), we can
+ leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in an internal topic, so we can write the offset to Kafka in the
+ same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's stored position will revert to its old value and the produced data on the output topics will not
+ be visible to other consumers, depending on their "isolation level." In the default "read_uncommitted" isolation level, all messages are visible to consumers even if they were part of an aborted transaction,
+ but in "read_committed" isolation level, the consumer will only return messages from transactions which were committed (and any messages which were not part of a transaction).
<p>
When writing to an external system, the limitation is in the need to coordinate the consumer's position with what is actually stored as output. The classic way of achieving this would be to introduce a two-phase
- commit between the storage of the consumer position and the storage of the consumers output. But this can be handled more simply and generally by letting the consumer store its offset in the same place as
+ commit between the storage of the consumer position and the storage of the consumers output. This can be handled more simply and generally by letting the consumer store its offset in the same place as
its output. This is better because many of the output systems a consumer might want to write to will not support a two-phase commit. As an example of this, consider a
<a href="https://kafka.apache.org/documentation/#connect">Kafka Connect</a> connector which populates data in HDFS along with the offsets of the data it reads so that it is guaranteed that either data and
offsets are both updated or neither is. We follow similar patterns for many other data systems which require these stronger semantics and for which the messages do not have a primary key to allow for deduplication.
<p>
- So effectively Kafka supports exactly-once delivery in <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>, and the transactional producer/consumer can be used generally to provide
+ As a result, Kafka supports exactly-once delivery in <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>, and the transactional producer/consumer can be used generally to provide
exactly-once delivery when transferring and processing data between Kafka topics. Exactly-once delivery for other destination systems generally requires cooperation with such systems, but Kafka provides the
offset which makes implementing this feasible (see also <a href="https://kafka.apache.org/documentation/#connect">Kafka Connect</a>). Otherwise, Kafka guarantees at-least-once delivery by default, and allows
the user to implement at-most-once delivery by disabling retries on the producer and committing offsets in the consumer prior to processing a batch of messages.
- <h3 class="anchor-heading"><a id="replication" class="anchor-link"></a><a href="#replication">4.7 Replication</a></h3>
+ <h3 class="anchor-heading"><a id="usingtransactions" class="anchor-link"></a><a href="#usingtransactions">4.7 Using Transactions</a></h3>
+ <p>
+ As mentioned above, the simplest way to get exactly-once semantics from Kafka is to use <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>. However, it is also possible to achieve
+ the same transactional guarantees using the Kafka producer and consumer directly by using them in the same way as Kafka Streams does.
+ <p>
+ Kafka transactions are a bit different than transactions in other messaging systems. In Kafka, the consumer and producer are separate and it is only the producer which is transactional. It is however able to
+ make transactional updates to the consumer's position (confusingly called the "committed offset"), and it is this which gives the overall exactly-once behavior.
+ <p>
+ There are three key aspects to exactly-once processing using the producer and consumer, which match how Kafka Streams works.
+ <ol>
+ <li>The consumer uses partition assignment to ensure that it is the only consumer in the consumer group currently processing each partition.</li>
+ <li>The consumer uses read-committed isolation level to ensure that it does not consume records produced by transactions which aborted.</li>
+ <li>The producer uses transactions so that all of the records it produces, and any offsets it updates on behalf of the consumer, are performed atomically.</li>
+ </ol>
+ <p>
+ The consumer configuration must include <code>isolation.level=read_committed</code> and <code>enable.auto.commit=false</code>. The producer configuration must set <code>transactional.id</code>
Review Comment:
It could also be a topic not written with transactions and in that case, read committed and read uncommitted have the same behavior.
Mostly just calling out it isn't a requirement, though probably a good practice.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17671: Create better documentation for transactions [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #17454:
URL: https://github.com/apache/kafka/pull/17454#discussion_r1907876128
##########
docs/design.html:
##########
@@ -290,24 +290,58 @@ <h3 class="anchor-heading"><a id="semantics" class="anchor-link"></a><a href="#s
messages have a primary key and so the updates are idempotent (receiving the same message twice just overwrites a record with another copy of itself).
</ol>
<p>
- So what about exactly once semantics (i.e. the thing you actually want)? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>
- application), we can leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in a topic, so we can write the offset to Kafka in the
- same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's position will revert to its old value and the produced data on the output topics will not be visible
- to other consumers, depending on their "isolation level." In the default "read_uncommitted" isolation level, all messages are visible to consumers even if they were part of an aborted transaction,
- but in "read_committed," the consumer will only return messages from transactions which were committed (and any messages which were not part of a transaction).
+ So what about exactly-once semantics? When consuming from a Kafka topic and producing to another topic (as in a <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a> application), we can
+ leverage the new transactional producer capabilities in 0.11.0.0 that were mentioned above. The consumer's position is stored as a message in an internal topic, so we can write the offset to Kafka in the
+ same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's stored position will revert to its old value (although the consumer has to refetch the
+ committed offset because it does not automatically rewind) and the produced data on the output topics will not be visible to other consumers, depending on their "isolation level". In the default
+ "read_uncommitted" isolation level, all messages are visible to consumers even if they were part of an aborted transaction, but in "read_committed" isolation level, the consumer will only return messages
+ from transactions which were committed (and any messages which were not part of a transaction).
<p>
When writing to an external system, the limitation is in the need to coordinate the consumer's position with what is actually stored as output. The classic way of achieving this would be to introduce a two-phase
- commit between the storage of the consumer position and the storage of the consumers output. But this can be handled more simply and generally by letting the consumer store its offset in the same place as
+ commit between the storage of the consumer position and the storage of the consumers output. This can be handled more simply and generally by letting the consumer store its offset in the same place as
its output. This is better because many of the output systems a consumer might want to write to will not support a two-phase commit. As an example of this, consider a
<a href="https://kafka.apache.org/documentation/#connect">Kafka Connect</a> connector which populates data in HDFS along with the offsets of the data it reads so that it is guaranteed that either data and
offsets are both updated or neither is. We follow similar patterns for many other data systems which require these stronger semantics and for which the messages do not have a primary key to allow for deduplication.
<p>
- So effectively Kafka supports exactly-once delivery in <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>, and the transactional producer/consumer can be used generally to provide
+ As a result, Kafka supports exactly-once delivery in <a href="https://kafka.apache.org/documentation/streams">Kafka Streams</a>, and the transactional producer/consumer can be used generally to provide
Review Comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org