Status

Current state: "Draft"

Discussion thread: here [Change the link from the KIP proposal email archive to your own email thread]

JIRA: here [Change the link from KAFKA-1 to your own ticket]

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Background

2PC Refresher

Two phase commit (a.k.a. 2PC) protocol is used to implement distributed transactions across multiple participants.  Each participant manages its own state that is not recovered from other participants (if some datasets can be fully recovered from another source they are considered as one participant in 2PC protocol, e.g. a database could have multiple tables and indices, but they are all recovered from one log, so it’s one participant; on the other hand a broker manages multiple partitions but they don’t recover from each other, so they are multiple participants).  One of the participants (or a separate entity) is designated to be a transaction coordinator.  A transaction coordinator could be different for each transaction (e.g. the first participant of a transaction becomes the transaction coordinator) or it could be the same (e.g. if we use a designated transaction coordinator it’s generally used for all transactions); in any case the role of the transaction coordinator for a transaction is that it stores the decision for the transaction outcome (commit or abort) and is able to eventually deliver it to all transaction participants.

The 2PC protocol is the following:

1. Each participant does some atomic transformations.  It is generally assumed that each participant can support non-distributed transactions, in which case no additional logic is required by the 2PC: a failed transaction aborts and gets back to a state before the transaction started.  If any participant fails while doing some atomic transformation, the distributed transaction is aborted.
2. When a distributed transaction is ready to commit, all participants are asked to prepare the transaction to commit.  This is the PREPARE phase (the first phase of 2PC).  Each participant is free to say ‘yes’ or ‘no’; if any participant says ‘no’ (or just fails or times out or etc.) the transaction is aborted.  If a participant says ‘yes’ during the PREPARE phase, it must be able to commit or abort at will, based on the coordinator’s decision and must remain in-doubt until it receives the coordinator’s decision across all failures of the participant.  A transaction that is prepared in a participant is said to be in an in-doubt state (or just in-doubt transaction) – in this state neither the new nor the old state can be revealed, locks are held, etc.  All failure modes should be eliminated – data must be flushed and cannot be lost on restart, log needs to have space to store the commit marker, etc. – a commit cannot fail.
3. Once all the participants are prepared, we can make a transaction outcome decision.  It could be either commit or abort.  If any of the participants replied ‘no’ during the PREPARE phase the transaction would be aborted.  If all participants reply ‘yes’ then the transaction coordinator tries to commit the transaction (e.g. by writing a record to its log).  If the commit operation succeeds in the transaction coordinator, then the transaction becomes committed.  This is the COMMIT phase, the 2nd phase of the 2PC protocol.  The decision is communicated to all participants eventually.  The fact that some participants may receive the outcome decision later than the other doesn’t affect correctness, because a participant that hasn't received the decision yet doesn’t reveal the outcome (locks are held, etc.).

There are variations of implementations (e.g. how the commit decision is communicated – could be pushed by the transaction coordinator, pulled by participants, “gossiped” by the client [if the client saw a commit of a transaction on one participant other participants can infer that it’s committed], etc.).

Kafka as a Participant in 2PC

Kafka is a distributed system, so transactions in Kafka are already distributed internally.  The 2PC maps to Kafka protocol as the following:

1. Each partition is registered with the Kafka transaction coordinator so that it would abort the transaction in case of abort.  Messages are produced to each partition as needed.
2. When a transaction is ready to commit, data is flushed to all data partitions (and commit offsets are flushed to the corresponding offset_commit partitions).  If any flush fails, then it’s treated as ‘no’ answer and the transaction is aborted.  If all flushes are successful, then the commit decision is sent to the Kafka transaction coordinator to be recorded.  The “prepared” transaction data is not visible to consumers reading with ‘read_committed’ isolation level.
3. The Kafka transaction coordinator writes a PREPARE_COMMIT (or PREPARE_ABORT) record to its log to record a transaction outcome decision.  The name is somewhat misleading, because this is the COMMIT phase and once the PREPARE_COMMIT record is written, the transaction is committed and it’s just a matter of making the transaction participants aware of decisions.  The Kafka transaction coordinator pushes transaction outcome decisions until it’s propagated to all participating partitions and wouldn’t let another transaction (with the same transactional.id) started, until the transaction decision is delivered to all participating partitions.

