Versions Compared

Key

  • This line was added.
  • This line was removed.
  • Formatting was changed.
Comment: Initial draft version

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 stateDraft

...

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 (

Jira
serverASF JIRA
serverId5aa69414-a9e9-3523-82ec-879b028fb15b
keyKAFKA-20381
and
Jira
serverASF JIRA
serverId5aa69414-a9e9-3523-82ec-879b028fb15b
keyKAFKA-19873
, 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.

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:

RPCOperationResource
TxnHeartbeat Transactional IDWRITE

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.

ConfigurationDescriptionValues
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

...