[jira] [Created] (KAFKA-12549) Allow state stores to opt-in transactional support

["Guozhang Wang (Jira)" <ji...@apache.org>]

Guozhang Wang created KAFKA-12549:
-------------------------------------

             Summary: Allow state stores to opt-in transactional support
                 Key: KAFKA-12549
                 URL: https://issues.apache.org/jira/browse/KAFKA-12549
             Project: Kafka
          Issue Type: New Feature
          Components: streams
            Reporter: Guozhang Wang


Right now Kafka Stream's EOS implementation does not make any assumptions about the state store's transactional support. Allowing the state stores to optionally provide transactional support can have multiple benefits. E.g., if we add some APIs into the {{StateStore}} interface, like {{beginTxn}}, {{commitTxn}} and {{abortTxn}}. Then these APIs can be used under both ALOS and EOS such that:

* store.beginTxn
* store.put // during processing
* streams commit // either through eos protocol or not
* store.commitTxn

We can have the following benefits:
* Reduce the duplicated records upon crashes for ALOS (note this is not EOS still, but some middle-ground where uncommitted data within a state store would not be retained if store.commitTxn failed).
* No need to wipe the state store and re-bootstrap from scratch upon crashes for EOS. E.g., if a crash-failure happened between streams commit completes and store.commitTxn. We can instead just roll-forward the transaction by replaying the changelog from the second recent  streams committed offset towards the most recent committed offset.
* Remote stores that support txn then does not need to support wiping (https://issues.apache.org/jira/browse/KAFKA-12475).
* We can fix the known issues of emit-on-change (https://cwiki.apache.org/confluence/display/KAFKA/KIP-557%3A+Add+emit+on+change+support+for+Kafka+Streams).
* We can support "query committed data only" for interactive queries (see below for reasons).

As for the implementation of these APIs, there are several options:
* The state store itself have natural transaction features (e.g. RocksDB).
* Use an in-memory buffer for all puts within a transaction, and upon `commitTxn` write the whole buffer as a batch to the underlying state store, or just drop the whole buffer upon aborting. Then for interactive queries, one can optionally only query the underlying store for committed data only.
* Use a separate store as the transient persistent buffer. Upon `beginTxn` create a new empty transient store, and upon `commitTxn` merge the store into the underlying store. Same applies for interactive querying committed-only data.



--
This message was sent by Atlassian Jira
(v8.3.4#803005)