[DISCUSS] RIP-47 Producer idempotent send

[柳尘 <yu...@gmail.com>]

*Status*


* - Current State: Discussing- Authors: [complone](github.com/complone
<http://github.com/complone>)- Shepherds: - Mailing List discussion:
dev@rocketmq.apache.org <de...@rocketmq.apache.org>- Pull Request: -
Released: no*Background & MotivationWhat do we need to do


* - Will we add a new module? yes- Will we add new APIs? yes- Will we add
new feature? yes*Why should we do that







* - Are there any problems of our current project?At present, in the
process of rocketmq producing messages, there is uncertainty in the network
call itself, that is, the so-called processing state, so there will be
repetitions. Many other MQ products also have this problem. The usual
approach is to ask consumers to deduplicate messages when consuming
messages. Idempotent bases should be generated by message producers. When
sending a message, we can pass the message's key. Set the id, the
corresponding API is ```org.apache.rocketmq.common.message.setKeys(String
keys)```.   - What can we benefit proposed changes?We introduce a notion of
TransactionalId, to enable users to uniquely identify producers in a
persistent way. Different instances of a producer with the same
TransactionalId will be able to resume (or abort) any transactions
instantiated by the previous instance.We introduce the notion of a producer
epoch, which enables us to ensure that there is only one legitimate active
instance of a producer with a given TransactionalId, and hence enables us
to maintain transaction guarantees in the event of failures.*Goals








* - What problem is this proposal designed to solve?When tuned for
reliability, users can guarantee that every message write will be persisted
at least once without data loss. Duplications may appear in the stream due
to producer retries. For example, a broker might crash between committing a
message and sending an ack to the producer, causing the producer to retry,
resulting in duplicate messages in the stream.However, idempotent producers
don’t provide guarantees for writes across multiple MessageQueues. For
this, one needs stronger transactional guarantees, ie. the ability to write
to several MessageQueues atomically. By atomically, we mean the ability to
commit a set of messages across MessageQueues as a unit: either all
messages are committed, or none of them are. - To what degree should we
solve the problem?Within a transaction, we also need to make sure that
there is no duplicate messages generated by the producer. To achieve this,
we are going to add sequence numbers to messages to allow the brokers to
de-duplicate messages per producer and topic partition. For each topic
partition that is written to, the producer maintains a sequence number
counter and assigns the next number in the sequence for each new message.
The broker verifies that the next message produced has been assigned the
next number and otherwise returns an error. In addition, since the sequence
number is per producer and topic partition, we also need to uniquely
identify a producer across multiple sessions (i.e. when the producer fails
and recreates, etc). Hence we introduce a new TransactionalId to
distinguish producers, along with an epoch number so that zombie writers
with the same TransactionalId can be fenced.At any given point in time, a
producer can only have one ongoing transaction, so we can distinguish
messages that belong to different transactions by their respective
TransactionalId. Producers with the same TransactionalId will talk to the
same transaction coordinator which also keeps track of their
TransactionalIds in addition to managing their transaction status. *
Non-Goals


* - What problem is this proposal NOT designed to solve?- Are there any
limits of this proposal?*ChangesArchitecture

Interface Design/Change

























* - Method signature changesA set of new public APIs to the
TransactionMQProducer class, and describe how these APIs will be
implemented./* initialize the producer as a transactional producer
*/initTransactions()The following steps will be taken when
initTransactions() is called: 1. If no TransactionalId has been provided in
configuration, skip to step 3.2. Send a FindCoordinatorRequest with the
configured TransactionalId and with CoordinatorType encoded as
“transaction” to a random broker. Block for the corresponding response,
which will return the assigned transaction coordinator for this producer.3.
Send an InitPidRequest to the transaction coordinator or to a random broker
if no TransactionalId was provided in configuration. Block for the
corresponding response to get the returned PID./* start a transaction to
produce messages */beginTransaction()  The following steps are executed on
the producer when beginTransaction is called: 1. Check if the producer is
transactional (i.e. init has been called), if not throw an exception (we
omit this step in the rest of the APIs, but they all need to execute it).2.
Check whether a transaction has already been started. If so, raise an
exception./* send offsets for a given consumer group within this
transaction */sendOffsetsToTransaction(            Map<TopicPartition,
OffsetAndMetadata> offsets,             String consumerGroupId)  The
following steps are executed on the producer when sendOffsetsToTransaction
is called: 1. Check if it is currently within a transaction, if not throw
an exception; otherwise proceed to the next step.2. Check if this function
has ever been called for the given groupId within this transaction. If not
then send an AddOffsetsToTxnRequest to the transaction coordinator, block
until the corresponding response is received; otherwise proceed to the next
step.3. Send a TxnOffsetCommitRequest to the coordinator return from the
response in the previous step, block until the corresponding response is
received./* commit the transaction with its produced messages
*/commitTransaction() The following steps are executed on the producer when
commitTransaction is called: 1. Check if there is an active transaction, if
not throw an exception; otherwise proceed to the next step.2. Call flush to
make sure all sent messages in this transactions are acknowledged.3. Send
an EndTxnRequest with COMMIT command to the transaction coordinator, block
until the corresponding response is received. - Method behavior changes -
CLI command changes- Log format or content changes* Compatibility,
Deprecation, and Migration Plan


* - Are backward and forward compatibility taken into consideration?- Are
there deprecated APIs?- How do we do migration?*Implementation Outline
Phase 1


*First, before sending data, you need to manually create a transaction
coordinator to control the one to many relationship between PID and
transcationId. To this end, you need to ensure that the consumption
progress of offset is notified to the tc in a timely manner and that the
messages of submitted but unsettled transactions will not be deleted by
mistake during the simultaneous process of the scheduled deletion task*Phase
2

*Whether it is necessary to judge the progress of the current offset and
the global PID visible transaction offset in PullLiteProducer in order to
enable PID ->transactionid to achieve cross partition transaction
consistency*Phase 3


*When downstream consumers consume, they need to ensure that messages
marked as transactionId in the upstream market are perceived by downstream
consumers*Rejected Alternatives How does alternatives solve the issue you
proposed?Pros and Cons of alternativesWhy should we reject above
alternatives

Re: [DISCUSS] RIP-47 Producer idempotent send

[柳尘 <yu...@gmail.com>]

ok i have resent

Xinyu Zhou <yu...@apache.org> 于2022年9月29日周四 20:10写道：

> Hi, thanks for starting this discussion.
>
> But the format is broken, it's hard to read, could you please resend a
> well-formatted version? Or, paste a doc link?
>
> Regards
>
> On Thu, Sep 29, 2022 at 7:23 PM 柳尘 <yu...@gmail.com> wrote:
>
> > *Status*
> >
> >
> > * - Current State: Discussing- Authors: [complone](github.com/complone
> > <http://github.com/complone>)- Shepherds: - Mailing List discussion:
> > dev@rocketmq.apache.org <de...@rocketmq.apache.org>- Pull Request: -
> > Released: no*Background & MotivationWhat do we need to do
> >
> >
> > * - Will we add a new module? yes- Will we add new APIs? yes- Will we add
> > new feature? yes*Why should we do that
> >
> >
> >
> >
> >
> >
> >
> > * - Are there any problems of our current project?At present, in the
> > process of rocketmq producing messages, there is uncertainty in the
> network
> > call itself, that is, the so-called processing state, so there will be
> > repetitions. Many other MQ products also have this problem. The usual
> > approach is to ask consumers to deduplicate messages when consuming
> > messages. Idempotent bases should be generated by message producers. When
> > sending a message, we can pass the message's key. Set the id, the
> > corresponding API is ```org.apache.rocketmq.common.message.setKeys(String
> > keys)```.   - What can we benefit proposed changes?We introduce a notion
> of
> > TransactionalId, to enable users to uniquely identify producers in a
> > persistent way. Different instances of a producer with the same
> > TransactionalId will be able to resume (or abort) any transactions
> > instantiated by the previous instance.We introduce the notion of a
> producer
> > epoch, which enables us to ensure that there is only one legitimate
> active
> > instance of a producer with a given TransactionalId, and hence enables us
> > to maintain transaction guarantees in the event of failures.*Goals
> >
> >
> >
> >
> >
> >
> >
> >
> > * - What problem is this proposal designed to solve?When tuned for
> > reliability, users can guarantee that every message write will be
> persisted
> > at least once without data loss. Duplications may appear in the stream
> due
> > to producer retries. For example, a broker might crash between
> committing a
> > message and sending an ack to the producer, causing the producer to
> retry,
> > resulting in duplicate messages in the stream.However, idempotent
> producers
> > don’t provide guarantees for writes across multiple MessageQueues. For
> > this, one needs stronger transactional guarantees, ie. the ability to
> write
> > to several MessageQueues atomically. By atomically, we mean the ability
> to
> > commit a set of messages across MessageQueues as a unit: either all
> > messages are committed, or none of them are. - To what degree should we
> > solve the problem?Within a transaction, we also need to make sure that
> > there is no duplicate messages generated by the producer. To achieve
> this,
> > we are going to add sequence numbers to messages to allow the brokers to
> > de-duplicate messages per producer and topic partition. For each topic
> > partition that is written to, the producer maintains a sequence number
> > counter and assigns the next number in the sequence for each new message.
> > The broker verifies that the next message produced has been assigned the
> > next number and otherwise returns an error. In addition, since the
> sequence
> > number is per producer and topic partition, we also need to uniquely
> > identify a producer across multiple sessions (i.e. when the producer
> fails
> > and recreates, etc). Hence we introduce a new TransactionalId to
> > distinguish producers, along with an epoch number so that zombie writers
> > with the same TransactionalId can be fenced.At any given point in time, a
> > producer can only have one ongoing transaction, so we can distinguish
> > messages that belong to different transactions by their respective
> > TransactionalId. Producers with the same TransactionalId will talk to the
> > same transaction coordinator which also keeps track of their
> > TransactionalIds in addition to managing their transaction status. *
> > Non-Goals
> >
> >
> > * - What problem is this proposal NOT designed to solve?- Are there any
> > limits of this proposal?*ChangesArchitecture
> >
> > Interface Design/Change
> >
> >
> >
> >
> >
> >
> >
> >
> >
> >
> >
> >
> >
> >
> >
> >
> >
> >
> >
> >
> >
> >
> >
> >
> >
> > * - Method signature changesA set of new public APIs to the
> > TransactionMQProducer class, and describe how these APIs will be
> > implemented./* initialize the producer as a transactional producer
> > */initTransactions()The following steps will be taken when
> > initTransactions() is called: 1. If no TransactionalId has been provided
> in
> > configuration, skip to step 3.2. Send a FindCoordinatorRequest with the
> > configured TransactionalId and with CoordinatorType encoded as
> > “transaction” to a random broker. Block for the corresponding response,
> > which will return the assigned transaction coordinator for this
> producer.3.
> > Send an InitPidRequest to the transaction coordinator or to a random
> broker
> > if no TransactionalId was provided in configuration. Block for the
> > corresponding response to get the returned PID./* start a transaction to
> > produce messages */beginTransaction()  The following steps are executed
> on
> > the producer when beginTransaction is called: 1. Check if the producer is
> > transactional (i.e. init has been called), if not throw an exception (we
> > omit this step in the rest of the APIs, but they all need to execute
> it).2.
> > Check whether a transaction has already been started. If so, raise an
> > exception./* send offsets for a given consumer group within this
> > transaction */sendOffsetsToTransaction(            Map<TopicPartition,
> > OffsetAndMetadata> offsets,             String consumerGroupId)  The
> > following steps are executed on the producer when
> sendOffsetsToTransaction
> > is called: 1. Check if it is currently within a transaction, if not throw
> > an exception; otherwise proceed to the next step.2. Check if this
> function
> > has ever been called for the given groupId within this transaction. If
> not
> > then send an AddOffsetsToTxnRequest to the transaction coordinator, block
> > until the corresponding response is received; otherwise proceed to the
> next
> > step.3. Send a TxnOffsetCommitRequest to the coordinator return from the
> > response in the previous step, block until the corresponding response is
> > received./* commit the transaction with its produced messages
> > */commitTransaction() The following steps are executed on the producer
> when
> > commitTransaction is called: 1. Check if there is an active transaction,
> if
> > not throw an exception; otherwise proceed to the next step.2. Call flush
> to
> > make sure all sent messages in this transactions are acknowledged.3. Send
> > an EndTxnRequest with COMMIT command to the transaction coordinator,
> block
> > until the corresponding response is received. - Method behavior changes -
> > CLI command changes- Log format or content changes* Compatibility,
> > Deprecation, and Migration Plan
> >
> >
> > * - Are backward and forward compatibility taken into consideration?- Are
> > there deprecated APIs?- How do we do migration?*Implementation Outline
> > Phase 1
> >
> >
> > *First, before sending data, you need to manually create a transaction
> > coordinator to control the one to many relationship between PID and
> > transcationId. To this end, you need to ensure that the consumption
> > progress of offset is notified to the tc in a timely manner and that the
> > messages of submitted but unsettled transactions will not be deleted by
> > mistake during the simultaneous process of the scheduled deletion
> > task*Phase
> > 2
> >
> > *Whether it is necessary to judge the progress of the current offset and
> > the global PID visible transaction offset in PullLiteProducer in order to
> > enable PID ->transactionid to achieve cross partition transaction
> > consistency*Phase 3
> >
> >
> > *When downstream consumers consume, they need to ensure that messages
> > marked as transactionId in the upstream market are perceived by
> downstream
> > consumers*Rejected Alternatives How does alternatives solve the issue you
> > proposed?Pros and Cons of alternativesWhy should we reject above
> > alternatives
> >
>

Re: [DISCUSS] RIP-47 Producer idempotent send

[Xinyu Zhou <yu...@apache.org>]

Hi, thanks for starting this discussion.

But the format is broken, it's hard to read, could you please resend a
well-formatted version? Or, paste a doc link?

Regards

On Thu, Sep 29, 2022 at 7:23 PM 柳尘 <yu...@gmail.com> wrote:

> *Status*
>
>
> * - Current State: Discussing- Authors: [complone](github.com/complone
> <http://github.com/complone>)- Shepherds: - Mailing List discussion:
> dev@rocketmq.apache.org <de...@rocketmq.apache.org>- Pull Request: -
> Released: no*Background & MotivationWhat do we need to do
>
>
> * - Will we add a new module? yes- Will we add new APIs? yes- Will we add
> new feature? yes*Why should we do that
>
>
>
>
>
>
>
> * - Are there any problems of our current project?At present, in the
> process of rocketmq producing messages, there is uncertainty in the network
> call itself, that is, the so-called processing state, so there will be
> repetitions. Many other MQ products also have this problem. The usual
> approach is to ask consumers to deduplicate messages when consuming
> messages. Idempotent bases should be generated by message producers. When
> sending a message, we can pass the message's key. Set the id, the
> corresponding API is ```org.apache.rocketmq.common.message.setKeys(String
> keys)```.   - What can we benefit proposed changes?We introduce a notion of
> TransactionalId, to enable users to uniquely identify producers in a
> persistent way. Different instances of a producer with the same
> TransactionalId will be able to resume (or abort) any transactions
> instantiated by the previous instance.We introduce the notion of a producer
> epoch, which enables us to ensure that there is only one legitimate active
> instance of a producer with a given TransactionalId, and hence enables us
> to maintain transaction guarantees in the event of failures.*Goals
>
>
>
>
>
>
>
>
> * - What problem is this proposal designed to solve?When tuned for
> reliability, users can guarantee that every message write will be persisted
> at least once without data loss. Duplications may appear in the stream due
> to producer retries. For example, a broker might crash between committing a
> message and sending an ack to the producer, causing the producer to retry,
> resulting in duplicate messages in the stream.However, idempotent producers
> don’t provide guarantees for writes across multiple MessageQueues. For
> this, one needs stronger transactional guarantees, ie. the ability to write
> to several MessageQueues atomically. By atomically, we mean the ability to
> commit a set of messages across MessageQueues as a unit: either all
> messages are committed, or none of them are. - To what degree should we
> solve the problem?Within a transaction, we also need to make sure that
> there is no duplicate messages generated by the producer. To achieve this,
> we are going to add sequence numbers to messages to allow the brokers to
> de-duplicate messages per producer and topic partition. For each topic
> partition that is written to, the producer maintains a sequence number
> counter and assigns the next number in the sequence for each new message.
> The broker verifies that the next message produced has been assigned the
> next number and otherwise returns an error. In addition, since the sequence
> number is per producer and topic partition, we also need to uniquely
> identify a producer across multiple sessions (i.e. when the producer fails
> and recreates, etc). Hence we introduce a new TransactionalId to
> distinguish producers, along with an epoch number so that zombie writers
> with the same TransactionalId can be fenced.At any given point in time, a
> producer can only have one ongoing transaction, so we can distinguish
> messages that belong to different transactions by their respective
> TransactionalId. Producers with the same TransactionalId will talk to the
> same transaction coordinator which also keeps track of their
> TransactionalIds in addition to managing their transaction status. *
> Non-Goals
>
>
> * - What problem is this proposal NOT designed to solve?- Are there any
> limits of this proposal?*ChangesArchitecture
>
> Interface Design/Change
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
> * - Method signature changesA set of new public APIs to the
> TransactionMQProducer class, and describe how these APIs will be
> implemented./* initialize the producer as a transactional producer
> */initTransactions()The following steps will be taken when
> initTransactions() is called: 1. If no TransactionalId has been provided in
> configuration, skip to step 3.2. Send a FindCoordinatorRequest with the
> configured TransactionalId and with CoordinatorType encoded as
> “transaction” to a random broker. Block for the corresponding response,
> which will return the assigned transaction coordinator for this producer.3.
> Send an InitPidRequest to the transaction coordinator or to a random broker
> if no TransactionalId was provided in configuration. Block for the
> corresponding response to get the returned PID./* start a transaction to
> produce messages */beginTransaction()  The following steps are executed on
> the producer when beginTransaction is called: 1. Check if the producer is
> transactional (i.e. init has been called), if not throw an exception (we
> omit this step in the rest of the APIs, but they all need to execute it).2.
> Check whether a transaction has already been started. If so, raise an
> exception./* send offsets for a given consumer group within this
> transaction */sendOffsetsToTransaction(            Map<TopicPartition,
> OffsetAndMetadata> offsets,             String consumerGroupId)  The
> following steps are executed on the producer when sendOffsetsToTransaction
> is called: 1. Check if it is currently within a transaction, if not throw
> an exception; otherwise proceed to the next step.2. Check if this function
> has ever been called for the given groupId within this transaction. If not
> then send an AddOffsetsToTxnRequest to the transaction coordinator, block
> until the corresponding response is received; otherwise proceed to the next
> step.3. Send a TxnOffsetCommitRequest to the coordinator return from the
> response in the previous step, block until the corresponding response is
> received./* commit the transaction with its produced messages
> */commitTransaction() The following steps are executed on the producer when
> commitTransaction is called: 1. Check if there is an active transaction, if
> not throw an exception; otherwise proceed to the next step.2. Call flush to
> make sure all sent messages in this transactions are acknowledged.3. Send
> an EndTxnRequest with COMMIT command to the transaction coordinator, block
> until the corresponding response is received. - Method behavior changes -
> CLI command changes- Log format or content changes* Compatibility,
> Deprecation, and Migration Plan
>
>
> * - Are backward and forward compatibility taken into consideration?- Are
> there deprecated APIs?- How do we do migration?*Implementation Outline
> Phase 1
>
>
> *First, before sending data, you need to manually create a transaction
> coordinator to control the one to many relationship between PID and
> transcationId. To this end, you need to ensure that the consumption
> progress of offset is notified to the tc in a timely manner and that the
> messages of submitted but unsettled transactions will not be deleted by
> mistake during the simultaneous process of the scheduled deletion
> task*Phase
> 2
>
> *Whether it is necessary to judge the progress of the current offset and
> the global PID visible transaction offset in PullLiteProducer in order to
> enable PID ->transactionid to achieve cross partition transaction
> consistency*Phase 3
>
>
> *When downstream consumers consume, they need to ensure that messages
> marked as transactionId in the upstream market are perceived by downstream
> consumers*Rejected Alternatives How does alternatives solve the issue you
> proposed?Pros and Cons of alternativesWhy should we reject above
> alternatives
>