To protect from client failures and avoid transactions to be in-doubt for a long time, Kafka has a limit on how long a transaction can be running - transaction.max.timeout.ms (15 minutes by default), so if a transaction is running longer than the limit it’s going to be aborted.

From that perspective, Kafka already has a 2PC implementation, but it’s controlled by Kafka transaction coordinator.  If we could support an external transaction coordinator, Kafka would be able to become a participant in a distributed transaction with other systems.

The main requirement for a participant is that once prepared it cannot abort (or commit) on its own and must wait for a decision made by the external transaction coordinator.  Currently, Kafka violates this requirement in two cases:

1. On client restart, the client must use InitProducerId RPC to start using transactions.  InitProducerId aborts the on-going transaction, so even though the transaction is prepared (all data is flushed in place) it cannot survive client restart.
2. A transaction that’s running longer than transaction.max.timeout.ms would be aborted (thus would make a decision that may be different from the external transaction coordinator’s decision).

Fixing these cases would let Kafka become a participant in an externally coordinated distributed transaction.

Motivation

One challenge if you want to make a log of events a key output of a stateful application is how to update both Kafka and your derived stores together.  E.g. say you want to use Kafka as a log of events or changes, but want to have a relational database and/or search index on the data.  How can you keep these in sync?

Currently we have three approaches that are possible:

1. DB in front—you write to the DB and capture the changes using the database CDC mechanism (maybe using Kafka Connect) and publish them to Kafka
2. Kafka in front—you write to Kafka you subscribe to the Kafka topic and apply whatever is written there to your DB
3. Dual write—you write to both Kafka and your DB independently

Each of these has the corresponding drawbacks:

1. DB in front with change capture works well for collecting data from existing apps, but for event sourcing this is a bit of a pain since the data will all be formatted in whatever odd way the DB log captures it, and it will be highly normalized in a way that may not be very useful outside your DB schema.
2. Kafka in front lets you have nicely structured events, but you lose read-after-write consistency with your database which is a hard requirement for many (most?) uses.
3. Dual write is easy to implement but means that whenever there is a failure there is a high likelihood that the log and the database will diverge.

It would be nice to have a way to get the best characteristics of all approaches:

1. Well structured event log written directly by the app
2. Read-after-write consistency
3. Consistency between Kafka and all other data stores

Looking at the approaches and their drawbacks, dual writes approach seems to be the most promising as the drawbacks of the other approaches are pretty fundamental to them.  The dual writes would work most of the time and we just need to address the atomicity of the failures cases.

To approach a solution for the atomicity of dual writes, let's consider a recipe for atomic consumption from Kafka to a database:

1. Consume data from Kafka
2. Begin database transaction
3. Write consumed data to the database
4. Write offsets of consumed data to the database
5. Commit database transaction

The overhead during normal processing is minimal.  In the failure scenario, the offsets of consumed data can be read from the database and the consumption can be resumed from those offsets.  Given that the offsets got committed to the database atomically with the data, the state of the database would be consistent with what's consumed from Kafka.

The key idea that makes the consumption atomic is that the database tracks the consumption state atomically with the data update.

We can extend this idea to solve dual writes atomicity problem: we can produce to Kafka in a Kafka transaction and then write the transaction state atomically with the data update.  Then in the failure scenario, the transaction state in the database will help us to know whether the Kafka transaction needs to be aborted or committed.

Using this idea the dual write recipe becomes the following:

1. Begin database transaction
2. Begin Kafka transaction
3. Produce data to Kafka
4. Prepare Kafka transaction [currently implicit operation, expressed as flush]
5. Write produced data to the database
6. Write offsets of produced data to the database
7. Commit database transaction
8. Commit Kafka transaction

