Hi Roger, Thank you for summarizing the cons. I agree and I'm curious what would be the alternatives to solve these problems better and if they can be incorporated into this proposal (or built independently in addition to or on top of this proposal). E.g. one potential extension we discussed earlier in the thread could be multiplexing logical transactional "streams" with a single producer.
-Artem On Wed, Aug 23, 2023 at 4:50 PM Roger Hoover <roger.hoo...@gmail.com> wrote: > Thanks. I like that you're moving Kafka toward supporting this dual-write > pattern. Each use case needs to consider the tradeoffs. You already > summarized the pros very well in the KIP. I would summarize the cons > as follows: > > - you sacrifice availability - each write requires both DB and Kafka to be > available so I think your overall application availability is 1 - p(DB is > unavailable)*p(Kafka is unavailable). > - latency will be higher and throughput lower - each write requires both > writes to DB and Kafka while holding an exclusive lock in DB. > - you need to create a producer per unit of concurrency in your app which > has some overhead in the app and Kafka side (number of connections, poor > batching). I assume the producers would need to be configured for low > latency (linger.ms=0) > - there's some complexity in managing stable transactional ids for each > producer/concurrency unit in your application. With k8s deployment, you > may need to switch to something like a StatefulSet that gives each pod a > stable identity across restarts. On top of that pod identity which you can > use as a prefix, you then assign unique transactional ids to each > concurrency unit (thread/goroutine). > > On Wed, Aug 23, 2023 at 12:53 PM Artem Livshits > <alivsh...@confluent.io.invalid> wrote: > > > Hi Roger, > > > > Thank you for the feedback. You make a very good point that we also > > discussed internally. Adding support for multiple concurrent > > transactions in one producer could be valuable but it seems to be a > fairly > > large and independent change that would deserve a separate KIP. If such > > support is added we could modify 2PC functionality to incorporate that. > > > > > Maybe not too bad but a bit of pain to manage these ids inside each > > process and across all application processes. > > > > I'm not sure if supporting multiple transactions in one producer would > make > > id management simpler: we'd need to store a piece of data per > transaction, > > so whether it's N producers with a single transaction or N transactions > > with a single producer, it's still roughly the same amount of data to > > manage. In fact, managing transactional ids (current proposal) might be > > easier, because the id is controlled by the application and it knows how > to > > complete the transaction after crash / restart; while a TID would be > > generated by Kafka and that would create a question of starting Kafka > > transaction, but not saving its TID and then crashing, then figuring out > > which transactions to abort and etc. > > > > > 2) creating a separate producer for each concurrency slot in the > > application > > > > This is a very valid concern. Maybe we'd need to have some multiplexing > of > > transactional logical "streams" over the same connection. Seems like a > > separate KIP, though. > > > > > Otherwise, it seems you're left with single-threaded model per > > application process? > > > > That's a fair assessment. Not necessarily exactly single-threaded per > > application, but a single producer per thread model (i.e. an application > > could have a pool of threads + producers to increase concurrency). > > > > -Artem > > > > On Tue, Aug 22, 2023 at 7:22 PM Roger Hoover <roger.hoo...@gmail.com> > > wrote: > > > > > Artem, > > > > > > Thanks for the reply. > > > > > > If I understand correctly, Kafka does not support concurrent > transactions > > > from the same producer (transactional id). I think this means that > > > applications that want to support in-process concurrency (say > > thread-level > > > concurrency with row-level DB locking) would need to manage separate > > > transactional ids and producers per thread and then store txn state > > > accordingly. The potential usability downsides I see are > > > 1) managing a set of transactional ids for each application process > that > > > scales up to it's max concurrency. Maybe not too bad but a bit of pain > > to > > > manage these ids inside each process and across all application > > processes. > > > 2) creating a separate producer for each concurrency slot in the > > > application - this could create a lot more producers and resultant > > > connections to Kafka than the typical model of a single producer per > > > process. > > > > > > Otherwise, it seems you're left with single-threaded model per > > application > > > process? > > > > > > Thanks, > > > > > > Roger > > > > > > On Tue, Aug 22, 2023 at 5:11 PM Artem Livshits > > > <alivsh...@confluent.io.invalid> wrote: > > > > > > > Hi Roger, Arjun, > > > > > > > > Thank you for the questions. > > > > > It looks like the application must have stable transactional ids > over > > > > time? > > > > > > > > The transactional id should uniquely identify a producer instance and > > > needs > > > > to be stable across the restarts. If the transactional id is not > > stable > > > > across restarts, then zombie messages from a previous incarnation of > > the > > > > producer may violate atomicity. If there are 2 producer instances > > > > concurrently producing data with the same transactional id, they are > > > going > > > > to constantly fence each other and most likely make little or no > > > progress. > > > > > > > > The name might be a little bit confusing as it may be mistaken for a > > > > transaction id / TID that uniquely identifies every transaction. The > > > name > > > > and the semantics were defined in the original exactly-once-semantics > > > (EoS) > > > > proposal ( > > > > > > > > > > > > > > https://cwiki.apache.org/confluence/display/KAFKA/KIP-98+-+Exactly+Once+Delivery+and+Transactional+Messaging > > > > ) > > > > and KIP-939 just build on top of that. > > > > > > > > > I'm curious to understand what happens if the producer dies, and > does > > > not > > > > come up and recover the pending transaction within the transaction > > > timeout > > > > interval. > > > > > > > > If the producer / application never comes back, the transaction will > > > remain > > > > in prepared (a.k.a. "in-doubt") state until an operator forcefully > > > > terminates the transaction. That's why there is a new ACL is defined > > in > > > > this proposal -- this functionality should only provided to > > applications > > > > that implement proper recovery logic. > > > > > > > > -Artem > > > > > > > > On Tue, Aug 22, 2023 at 12:52 AM Arjun Satish < > arjun.sat...@gmail.com> > > > > wrote: > > > > > > > > > Hello Artem, > > > > > > > > > > Thanks for the KIP. > > > > > > > > > > I have the same question as Roger on concurrent writes, and an > > > additional > > > > > one on consumer behavior. Typically, transactions will timeout if > not > > > > > committed within some time interval. With the proposed changes in > > this > > > > KIP, > > > > > consumers cannot consume past the ongoing transaction. I'm curious > to > > > > > understand what happens if the producer dies, and does not come up > > and > > > > > recover the pending transaction within the transaction timeout > > > interval. > > > > Or > > > > > are we saying that when used in this 2PC context, we should > configure > > > > these > > > > > transaction timeouts to very large durations? > > > > > > > > > > Thanks in advance! > > > > > > > > > > Best, > > > > > Arjun > > > > > > > > > > > > > > > On Mon, Aug 21, 2023 at 1:06 PM Roger Hoover < > roger.hoo...@gmail.com > > > > > > > > wrote: > > > > > > > > > > > Hi Artem, > > > > > > > > > > > > Thanks for writing this KIP. Can you clarify the requirements a > > bit > > > > more > > > > > > for managing transaction state? It looks like the application > must > > > > have > > > > > > stable transactional ids over time? What is the granularity of > > > those > > > > > ids > > > > > > and producers? Say the application is a multi-threaded Java web > > > > server, > > > > > > can/should all the concurrent threads share a transactional id > and > > > > > > producer? That doesn't seem right to me unless the application > is > > > > using > > > > > > global DB locks that serialize all requests. Instead, if the > > > > application > > > > > > uses row-level DB locks, there could be multiple, concurrent, > > > > independent > > > > > > txns happening in the same JVM so it seems like the granularity > > > > managing > > > > > > transactional ids and txn state needs to line up with granularity > > of > > > > the > > > > > DB > > > > > > locking. > > > > > > > > > > > > Does that make sense or am I misunderstanding? > > > > > > > > > > > > Thanks, > > > > > > > > > > > > Roger > > > > > > > > > > > > On Wed, Aug 16, 2023 at 11:40 PM Artem Livshits > > > > > > <alivsh...@confluent.io.invalid> wrote: > > > > > > > > > > > > > Hello, > > > > > > > > > > > > > > This is a discussion thread for > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > https://cwiki.apache.org/confluence/display/KAFKA/KIP-939%3A+Support+Participation+in+2PC > > > > > > > . > > > > > > > > > > > > > > The KIP proposes extending Kafka transaction support (that > > already > > > > uses > > > > > > 2PC > > > > > > > under the hood) to enable atomicity of dual writes to Kafka and > > an > > > > > > external > > > > > > > database, and helps to fix a long standing Flink issue. > > > > > > > > > > > > > > An example of code that uses the dual write recipe with JDBC > and > > > > should > > > > > > > work for most SQL databases is here > > > > > > > https://github.com/apache/kafka/pull/14231. > > > > > > > > > > > > > > The FLIP for the sister fix in Flink is here > > > > > > > > > > > > > > > > > > > > > > > > > > > > https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=255071710 > > > > > > > > > > > > > > -Artem > > > > > > > > > > > > > > > > > > > > > > > > > > > >
