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:   [One of "Under Discussion", "Accepted", "Rejected"]

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).

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

...

7983

Motivation

For each partition of a Kafka topic, a set of in-sync replicas is maintained. If a replica becomes out-of-sync, it continually fetches missing data from the partition leader until it is back in sync. 

Occasionally, we notice that large amounts of data need to be replicated to maintain the ISR set. This can happen if a broker is offline for hours, perhaps due to machine maintenance or hardware failures, or when a large number of partitions are reassigned.

Without any replication quotas, we observe significantly increased end-to-end latency on the cluster, measured by Xinfra Monitor. In the worst cases, tail latency jumps from 10s of milliseconds to multiple seconds. 

To address this, replication quotas were introduced in KIP-73, and KIP-542 improved upon replication throttling by applying it to out-of-sync replicas only. However, these throttles can only be configured for a specific topics and partitions. While this satisfies the need to throttle out-of-sync replication traffic while reassigning partitions, this solution is not sufficient at a larger scale, when an entire broker’s traffic needs to be throttled.

Currently, replication throttles can only be set for a specific topic and/or partitions, and not as a blanket configuration for the whole broker. With the current configuration options for replication throttles, the only way to throttle an entire broker’s traffic is to do the following:

1. Set leader.replication.throttled.rate and follower.replication.throttled.rate at the broker level.
2. For each topic on the broker, configure leader.replication.throttled.replicas and follower.replication.throttled.replicas.

Depending on how large the broker is, this process can be unnecessarily time consuming and expensive.

Opening this KIP to go along with the linked JIRA.

Public Interfaces

Two new dynamic broker configurations will be introduced to indicate if replication throttling is enabled for the entire broker.

New configurations:

• leader.replication.throttled - boolean representing if leader replication traffic is throttled on the whole broker

• follower.replication.throttled - boolean representing if follower replication traffic is throttled on the whole broker

This is an example of how leader replication throttling can be enabled for the whole broker:

bin/kafka-configs … --alter
--add-config 'leader.replication.throttled=true'
--entity-type broker
--entity-name brokerId

Proposed Changes

Currently, setting replication throttles at the topic and partition levels is supported. This KIP proposes changing the replication logic to also support configuring throttles at the broker level - meaning all replication traffic to/from any partition on a specified broker will be throttled.

This will be done by introducing two new dynamic broker configurations - leader.replication.throttled and follower.replication.throttled, which are boolean values representing if leader/follower replication traffic is throttled on the whole broker. This also involves adding the configuration options, updating those values accordingly in the ConfigHandler, and updating the isThrottled function in the ReplicationQuotaManager.

With these proposed changes, an admin can throttle an entire broker’s traffic by doing the following:

1. Set leader.replication.throttled.rate and follower.replication.throttled.rate at the broker level.
2. Set leader.replication.throttled=true and follower.replication.throttled=true at the broker level.

Compatibility, Deprecation, and Migration Plan

• What impact (if any) will there be on existing users?
• If we are changing behavior how will we phase out the older behavior?
• If we need special migration tools, describe them here.
• When will we remove the existing behavior?

Test Plan

Describe in few sentences how the KIP will be tested. We are mostly interested in system tests (since unit-tests are specific to implementation details). How will we know that the implementation works as expected? How will we know nothing broke?

Rejected Alternatives

As the only exposed changes are configuration options, there is no impact to existing users.

Test Plan

New tests will be added to the ReplicationQuotaManagerTest suite.

Rejected Alternatives

This is not a rejected alternative to this KIP, but somewhat related and worth mentioning. 

KAFKA-10190 proposes a change to apply follower.replication.throttled.rate, leader.replication.throttled.rate and replica.alter.log.dirs.io.max.bytes.per.second configs at the broker level, but seems to have been abandoned. Unlike this KIP, KAFKA-10190 does not enable applying throttles at the broker level, just setting the throttle rate.

However, this could be a useful feature to include in this KIP as wellIf there are alternative ways of accomplishing the same thing, what were they? The purpose of this section is to motivate why the design is the way it is and not some other way.