In the case of failure, we can read the offsets of produced data from the database, and see if the match the offsets of the prepared Kafka transaction and can make a decision whether the Kafka transaction needs to be aborted or committed -- if the offsets in the database are behind the offsets of Kafka transaction, then the transaction needs to be aborted.  Currently, there is no way to inspect the offsets of prepared Kafka transactions, but offsets are just a way to uniquely identify an ongoing transaction. Once KIP-890 is implemented, we could use {producerId, epoch} to uniquely identify the transaction.

Given that neither the database nor the application using this recipe needs to interpret the actual data that identifies transaction, it would be best to encapsulate the offset information in an object that can be serialized to / deserialized from a string, so that it's easy to write to / read from the database.  The requirements on the database functionality are minimal -- it just needs to support some form of strictly consistent atomic writes, e.g. a key-value store such as RocksDB could be used with dual write recipe as well as full featured SQL or NoSQL databases.

If we had

• Kafka two phase commit support (which is almost there as explained in the background section)
• KIP-890, that makes it possible to uniquely identify transactions by {producerId, epoch}
• PreparedTransactionState class that under the covers contains {producerId, epoch} and can be stored as a string
• .prepareTransaction function that returns PreparedTransactionState 
• .completePreparedTransaction function that could take PreparedTransactionState (and commit / abort based on that)

then the dual write recipe would be more formally defined as:

Database schema prerequisite:

Some way to store a mapping of transactionalId (string) to preparedTransactionState (string).  How it's going to be expressed depends on the database.  In a SQL database it could be a table defined by CREATE TABLE kafka_transaction_state (transactional_id varchar(255), prepared_transaction_state varchar(max));.  In RocksDB it could be a column family named kafka_transaction_state where the transactionalId would be the key and serialized preparedTransactionState would be a value.  In Zookeeper it could be a path /kafka_transaction_state and the zknodes under this path would be transactional ids and their values would be prepared transaction states.  Etc.

Normal processing recipe:

1. Begin database transaction
2. beginTransaction in Kafka
3. send messages to Kafka
4. prepareTransaction in Kafka, get the preparedTransactionState
5. Update data to reflect produced messages
6. Update transactionalId -> preparedTransactionState mapping with the new preparedTransactionState
7. Commit database transaction
8. commitTransaction in Kafka

Recovery recipe:

1. Read preparedTransactionState from the database
2. completePreparedTransaction with the preparedTransactionState

The recovery recipe is used when committing the database transaction fails or after restarts, in other cases we could just abort the transaction on failure.

Once the basic support for the dual write recipe is in place as a result of the KIP, we can implement libraries and / or template examples to facilitate integration with various databases, so that the application writers wouldn't have to implement the recipe from scratch for each application.

Here is a code example that uses the dual write recipe with JDBC and should work for most SQL databases: https://github.com/apache/kafka/pull/14231.

Another important use case that currently has no good solution is supporting Exactly Once Semantics (EOS) between Flink’s KafkaSink operator and Kafka.  The KafkaSink operator uses Kafka transactions to implement EOS across Flink and Kafka.  In that case, Flink’s job manager effectively acts as an external transaction coordinator in a 2PC protocol and Kafka is one of the participants.  For more details about the Flink KafkaSink operation and how it could utilize Kafka 2PC see FLIP-319.

The KafkaSink operator manages to work around the first problem (InitProducerId aborting the transaction) using reflection: KafkaSink keeps track of the producerId and epoch, so if it needs to commit a transaction after the producer has crashed, it could just issue a commit without going through the InitProducerId (which would abort the transaction).  Obviously, using reflection leads to maintenance nightmare so adding official support into Kafka so that KafkaSink could use a public API instead of reflection is a great improvement.

The second problem (transaction timeout) cannot be robustly solved without support from Kafka.  The current workaround is to crank up transaction.max.timeout.ms and hope that Flink won’t hold a transaction for a long period of time, but “hoping” is not a very robust way to provide technical guarantees.

Scope

