DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
| Table of Contents |
|---|
This page is meant as a template for writing a KIP. To create a KIP choose Tools->Copy on this page and modify with your content and replace the heading with the next KIP number and a description of your issue. Replace anything in italics with your own description.
Status
Current state: Draft
...
Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).
Motivation
Describe the problems you are trying to solve.
Public Interfaces
Briefly list any new interfaces that will be introduced as part of this proposal or any existing interfaces that will be removed or changed. The purpose of this section is to concisely call out the public contract that will come along with this feature.
A public interface is any change to the following:
Binary log format
The network protocol and api behavior
Any class in the public packages under clientsConfiguration, especially client configuration
org/apache/kafka/common/serialization
org/apache/kafka/common
org/apache/kafka/common/errors
org/apache/kafka/clients/producer
org/apache/kafka/clients/consumer (eventually, once stable)
Monitoring
Command line tools and arguments
- Anything else that will likely break existing users in some way when they upgrade
Proposed Changes
...
Kafka transactions are often used in combination with stream processing engines such as Apache Flink and Apache Spark, as well as with Kafka's native stream processing library, Kafka Streams. Users have highlighted issues (
and Jira server ASF JIRA serverId 5aa69414-a9e9-3523-82ec-879b028fb15b key KAFKA-20381
, for example) with how the transaction timeout is responsible for two potentially conflicting requirements. First, we want to be able to ensure that the client applications performing the transactions are still alive, and second, we want to ensure that the transactions have sufficient time to complete. If users set the transaction timeout to be low, transactions are rolled back promptly when the applications fail, but there is a risk that transactions are aborted preemptively because there was not quite enough time. Conversely, if users set the transaction timeout to be high, transactions have sufficient time to complete, but the delay before a failing application's transaction is aborted is also very long.Jira server ASF JIRA serverId 5aa69414-a9e9-3523-82ec-879b028fb15b key KAFKA-19873
This KIP proposes to introduce an explicit liveness check for transactions so that users can benefit from prompt recovery from failed applications when they need to allow the transaction timeout to be long.
Proposed Changes
Public Interfaces
Kafka Protocol Changes
This KIP introduces the following new API:
- TxnHeartbeat
Access Control
This table gives the ACLs required for the new API:
| RPC | Operation | Resource |
|---|---|---|
TxnHeartbeat | Transactional ID | WRITE |
InitProducerId API
The KIP introduces version 7 (or version 6 if this KIP is adopted before KIP-939's APIs are marked as stable).
Request schema
| Code Block |
|---|
{
"apiKey": 22,
"type": "request",
"listeners": ["broker"],
"name": "InitProducerIdRequest",
// Version 1 is the same as version 0.
//
// Version 2 is the first flexible version.
//
// Version 3 adds ProducerId and ProducerEpoch, allowing producers to try to resume after an INVALID_PRODUCER_EPOCH error
//
// Version 4 adds the support for new error code PRODUCER_FENCED.
//
// Version 5 adds support for new error code TRANSACTION_ABORTABLE (KIP-890).
//
// Version 6 adds support for 2PC (KIP-939).
//
// Version 7 adds support for transaction heartbeats (KIP-1309).
"validVersions": "0-7",
"flexibleVersions": "2+",
"fields": [
{ "name": "TransactionalId", "type": "string", "versions": "0+", "nullableVersions": "0+", "entityType": "transactionalId",
"about": "The transactional id, or null if the producer is not transactional." },
{ "name": "TransactionTimeoutMs", "type": "int32", "versions": "0+",
"about": "The time in ms to wait before aborting idle transactions sent by this producer. This is only relevant if a TransactionalId has been defined." },
{ "name": "TransactionHeartbeatIntervalMs", "type": "int32", "versions": "7+", "default": "0", "ignorable": "true",
"about": "The time in ms between heartbeats from the producer. If >0, idle transactions sent by this producer are aborted after 3 times this interval." },
{ "name": "ProducerId", "type": "int64", "versions": "3+", "default": "-1", "entityType": "producerId",
"about": "The producer id. This is used to disambiguate requests if a transactional id is reused following its expiration." },
{ "name": "ProducerEpoch", "type": "int16", "versions": "3+", "default": "-1",
"about": "The producer's current epoch. This will be checked against the producer epoch on the broker, and the request will return an error if they do not match." },
{ "name": "Enable2Pc", "type": "bool", "versions": "6+", "default": "false",
"about": "True if the client wants to enable two-phase commit (2PC) protocol for transactions." },
{ "name": "KeepPreparedTxn", "type": "bool", "versions": "6+", "default": "false",
"about": "True if the client wants to keep the currently ongoing transaction instead of aborting it." }
]
} |
This KIP adds the new field TransactionalHeartbeatIntervalMs .
Response schema
Version 7 is the same as version 6.
If TransactionHeartbeatIntervalMs is non-zero and Enable2Pc is true, the error code is INVALID_REQUEST .
If TransactionHeartbeatIntervalMs is non-zero and greater than TransactionTimeoutMs , the error code is INVALID_REQUEST .
TxnHeartbeat API
The TxnHeartbeat API is used by transactional producers to demonstrate their liveness to the transaction coordinator. They are only sent by transactional producers when heartbeats are enabled and when two-phase commit is not being used, typically because it is desirable to detect producer failures more quickly than the transaction timeout.
Request schema
| Code Block |
|---|
{
"apiKey": TBD,
"type": "request",
"listeners": ["broker"],
"name": "TxnHeartbeatRequest",
// Version 0 is the first version (KIP-1309).
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
{ "name": "TransactionalId", "type": "string", "versions": "0+", "entityType": "transactionalId",
"about": "The transactional id." },
{ "name": "ProducerId", "type": "int64", "versions": "0+", "default": "-1", "entityType": "producerId",
"about": "The producer id. This is used to disambiguate requests if a transactional id is reused following its expiration." },
{ "name": "ProducerEpoch", "type": "int16", "versions": "0+", "default": "-1",
"about": "The producer's current epoch. This will be checked against the producer epoch on the broker, and the request will return an error if they do not match." },
]
} |
Response schema
| Code Block |
|---|
{
"apiKey": TBD,
"type": "response",
"name": "TxnHeartbeatResponse",
// Version 0 is the first version (KIP-1309).
"validVersions": "0",
"flexibleVersions": "0+",
// Supported errors:
// - TRANSACTIONAL_ID_AUTHORIZATION_FAILED (version 0+)
// - TRANSACTIONAL_ID_NOT_FOUND (version 0+)
// - NOT_COORDINATOR (version 0+)
// - COORDINATOR_NOT_AVAILABLE (version 0+)
// - COORDINATOR_LOAD_IN_PROGRESS (version 0+)
// - INVALID_REQUEST (version 1+)
"fields": [
{ "name": "ThrottleTimeMs", "type": "int32", "versions": "0+", "ignorable": true,
"about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
{ "name": "ErrorCode", "type": "int16", "versions": "0+",
"about": "The error code, or 0 if there was no error." },
{ "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "The error message, or null if there was no error." }
]
} |
Maybe need to include TRANSACTION_ABORTABLE and UNKNOWN_PRODUCER_ID error codes here.
Configuration
Producer configuration
The following configuration property is added for producers.
| Configuration | Description | Values |
|---|---|---|
transaction.heartbeat.interval.ms | The expected time between heartbeats to the transaction coordinator. Heartbeats can be used to ensure that a transactional producer remains alive and it is desirable to ensure liveness more frequently than the transaction.timeout.ms . When enabled, a transactional producer will send heartbeats at this interval to ensure that the transaction remains open before the coordinator proactively aborts it. If the transaction coordinator does not receive a heartbeat after 3 times this interval, it will proactively abort the transaction. | Default: 0 (disabled) |
Compatibility, Deprecation, and Migration Plan
...