The scope of this KIP is to enable Kafka to be a participant in a 2PC protocol and build a foundation for dual write recipe.  Building libraries for dual writes for various databases is out of scope of this KIP.  Any additional distributed transaction functionality (e.g. using Kafka as a transaction coordinator, supporting Open XA) is out of the scope of this KIP.

Solution Requirements and Constraints

In the essence we need two changes:

1. Officially support functionality that can be used (and is used by Flink) anyway via reflection.
2. Relax / remove the timeout for transactions participating in 2PC.

We should also wait for KIP-890 to be implemented, so that we could use {producerId, epoch} to identify a transaction.

Adding support for 1 should have no technical implications, we just publicly support functionality that’s currently can be used (and is used by Flink) via reflection anyway.

Adding support for 2 requires considering why we have timeout in the first place and how we can mitigate problems that could arise if we remove timeout.

When a partition has an ongoing transaction, there are some implications: consumers that use read_committed isolation level cannot consume past the ongoing transaction (even if there are committed transactions later in the partition, consumers with read_committed isolations won’t read past messages that are part of an ongoing transaction), compaction wouldn’t compact past ongoing transactions.

transaction.max.timeout.ms guarantees that an ongoing transaction is aborted within a reasonable amount of time, but to avoid violation of the 2PC protocol we need to keep the transaction open, which could put pressure on the system.  To mitigate this impact, we should restrict the ability to run 2PC protocol via a privilege, so that it’s easy to protect the cluster from a random rogue application.

Even with restricting 2PC functionality to well behaving applications, we cannot prevent cases of long running and / or abandoned transactions.  We need some metrics to detect long running transactions.  We also need a way for the admin to find and forcibly abort long running and abandoned transactions in case the application cannot properly complete them.

Proposed Changes

The InitProducerId API would provide way to indicate that:

• This transactional producer is a participant in 2PC protocol
• Don’t abort the ongoing transaction (so that it’s possible to commit a prepared transaction even after producer restarts)

The broker will check if the client is allowed to use 2PC protocol, the request will fail if 2PC protocol is not allowed for the client.

If the InitProducerId was asked to not abort the ongoing transaction, then the application is only allowed to call .commitTransaction, .abortTransaction, or .completeTransaction calling other methods (e.g. .send) would fail.  This is because the reason to call InitProducerId without aborting the ongoing transaction is to complete a prepared transaction after the producer’s crash.

If the transaction is a participant of the 2PC protocol, we don’t limit the transactional timeout by the transaction.max.timeout.ms any more, so if the client passes max integer value, the broker can accept it.

The admin client would support a new method to abort a transaction with a given transactional id.  The method would just execute InitProducerId.

A new metric would be added to track ongoing transaction time.

Public Interfaces

RPC Changes

We will bump the InitProducerId API.  The new request schema is going to be the following:

InitProducerIdRequest => TransactionalId TransactionTimeoutMs ProducerId Epoch Enable2Pc KeepPreparedTxn
 TransactionalId => NULLABLE_STRING
 TransactionTimeoutMs => INT32
 ProducerId => INT64
 Epoch => INT16
 Enable2Pc => BOOL  // NEW
 KeepPreparedTxn => BOOL  // NEW

Implementation note for InitProducerId API – even though we may not change the state when KeepPreparedTxn is specified, we still need to do a write (of the same state) to make sure we’re not getting data from a stale leader.

We’ll also bump ListTransactionsRequest to support listing only transactions that are running longer than a certain amount of time.

ListTransactionsRequest => StatesFilter ProducerIdFilter RunningLongerThanMs
 StatesFilter => []STRING
 ProducerIdFilter => []STRING
 RunningLongerThanMs => INT64 // NEW

Metric Changes

A new metric will be added

active-transaction-total-time-max The max time a currently-open transaction has been open

To calculate the metrics we’ll just walk through the ongoing transactions and record the max value.

Client Configuration Changes

transaction.two.phase.commit.enable The default would be ‘false’.  If set to ‘true’, then the broker is informed that the client is participating in two phase commit protocol and can set transaction timeout to values that exceed transaction.max.timeout.ms setting on the broker (if the timeout is not set explicitly on the client and the two phase commit is set to ‘true’ then the transaction never expires).

transaction.timeout.ms The semantics is not changed, but it can be set to values that exceed transaction.max.timeout.ms if two.phase.commit.enable is set to ‘true’.

Broker Configuration Changes

transaction.two.phase.commit.enable The default would be ‘false’.  If it’s ‘false’, 2PC functionality is disabled even if the ACL is set, clients that attempt to use this functionality would receive TRANSACTIONAL_ID_AUTHORIZATION_FAILED error.

KafkaProducer API Changes

New KafkaProducer.PreparedTxnState class is going to be defined as following:

static public class PreparedTxnState {
  public String toString();
  public PreparedTxnState(String serializedState);
  public PreparedTxnState();
}

The objects of this class can serialize to / deserialize from a string value and can be written to / read from a database.

New overloaded method will be added to KafkaProducer:

public void initTransactions(boolean keepPreparedTxn) 

If the value is 'true' then the corresponding field is set in the InitProducerIdRequest and the KafkaProducer object is set into a state which only allows calling .commitTransaction or .abortTransaction.

If the transaction.two.phase.commit.enable setting is set to ‘false’ but keepPreparedTxn is set to ‘true’ then the call fails with the INVALID_TXN_STATE error.

New method will be added to KafkaProducer:

public PreparedTxnState prepareTransaction() 

This would flush all the pending messages and transition the producer into a mode where only .commitTransaction, .abortTransaction, or .completeTransaction could be called.  If the call is successful (all messages successfully got flushed to all partitions) the transaction is prepared.  If the 2PC is not enabled, we return the INVALID_TXN_STATE error.

New method would be added to KafkaProducer:

public void completeTransaction(PreparedTxnState preparedTxnState)

The method would compare the currently prepared transaction state and the state passed in the argument and either commit or abort the transaction.  If the producer is not in prepared state (i.e. either prepareTransaction was called or initTransaction(true) was called) we return an INVALID_TXN_STATE error.

AdminClient API Changes

Class ListTransactionsOptions will support 2 new methods:

public ListTransactionsOptions runningLongerThanMs(long runningLongerThanMs) // set 

public long runningLongerThanMs() // get 

The Admin  interface will support a new method:

public TerminateTransactionResult forceTerminateTransaction(String transactionalId) 

TerminateTransactionResult just contains KafkaFuture<void> result method.

NOTE that there is an existing abortTransaction  method that is used to abort “hanging” transactions (artifact of some gaps in the transaction protocol implementation that will be addressed in KIP-890, i.e. once part 1 of KIP-890 is implemented we won’t have “hanging” transactions).  “Hanging” transactions are not known to the Kafka transaction coordinator, they are just dangling messages in data partitions that cannot be aborted via the normal transaction protocol.  So abortTransaction actually needs information about data partitions so that it could go and insert markers directly there.

On the other hand, the forceTerminateTransaction method would operate on a well-formed, but long running transaction for a given transactional id.  Under the covers it would just use InitProducerId call with keepPreparedTxn=false.

ACL Changes

A new value will be added to the enum AclOperation: TWO_PHASE_COMMIT ((byte) 15 .  When InitProducerId comes with enable2Pc=true, it would have to have both WRITE and TWO_PHASE_COMMIT operation enabled on the transactional id resource.

Command Line Tool Changes

The kafka-transactions.sh --list command is going to support a new flag --runningLongerThanMs that would take the number of milliseconds.  If this flag is provided, only transactions that have been running longer than the specified number of milliseconds would be listed.

The kafka-transactions.sh tool is going to support a new command --forceTerminateTransaction.  It has one required argument --transactionalId that would take the transactional id for the transaction to be terminated.

The kafka-acls.sh tool is going to support a new --operation TwoPhaseCommit.

Compatibility, Deprecation, and Migration Plan

The proposal doesn't remove or update any existing functionality, it just adds new functionality that would only be executed if the new configurations and APIs are used.

Test Plan

The corresponding unit an integration tests will be added.

Rejected Alternatives

Explicit “prepare” RPC

Given that the 2PC protocol is defined in terms of “prepare” and “commit” phases it seems natural to just add a “prepare” RPC to Kafka.  The RPC would tell the Kafka transaction coordinator to transition the transaction into a new “prepared” state (note that our current state names are misleading – PREPARE_COMMIT is actually the “commit” phase).

There are some potential benefits of doing that:

1. Transactions that haven’t reached “prepared” state can be aborted via timeout.
2. New updates to “prepared” transactions can be rejected.
3. InitProducerId would know to not abort prepared transactions.
4. We could query for prepared transactions.

The disadvantage of an explicit “prepare” state is that we’d need to run a synchronous operation on the Kafka transaction coordinator topic and (if we want to support the benefit 2) send “prepare” markers to all leaders (so that they could bounce off new updates).

At a closer examination the benefits are not really eliminating any complexities:

1. We still need tooling and operational support to handle transactions that are stuck in the “prepared” state.
2. The external transaction coordinator would have to keep the state anyway and would know to not send any messages to prepared transactions.
3. The external transaction coordinator would have to keep the state anyway and can pass this info to InitProducerId (via the keepPreparedTxn flag) instead of keeping this info in 2 places.
4. The external transaction coordinator has already the knowledge of prepared transactions.

So we decided to keep the “implicit prepare” the same way we have it in Kafka today and avoid extra synchronous operations that would just duplicate the state that is kept in the external transaction coordinator.

HeartBeat RPC

Potentially, it could be good to distinguish between abandoned transactions (i.e. a producer started a 2PC transaction and then died) and just long running transactions.  We could add a heartbeat RPC between the 2 cases.

HeartBeat RPC definitely sounds like a “good thing to do”.  It is not clear, though, what would be the cases when we need to handle these situations differently – a long running transaction is a concern regardless of whether it got abandoned or the application is still working on it (also the application might have a bug or etc.), the operator would still need to investigate what’s going on with the application and determine if it’s safe to abort the transaction without violating “prepared” guarantees.

So it doesn’t seem to justify extra complexity, and we just add tooling for inspection of any long running transactions.

Using TransactionTimeoutMs=MAX_INT Instead of Enable2Pc

Having a Boolean flag and then a timeout value seems redundant, we really need just either one of the other, so technically instead of adding an independent flag we could use a special timeout value to indicate that it’s a 2PC transaction.  This, however, would couple intent with specific implementation; an explicit Boolean seems to reflect the intent better.

Allowing keepPreparedTxn=true without 2PC

Without 2PC the notion of “prepared” transaction is subverted by the transaction.max.timeout.ms so Kafka cannot promise to keep transaction in-doubt until a decision is reached.  So from a purity perspective using keepPreparedTxn=true doesn’t reflect the semantics.

In practice, however, Flink already effectively has a way to keep a “prepared” transaction by using reflection, so if we want to support all the following properties:

1. Get rid of reflection use in Flink and move to public API in new versions of Flink
2. Support new versions of Flink in Kafka clusters that don’t want to grant 2PC privileges.

Then we need to enable using keepPreparedTxn=true even if 2PC is disabled.

Using Partition Offsets as PreparedTxnState

With KIP-890, we can identify transactions as {producerId, epoch}.  Without KIP-890 the only way to remember a prepared transaction state is to keep track of partitions and their offsets.  The first transaction offsets are already tracked on the partition leaders, so this information is present in the cluster in some form but getting it requires calling partitions leader, which brings complexity that doesn’t exist with KIP-890.

Another disadvantage is that the PrepareTxnState isn’t bounded if it tracks partitions and their offsets – some producers would produce data into topics with a large number of partitions, some producers would produce data into topics with a small number of partitions.  Thus, the solution could work with some configurations (e.g. it was tested with VARCHAR(255)) and break when topic-partition properties get changed.  On the other hand, {producerId, epoch} has a small and fixed size.