ops.html

<!--
 Licensed to the Apache Software Foundation (ASF) under one or more
 contributor license agreements.  See the NOTICE file distributed with
 this work for additional information regarding copyright ownership.
 The ASF licenses this file to You under the Apache License, Version 2.0
 (the "License"); you may not use this file except in compliance with
 the License.  You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

 Unless required by applicable law or agreed to in writing, software
 distributed under the License is distributed on an "AS IS" BASIS,
 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 See the License for the specific language governing permissions and
 limitations under the License.
-->

<script id="ops-template" type="text/x-handlebars-template">

  <p>Here is some information on actually running Kafka as a production system. Please send us any additional tips you know of.</p>

  <h3 class="anchor-heading"><a id="basic_ops" class="anchor-link"></a><a href="#basic_ops">6.1 Basic Kafka Operations</a></h3>

  This section will review the most common operations you will perform on your Kafka cluster. All of the tools reviewed in this section are available under the <code>bin/</code> directory of the Kafka distribution and each tool will print details on all possible commandline options if it is run with no arguments.

  <h4 class="anchor-heading"><a id="basic_ops_add_topic" class="anchor-link"></a><a href="#basic_ops_add_topic">Adding and removing topics</a></h4>

  You have the option of either adding topics manually or having them be created automatically when data is first published to a non-existent topic. If topics are auto-created then you may want to tune the default <a href="#topicconfigs">topic configurations</a> used for auto-created topics.
  <p>
  Topics are added and modified using the topic tool:
  <pre><code class="language-bash">$ bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic my_topic_name \
    --partitions 20 --replication-factor 3 --config x=y</code></pre>
  The replication factor controls how many servers will replicate each message that is written. If you have a replication factor of 3 then up to 2 servers can fail before you will lose access to your data. We recommend you use a replication factor of 2 or 3 so that you can transparently bounce machines without interrupting data consumption.
  <p>
  The partition count controls how many logs the topic will be sharded into. There are several impacts of the partition count. First each partition must fit entirely on a single server. So if you have 20 partitions the full data set (and read and write load) will be handled by no more than 20 servers (not counting replicas). Finally the partition count impacts the maximum parallelism of your consumers. This is discussed in greater detail in the <a href="#intro_consumers">concepts section</a>.
  <p>
  Each sharded partition log is placed into its own folder under the Kafka log directory. The name of such folders consists of the topic name, appended by a dash (-) and the partition id. Since a typical folder name can not be over 255 characters long, there will be a limitation on the length of topic names. We assume the number of partitions will not ever be above 100,000. Therefore, topic names cannot be longer than 249 characters. This leaves just enough room in the folder name for a dash and a potentially 5 digit long partition id.
  <p>
  The configurations added on the command line override the default settings the server has for things like the length of time data should be retained. The complete set of per-topic configurations is documented <a href="#topicconfigs">here</a>.

  <h4 class="anchor-heading"><a id="basic_ops_modify_topic" class="anchor-link"></a><a href="#basic_ops_modify_topic">Modifying topics</a></h4>

  You can change the configuration or partitioning of a topic using the same topic tool.
  <p>
  To add partitions you can do
  <pre><code class="language-bash">$ bin/kafka-topics.sh --bootstrap-server localhost:9092 --alter --topic my_topic_name \
    --partitions 40</code></pre>
  Be aware that one use case for partitions is to semantically partition data, and adding partitions doesn't change the partitioning of existing data so this may disturb consumers if they rely on that partition. That is if data is partitioned by <code>hash(key) % number_of_partitions</code> then this partitioning will potentially be shuffled by adding partitions but Kafka will not attempt to automatically redistribute data in any way.
  <p>
  To add configs:
  <pre><code class="language-bash">$ bin/kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --entity-name my_topic_name --alter --add-config x=y</code></pre>
  To remove a config:
  <pre><code class="language-bash">$ bin/kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --entity-name my_topic_name --alter --delete-config x</code></pre>
  And finally deleting a topic:
  <pre><code class="language-bash">$ bin/kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic my_topic_name</code></pre>
  <p>
  Kafka does not currently support reducing the number of partitions for a topic.
  <p>
  Instructions for changing the replication factor of a topic can be found <a href="#basic_ops_increase_replication_factor">here</a>.

  <h4 class="anchor-heading"><a id="basic_ops_restarting" class="anchor-link"></a><a href="#basic_ops_restarting">Graceful shutdown</a></h4>

  The Kafka cluster will automatically detect any broker shutdown or failure and elect new leaders for the partitions on that machine. This will occur whether a server fails or it is brought down intentionally for maintenance or configuration changes. For the latter cases Kafka supports a more graceful mechanism for stopping a server than just killing it.

  When a server is stopped gracefully it has two optimizations it will take advantage of:
  <ol>
      <li>It will sync all its logs to disk to avoid needing to do any log recovery when it restarts (i.e. validating the checksum for all messages in the tail of the log). Log recovery takes time so this speeds up intentional restarts.
      <li>It will migrate any partitions the server is the leader for to other replicas prior to shutting down. This will make the leadership transfer faster and minimize the time each partition is unavailable to a few milliseconds.
  </ol>

  Syncing the logs will happen automatically whenever the server is stopped other than by a hard kill, but the controlled leadership migration requires using a special setting:
  <pre><code class="language-text">controlled.shutdown.enable=true</code></pre>
  Note that controlled shutdown will only succeed if <i>all</i> the partitions hosted on the broker have replicas (i.e. the replication factor is greater than 1 <i>and</i> at least one of these replicas is alive). This is generally what you want since shutting down the last replica would make that topic partition unavailable.

  <h4 class="anchor-heading"><a id="basic_ops_leader_balancing" class="anchor-link"></a><a href="#basic_ops_leader_balancing">Balancing leadership</a></h4>

  Whenever a broker stops or crashes, leadership for that broker's partitions transfers to other replicas. When the broker is restarted it will only be a follower for all its partitions, meaning it will not be used for client reads and writes.
  <p>
  To avoid this imbalance, Kafka has a notion of preferred replicas. If the list of replicas for a partition is 1,5,9 then node 1 is preferred as the leader to either node 5 or 9 because it is earlier in the replica list. By default the Kafka cluster will try to restore leadership to the preferred replicas. This behaviour is configured with:

  <pre><code class="language-text">auto.leader.rebalance.enable=true</code></pre>
    You can also set this to false, but you will then need to manually restore leadership to the restored replicas by running the command:
  <pre><code class="language-bash">$ bin/kafka-leader-election.sh --bootstrap-server localhost:9092 --election-type preferred --all-topic-partitions</code></pre>

  <h4 class="anchor-heading"><a id="basic_ops_racks" class="anchor-link"></a><a href="#basic_ops_racks">Balancing Replicas Across Racks</a></h4>
  The rack awareness feature spreads replicas of the same partition across different racks. This extends the guarantees Kafka provides for broker-failure to cover rack-failure, limiting the risk of data loss should all the brokers on a rack fail at once. The feature can also be applied to other broker groupings such as availability zones in EC2.
  <p></p>
  You can specify that a broker belongs to a particular rack by adding a property to the broker config:
  <pre><code class="language-text">broker.rack=my-rack-id</code></pre>
  When a topic is <a href="#basic_ops_add_topic">created</a>, <a href="#basic_ops_modify_topic">modified</a> or replicas are <a href="#basic_ops_cluster_expansion">redistributed</a>, the rack constraint will be honoured, ensuring replicas span as many racks as they can (a partition will span min(#racks, replication-factor) different racks).
  <p></p>
  The algorithm used to assign replicas to brokers ensures that the number of leaders per broker will be constant, regardless of how brokers are distributed across racks. This ensures balanced throughput.
  <p></p>
  However if racks are assigned different numbers of brokers, the assignment of replicas will not be even. Racks with fewer brokers will get more replicas, meaning they will use more storage and put more resources into replication. Hence it is sensible to configure an equal number of brokers per rack.

  <h4 class="anchor-heading"><a id="basic_ops_mirror_maker" class="anchor-link"></a><a href="#basic_ops_mirror_maker">Mirroring data between clusters & Geo-replication</a></h4>

  <p>
  Kafka administrators can define data flows that cross the boundaries of individual Kafka clusters, data centers, or geographical regions. Please refer to the section on <a href="#georeplication">Geo-Replication</a> for further information.
  </p>

  <h4 class="anchor-heading"><a id="basic_ops_consumer_lag" class="anchor-link"></a><a href="#basic_ops_consumer_lag">Checking consumer position</a></h4>
  Sometimes it's useful to see the position of your consumers. We have a tool that will show the position of all consumers in a consumer group as well as how far behind the end of the log they are. To run this tool on a consumer group named <i>my-group</i> consuming a topic named <i>my-topic</i> would look like this:
  <pre><code class="language-bash">$ bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group my-group
TOPIC                          PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG        CONSUMER-ID                                       HOST                           CLIENT-ID
my-topic                       0          2               4               2          consumer-1-029af89c-873c-4751-a720-cefd41a669d6   /127.0.0.1                     consumer-1
my-topic                       1          2               3               1          consumer-1-029af89c-873c-4751-a720-cefd41a669d6   /127.0.0.1                     consumer-1
my-topic                       2          2               3               1          consumer-2-42c1abd4-e3b2-425d-a8bb-e1ea49b29bb2   /127.0.0.1                     consumer-2</code></pre>

  <h4 class="anchor-heading"><a id="basic_ops_consumer_group" class="anchor-link"></a><a href="#basic_ops_consumer_group">Managing Consumer Groups</a></h4>

  With the ConsumerGroupCommand tool, we can list, describe, or delete the consumer groups. The consumer group can be deleted manually, or automatically when the last committed offset for that group expires. Manual deletion works only if the group does not have any active members.

  For example, to list all consumer groups across all topics:

  <pre><code class="language-bash">$ bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list
test-consumer-group</code></pre>

  To view offsets, as mentioned earlier, we "describe" the consumer group like this:

  <pre><code class="language-bash">$ bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group my-group
TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID                                    HOST            CLIENT-ID
topic3          0          241019          395308          154289          consumer2-e76ea8c3-5d30-4299-9005-47eb41f3d3c4 /127.0.0.1      consumer2
topic2          1          520678          803288          282610          consumer2-e76ea8c3-5d30-4299-9005-47eb41f3d3c4 /127.0.0.1      consumer2
topic3          1          241018          398817          157799          consumer2-e76ea8c3-5d30-4299-9005-47eb41f3d3c4 /127.0.0.1      consumer2
topic1          0          854144          855809          1665            consumer1-3fc8d6f1-581a-4472-bdf3-3515b4aee8c1 /127.0.0.1      consumer1
topic2          0          460537          803290          342753          consumer1-3fc8d6f1-581a-4472-bdf3-3515b4aee8c1 /127.0.0.1      consumer1
topic3          2          243655          398812          155157          consumer4-117fe4d3-c6c1-4178-8ee9-eb4a3954bee0 /127.0.0.1      consumer4</code></pre>

  There are a number of additional "describe" options that can be used to provide more detailed information about a consumer group:
  <ul>
    <li>--members: This option provides the list of all active members in the consumer group.
      <pre><code class="language-bash">$ bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group my-group --members
CONSUMER-ID                                    HOST            CLIENT-ID       #PARTITIONS
consumer1-3fc8d6f1-581a-4472-bdf3-3515b4aee8c1 /127.0.0.1      consumer1       2
consumer4-117fe4d3-c6c1-4178-8ee9-eb4a3954bee0 /127.0.0.1      consumer4       1
consumer2-e76ea8c3-5d30-4299-9005-47eb41f3d3c4 /127.0.0.1      consumer2       3
consumer3-ecea43e4-1f01-479f-8349-f9130b75d8ee /127.0.0.1      consumer3       0</code></pre>
    </li>
    <li>--members --verbose: On top of the information reported by the "--members" options above, this option also provides the partitions assigned to each member.
      <pre><code class="language-bash">$ bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group my-group --members --verbose
CONSUMER-ID                                    HOST            CLIENT-ID       #PARTITIONS     ASSIGNMENT
consumer1-3fc8d6f1-581a-4472-bdf3-3515b4aee8c1 /127.0.0.1      consumer1       2               topic1(0), topic2(0)
consumer4-117fe4d3-c6c1-4178-8ee9-eb4a3954bee0 /127.0.0.1      consumer4       1               topic3(2)
consumer2-e76ea8c3-5d30-4299-9005-47eb41f3d3c4 /127.0.0.1      consumer2       3               topic2(1), topic3(0,1)
consumer3-ecea43e4-1f01-479f-8349-f9130b75d8ee /127.0.0.1      consumer3       0               -</code></pre>
    </li>
    <li>--offsets: This is the default describe option and provides the same output as the "--describe" option.</li>
    <li>--state: This option provides useful group-level information.
      <pre><code class="language-bash">$ bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group my-group --state
COORDINATOR (ID)          ASSIGNMENT-STRATEGY       STATE                #MEMBERS
localhost:9092 (0)        range                     Stable               4</code></pre>
    </li>
  </ul>

  To manually delete one or multiple consumer groups, the "--delete" option can be used:
  <pre><code class="language-bash">$ bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --delete --group my-group --group my-other-group
Deletion of requested consumer groups ('my-group', 'my-other-group') was successful.</code></pre>

  <p>
  To reset offsets of a consumer group, "--reset-offsets" option can be used.
  This option supports one consumer group at the time. It requires defining following scopes: --all-topics or --topic. One scope must be selected, unless you use '--from-file' scenario. Also, first make sure that the consumer instances are inactive.
  See <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-122%3A+Add+Reset+Consumer+Group+Offsets+tooling">KIP-122</a> for more details.

  <p>
  It has 3 execution options:
  <ul>
    <li>
      (default) to display which offsets to reset.
    </li>
    <li>
      --execute : to execute --reset-offsets process.
    </li>
    <li>
      --export : to export the results to a CSV format.
    </li>
  </ul>

  <p>
  --reset-offsets also has the following scenarios to choose from (at least one scenario must be selected):
  <ul>
    <li>
      --to-datetime &lt;String: datetime&gt; : Reset offsets to offsets from datetime. Format: 'YYYY-MM-DDTHH:mm:SS.sss'
    </li>
    <li>
      --to-earliest : Reset offsets to earliest offset.
    </li>
    <li>
      --to-latest : Reset offsets to latest offset.
    </li>
    <li>
      --shift-by &lt;Long: number-of-offsets&gt; : Reset offsets shifting current offset by 'n', where 'n' can be positive or negative.
    </li>
    <li>
      --from-file : Reset offsets to values defined in CSV file.
    </li>
    <li>
      --to-current : Resets offsets to current offset.
    </li>
    <li>
      --by-duration &lt;String: duration&gt; : Reset offsets to offset by duration from current timestamp. Format: 'PnDTnHnMnS'
    </li>
    <li>
      --to-offset : Reset offsets to a specific offset.
    </li>
  </ul>

  Please note, that out of range offsets will be adjusted to available offset end. For example, if offset end is at 10 and offset shift request is
  of 15, then, offset at 10 will actually be selected.

  <p>
  For example, to reset offsets of a consumer group to the latest offset:

  <pre><code class="language-bash">$ bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --reset-offsets --group consumergroup1 --topic topic1 --to-latest
TOPIC                          PARTITION  NEW-OFFSET
topic1                         0          0</code></pre>

  <p>

  <h4 class="anchor-heading"><a id="basic_ops_cluster_expansion" class="anchor-link"></a><a href="#basic_ops_cluster_expansion">Expanding your cluster</a></h4>

  Adding servers to a Kafka cluster is easy, just assign them a unique broker id and start up Kafka on your new servers. However these new servers will not automatically be assigned any data partitions, so unless partitions are moved to them they won't be doing any work until new topics are created. So usually when you add machines to your cluster you will want to migrate some existing data to these machines.
  <p>
  The process of migrating data is manually initiated but fully automated. Under the covers what happens is that Kafka will add the new server as a follower of the partition it is migrating and allow it to fully replicate the existing data in that partition. When the new server has fully replicated the contents of this partition and joined the in-sync replica one of the existing replicas will delete their partition's data.
  <p>
  The partition reassignment tool can be used to move partitions across brokers. An ideal partition distribution would ensure even data load and partition sizes across all brokers. The partition reassignment tool does not have the capability to automatically study the data distribution in a Kafka cluster and move partitions around to attain an even load distribution. As such, the admin has to figure out which topics or partitions should be moved around.
  <p>
  The partition reassignment tool can run in 3 mutually exclusive modes:
  <ul>
  <li>--generate: In this mode, given a list of topics and a list of brokers, the tool generates a candidate reassignment to move all partitions of the specified topics to the new brokers. This option merely provides a convenient way to generate a partition reassignment plan given a list of topics and target brokers.</li>
  <li>--execute: In this mode, the tool kicks off the reassignment of partitions based on the user provided reassignment plan. (using the --reassignment-json-file option). This can either be a custom reassignment plan hand crafted by the admin or provided by using the --generate option</li>
  <li>--verify: In this mode, the tool verifies the status of the reassignment for all partitions listed during the last --execute. The status can be either of successfully completed, failed or in progress</li>
  </ul>
  <h5 class="anchor-heading"><a id="basic_ops_automigrate" class="anchor-link"></a><a href="#basic_ops_automigrate">Automatically migrating data to new machines</a></h5>
  The partition reassignment tool can be used to move some topics off of the current set of brokers to the newly added brokers. This is typically useful while expanding an existing cluster since it is easier to move entire topics to the new set of brokers, than moving one partition at a time. When used to do this, the user should provide a list of topics that should be moved to the new set of brokers and a target list of new brokers. The tool then evenly distributes all partitions for the given list of topics across the new set of brokers. During this move, the replication factor of the topic is kept constant. Effectively the replicas for all partitions for the input list of topics are moved from the old set of brokers to the newly added brokers.
  <p>
  For instance, the following example will move all partitions for topics foo1,foo2 to the new set of brokers 5,6. At the end of this move, all partitions for topics foo1 and foo2 will <i>only</i> exist on brokers 5,6.
  <p>
  Since the tool accepts the input list of topics as a json file, you first need to identify the topics you want to move and create the json file as follows:
  <pre><code class="language-bash">$ cat topics-to-move.json
{
  "topics": [
    { "topic": "foo1" },
    { "topic": "foo2" }
  ],
  "version": 1
}</code></pre>
  Once the json file is ready, use the partition reassignment tool to generate a candidate assignment:
  <pre><code class="language-bash">$ bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --topics-to-move-json-file topics-to-move.json --broker-list "5,6" --generate
Current partition replica assignment
{"version":1,
 "partitions":[{"topic":"foo1","partition":0,"replicas":[2,1],"log_dirs":["any"]},
               {"topic":"foo1","partition":1,"replicas":[1,3],"log_dirs":["any"]},
               {"topic":"foo1","partition":2,"replicas":[3,4],"log_dirs":["any"]},
               {"topic":"foo2","partition":0,"replicas":[4,2],"log_dirs":["any"]},
               {"topic":"foo2","partition":1,"replicas":[2,1],"log_dirs":["any"]},
               {"topic":"foo2","partition":2,"replicas":[1,3],"log_dirs":["any"]}]
}

Proposed partition reassignment configuration
{"version":1,
 "partitions":[{"topic":"foo1","partition":0,"replicas":[6,5],"log_dirs":["any"]},
               {"topic":"foo1","partition":1,"replicas":[5,6],"log_dirs":["any"]},
               {"topic":"foo1","partition":2,"replicas":[6,5],"log_dirs":["any"]},
               {"topic":"foo2","partition":0,"replicas":[5,6],"log_dirs":["any"]},
               {"topic":"foo2","partition":1,"replicas":[6,5],"log_dirs":["any"]},
               {"topic":"foo2","partition":2,"replicas":[5,6],"log_dirs":["any"]}]
}</code></pre>
  <p>
  The tool generates a candidate assignment that will move all partitions from topics foo1,foo2 to brokers 5,6. Note, however, that at this point, the partition movement has not started, it merely tells you the current assignment and the proposed new assignment. The current assignment should be saved in case you want to rollback to it. The new assignment should be saved in a json file (e.g. expand-cluster-reassignment.json) to be input to the tool with the --execute option as follows:
  <pre><code class="language-bash">$ bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --reassignment-json-file expand-cluster-reassignment.json --execute
Current partition replica assignment

{"version":1,
 "partitions":[{"topic":"foo1","partition":0,"replicas":[2,1],"log_dirs":["any"]},
               {"topic":"foo1","partition":1,"replicas":[1,3],"log_dirs":["any"]},
               {"topic":"foo1","partition":2,"replicas":[3,4],"log_dirs":["any"]},
               {"topic":"foo2","partition":0,"replicas":[4,2],"log_dirs":["any"]},
               {"topic":"foo2","partition":1,"replicas":[2,1],"log_dirs":["any"]},
               {"topic":"foo2","partition":2,"replicas":[1,3],"log_dirs":["any"]}]
}

Save this to use as the --reassignment-json-file option during rollback
Successfully started partition reassignments for foo1-0,foo1-1,foo1-2,foo2-0,foo2-1,foo2-2</code></pre>
  <p>
  Finally, the --verify option can be used with the tool to check the status of the partition reassignment. Note that the same expand-cluster-reassignment.json (used with the --execute option) should be used with the --verify option:
  <pre><code class="language-bash">$ bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --reassignment-json-file expand-cluster-reassignment.json --verify
Status of partition reassignment:
Reassignment of partition [foo1,0] is completed
Reassignment of partition [foo1,1] is still in progress
Reassignment of partition [foo1,2] is still in progress
Reassignment of partition [foo2,0] is completed
Reassignment of partition [foo2,1] is completed
Reassignment of partition [foo2,2] is completed</code></pre>

  <h5 class="anchor-heading"><a id="basic_ops_partitionassignment" class="anchor-link"></a><a href="#basic_ops_partitionassignment">Custom partition assignment and migration</a></h5>
  The partition reassignment tool can also be used to selectively move replicas of a partition to a specific set of brokers. When used in this manner, it is assumed that the user knows the reassignment plan and does not require the tool to generate a candidate reassignment, effectively skipping the --generate step and moving straight to the --execute step
  <p>
  For instance, the following example moves partition 0 of topic foo1 to brokers 5,6 and partition 1 of topic foo2 to brokers 2,3:
  <p>
  The first step is to hand craft the custom reassignment plan in a json file:
  <pre><code class="language-bash">$ cat custom-reassignment.json
{"version":1,"partitions":[{"topic":"foo1","partition":0,"replicas":[5,6]},{"topic":"foo2","partition":1,"replicas":[2,3]}]}</code></pre>
  Then, use the json file with the --execute option to start the reassignment process:
  <pre><code class="language-bash">$ bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --reassignment-json-file custom-reassignment.json --execute
Current partition replica assignment

{"version":1,
 "partitions":[{"topic":"foo1","partition":0,"replicas":[1,2],"log_dirs":["any"]},
               {"topic":"foo2","partition":1,"replicas":[3,4],"log_dirs":["any"]}]
}

Save this to use as the --reassignment-json-file option during rollback
Successfully started partition reassignments for foo1-0,foo2-1</code></pre>
  <p>
  The --verify option can be used with the tool to check the status of the partition reassignment. Note that the same custom-reassignment.json (used with the --execute option) should be used with the --verify option:
  <pre><code class="language-bash">$ bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --reassignment-json-file custom-reassignment.json --verify
Status of partition reassignment:
Reassignment of partition [foo1,0] is completed
Reassignment of partition [foo2,1] is completed</code></pre>

  <h4 class="anchor-heading"><a id="basic_ops_decommissioning_brokers" class="anchor-link"></a><a href="#basic_ops_decommissioning_brokers">Decommissioning brokers</a></h4>
  The partition reassignment tool does not have the ability to automatically generate a reassignment plan for decommissioning brokers yet. As such, the admin has to come up with a reassignment plan to move the replica for all partitions hosted on the broker to be decommissioned, to the rest of the brokers. This can be relatively tedious as the reassignment needs to ensure that all the replicas are not moved from the decommissioned broker to only one other broker. To make this process effortless, we plan to add tooling support for decommissioning brokers in the future.

  <h4 class="anchor-heading"><a id="basic_ops_increase_replication_factor" class="anchor-link"></a><a href="#basic_ops_increase_replication_factor">Increasing replication factor</a></h4>
  Increasing the replication factor of an existing partition is easy. Just specify the extra replicas in the custom reassignment json file and use it with the --execute option to increase the replication factor of the specified partitions.
  <p>
  For instance, the following example increases the replication factor of partition 0 of topic foo from 1 to 3. Before increasing the replication factor, the partition's only replica existed on broker 5. As part of increasing the replication factor, we will add more replicas on brokers 6 and 7.
  <p>
  The first step is to hand craft the custom reassignment plan in a json file:
  <pre><code class="language-bash">$ cat increase-replication-factor.json
{"version":1,
 "partitions":[{"topic":"foo","partition":0,"replicas":[5,6,7]}]}</code></pre>
  Then, use the json file with the --execute option to start the reassignment process:
  <pre><code class="language-bash">$ bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --reassignment-json-file increase-replication-factor.json --execute
Current partition replica assignment

{"version":1,
 "partitions":[{"topic":"foo","partition":0,"replicas":[5],"log_dirs":["any"]}]}

Save this to use as the --reassignment-json-file option during rollback
Successfully started partition reassignment for foo-0</code></pre>
  <p>
  The --verify option can be used with the tool to check the status of the partition reassignment. Note that the same increase-replication-factor.json (used with the --execute option) should be used with the --verify option:
  <pre><code class="language-bash">$ bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --reassignment-json-file increase-replication-factor.json --verify
Status of partition reassignment:
Reassignment of partition [foo,0] is completed</code></pre>
  You can also verify the increase in replication factor with the kafka-topics tool:
  <pre><code class="language-bash">$ bin/kafka-topics.sh --bootstrap-server localhost:9092 --topic foo --describe
Topic:foo	PartitionCount:1	ReplicationFactor:3	Configs:
  Topic: foo	Partition: 0	Leader: 5	Replicas: 5,6,7	Isr: 5,6,7</code></pre>

  <h4 class="anchor-heading"><a id="rep-throttle" class="anchor-link"></a><a href="#rep-throttle">Limiting Bandwidth Usage during Data Migration</a></h4>
  Kafka lets you apply a throttle to replication traffic, setting an upper bound on the bandwidth used to move replicas from machine to machine and from disk to disk. This is useful when rebalancing a cluster, adding or removing brokers or adding or removing disks, as it limits the impact these data-intensive operations will have on users.
  <p></p>
  There are two interfaces that can be used to engage a throttle. The simplest, and safest, is to apply a throttle when invoking the kafka-reassign-partitions.sh, but kafka-configs.sh can also be used to view and alter the throttle values directly.
  <p></p>
  So for example, if you were to execute a rebalance, with the below command, it would move partitions at no more than 50MB/s between brokers, and at no more than 100MB/s between disks on a broker.
  <pre class="language-bash">$ bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --execute --reassignment-json-file bigger-cluster.json --throttle 50000000 --replica-alter-log-dirs-throttle 100000000</pre>
  When you execute this script you will see the throttle engage:
  <pre><code class="language-text">The inter-broker throttle limit was set to 50000000 B/s
The replica-alter-dir throttle limit was set to 100000000 B/s
Successfully started partition reassignment for foo1-0</code></pre>
  <p>Should you wish to alter the throttle, during a rebalance, say to increase the inter-broker throughput so it completes quicker, you can do this by re-running the execute command with the --additional option passing the same reassignment-json-file:</p>
  <pre class="language-bash">$ bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --additional --execute --reassignment-json-file bigger-cluster.json --throttle 700000000
The inter-broker throttle limit was set to 700000000 B/s</pre>

  <p>Once the rebalance completes the administrator can check the status of the rebalance using the --verify option.
      If the rebalance has completed, the throttle will be removed via the --verify command. It is important that
      administrators remove the throttle in a timely manner once rebalancing completes by running the command with
      the --verify option. Failure to do so could cause regular replication traffic to be throttled. </p>
  <p>When the --verify option is executed, and the reassignment has completed, the script will confirm that the throttle was removed:</p>

  <pre><code class="language-bash">$ bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --verify --reassignment-json-file bigger-cluster.json
Status of partition reassignment:
Reassignment of partition [my-topic,1] is completed
Reassignment of partition [my-topic,0] is completed

Clearing broker-level throttles on brokers 1,2,3
Clearing topic-level throttles on topic my-topic</code></pre>

  <p>The administrator can also validate the assigned configs using the kafka-configs.sh. There are two sets of throttle
      configuration used to manage the throttling process. First set refers to the throttle value itself. This is configured, at a broker
      level, using the dynamic properties: </p>

  <pre><code class="language-text">leader.replication.throttled.rate
follower.replication.throttled.rate
replica.alter.log.dirs.io.max.bytes.per.second</code></pre>

  <p>Then there is the configuration pair of enumerated sets of throttled replicas: </p>

  <pre><code class="language-text">leader.replication.throttled.replicas
follower.replication.throttled.replicas</code></pre>

  <p>Which are configured per topic. </p>

  <p>All five config values are automatically assigned by kafka-reassign-partitions.sh (discussed below).</p>

  <p>To view the throttle limit configuration:</p>

  <pre><code class="language-bash">$ bin/kafka-configs.sh --describe --bootstrap-server localhost:9092 --entity-type brokers
Configs for brokers '2' are leader.replication.throttled.rate=700000000,follower.replication.throttled.rate=700000000,replica.alter.log.dirs.io.max.bytes.per.second=1000000000
Configs for brokers '1' are leader.replication.throttled.rate=700000000,follower.replication.throttled.rate=700000000,replica.alter.log.dirs.io.max.bytes.per.second=1000000000</code></pre>

  <p>This shows the throttle applied to both leader and follower side of the replication protocol (by default both sides
      are assigned the same throttled throughput value), as well as the disk throttle.</p>

  <p>To view the list of throttled replicas:</p>

  <pre><code class="language-bash">$ bin/kafka-configs.sh --describe --bootstrap-server localhost:9092 --entity-type topics
Configs for topic 'my-topic' are leader.replication.throttled.replicas=1:102,0:101,
    follower.replication.throttled.replicas=1:101,0:102</code></pre>

  <p>Here we see the leader throttle is applied to partition 1 on broker 102 and partition 0 on broker 101. Likewise the
      follower throttle is applied to partition 1 on
      broker 101 and partition 0 on broker 102. </p>

  <p>By default kafka-reassign-partitions.sh will apply the leader throttle to all replicas that exist before the
      rebalance, any one of which might be leader.
      It will apply the follower throttle to all move destinations. So if there is a partition with replicas on brokers
      101,102, being reassigned to 102,103, a leader throttle,
      for that partition, would be applied to 101,102 and a follower throttle would be applied to 103 only. </p>


  <p>If required, you can also use the --alter switch on kafka-configs.sh to alter the throttle configurations manually.
  </p>

  <h5>Safe usage of throttled replication</h5>

  <p>Some care should be taken when using throttled replication. In particular:</p>

  <p><i>(1) Throttle Removal:</i></p>
  The throttle should be removed in a timely manner once reassignment completes (by running kafka-reassign-partitions.sh --verify).

  <p><i>(2) Ensuring Progress:</i></p>
  <p>If the throttle is set too low, in comparison to the incoming write rate, it is possible for replication to not
      make progress. This occurs when:</p>
  <pre>max(BytesInPerSec) > throttle</pre>
  <p>
      Where BytesInPerSec is the metric that monitors the write throughput of producers into each broker. </p>
  <p>The administrator can monitor whether replication is making progress, during the rebalance, using the metric:</p>

  <pre>kafka.server:type=FetcherLagMetrics,name=ConsumerLag,clientId=([-.\w]+),topic=([-.\w]+),partition=([0-9]+)</pre>

  <p>The lag should constantly decrease during replication. If the metric does not decrease the administrator should
      increase the throttle throughput as described above. </p>


  <h4 class="anchor-heading"><a id="quotas" class="anchor-link"></a><a href="#quotas">Setting quotas</a></h4>
  Quotas overrides and defaults may be configured at (user, client-id), user or client-id levels as described <a href="#design_quotas">here</a>.
  By default, clients receive an unlimited quota.

  It is possible to set custom quotas for each (user, client-id), user or client-id group.
  <p>
  Configure custom quota for (user=user1, client-id=clientA):
  <pre><code class="language-bash">$ bin/kafka-configs.sh --bootstrap-server localhost:9092 --alter --add-config 'producer_byte_rate=1024,consumer_byte_rate=2048,request_percentage=200' --entity-type users --entity-name user1 --entity-type clients --entity-name clientA
Updated config for entity: user-principal 'user1', client-id 'clientA'.</code></pre>

  Configure custom quota for user=user1:
  <pre><code class="language-bash">$ bin/kafka-configs.sh --bootstrap-server localhost:9092 --alter --add-config 'producer_byte_rate=1024,consumer_byte_rate=2048,request_percentage=200' --entity-type users --entity-name user1
Updated config for entity: user-principal 'user1'.</code></pre>

  Configure custom quota for client-id=clientA:
  <pre><code class="language-bash">$ bin/kafka-configs.sh --bootstrap-server localhost:9092 --alter --add-config 'producer_byte_rate=1024,consumer_byte_rate=2048,request_percentage=200' --entity-type clients --entity-name clientA
Updated config for entity: client-id 'clientA'.</code></pre>

  It is possible to set default quotas for each (user, client-id), user or client-id group by specifying <i>--entity-default</i> option instead of <i>--entity-name</i>.
  <p>
  Configure default client-id quota for user=userA:
  <pre><code class="language-bash">$ bin/kafka-configs.sh --bootstrap-server localhost:9092 --alter --add-config 'producer_byte_rate=1024,consumer_byte_rate=2048,request_percentage=200' --entity-type users --entity-name user1 --entity-type clients --entity-default
Updated config for entity: user-principal 'user1', default client-id.</code></pre>

  Configure default quota for user:
  <pre><code class="language-bash">$ bin/kafka-configs.sh --bootstrap-server localhost:9092 --alter --add-config 'producer_byte_rate=1024,consumer_byte_rate=2048,request_percentage=200' --entity-type users --entity-default
Updated config for entity: default user-principal.</code></pre>

  Configure default quota for client-id:
  <pre><code class="language-bash">$ bin/kafka-configs.sh --bootstrap-server localhost:9092 --alter --add-config 'producer_byte_rate=1024,consumer_byte_rate=2048,request_percentage=200' --entity-type clients --entity-default
Updated config for entity: default client-id.</code></pre>

  Here's how to describe the quota for a given (user, client-id):
  <pre><code class="language-bash">$ bin/kafka-configs.sh --bootstrap-server localhost:9092 --describe --entity-type users --entity-name user1 --entity-type clients --entity-name clientA
Configs for user-principal 'user1', client-id 'clientA' are producer_byte_rate=1024,consumer_byte_rate=2048,request_percentage=200</code></pre>
  Describe quota for a given user:
  <pre><code class="language-bash">$ bin/kafka-configs.sh --bootstrap-server localhost:9092 --describe --entity-type users --entity-name user1
Configs for user-principal 'user1' are producer_byte_rate=1024,consumer_byte_rate=2048,request_percentage=200</code></pre>
  Describe quota for a given client-id:
  <pre><code class="language-bash">$ bin/kafka-configs.sh --bootstrap-server localhost:9092 --describe --entity-type clients --entity-name clientA
Configs for client-id 'clientA' are producer_byte_rate=1024,consumer_byte_rate=2048,request_percentage=200</code></pre>
  If entity name is not specified, all entities of the specified type are described. For example, describe all users:
  <pre><code class="language-bash">$ bin/kafka-configs.sh --bootstrap-server localhost:9092 --describe --entity-type users
Configs for user-principal 'user1' are producer_byte_rate=1024,consumer_byte_rate=2048,request_percentage=200
Configs for default user-principal are producer_byte_rate=1024,consumer_byte_rate=2048,request_percentage=200</code></pre>
  Similarly for (user, client):
  <pre><code class="language-bash">$ bin/kafka-configs.sh --bootstrap-server localhost:9092 --describe --entity-type users --entity-type clients
Configs for user-principal 'user1', default client-id are producer_byte_rate=1024,consumer_byte_rate=2048,request_percentage=200
Configs for user-principal 'user1', client-id 'clientA' are producer_byte_rate=1024,consumer_byte_rate=2048,request_percentage=200</code></pre>

  <h3 class="anchor-heading"><a id="datacenters" class="anchor-link"></a><a href="#datacenters">6.2 Datacenters</a></h3>

  Some deployments will need to manage a data pipeline that spans multiple datacenters. Our recommended approach to this is to deploy a local Kafka cluster in each datacenter, with application instances in each datacenter interacting only with their local cluster and mirroring data between clusters (see the documentation on <a href="#georeplication">Geo-Replication</a> for how to do this).
  <p>
  This deployment pattern allows datacenters to act as independent entities and allows us to manage and tune inter-datacenter replication centrally. This allows each facility to stand alone and operate even if the inter-datacenter links are unavailable: when this occurs the mirroring falls behind until the link is restored at which time it catches up.
  <p>
  For applications that need a global view of all data you can use mirroring to provide clusters which have aggregate data mirrored from the local clusters in <i>all</i> datacenters. These aggregate clusters are used for reads by applications that require the full data set.
  <p>
  This is not the only possible deployment pattern. It is possible to read from or write to a remote Kafka cluster over the WAN, though obviously this will add whatever latency is required to get the cluster.
  <p>
  Kafka naturally batches data in both the producer and consumer so it can achieve high-throughput even over a high-latency connection. To allow this though it may be necessary to increase the TCP socket buffer sizes for the producer, consumer, and broker using the <code>socket.send.buffer.bytes</code> and <code>socket.receive.buffer.bytes</code> configurations. The appropriate way to set this is documented <a href="https://en.wikipedia.org/wiki/Bandwidth-delay_product">here</a>.
  <p>
  It is generally <i>not</i> advisable to run a <i>single</i> Kafka cluster that spans multiple datacenters over a high-latency link. This will incur very high replication latency both for Kafka writes and ZooKeeper writes, and neither Kafka nor ZooKeeper will remain available in all locations if the network between locations is unavailable.

  <h3 class="anchor-heading"><a id="georeplication" class="anchor-link"></a><a href="#georeplication">6.3 Geo-Replication (Cross-Cluster Data Mirroring)</a></h3>

  <h4 class="anchor-heading"><a id="georeplication-overview" class="anchor-link"></a><a href="#georeplication-overview">Geo-Replication Overview</a></h4>

  <p>
    Kafka administrators can define data flows that cross the boundaries of individual Kafka clusters, data centers, or geo-regions. Such event streaming setups are often needed for organizational, technical, or legal requirements. Common scenarios include:
  </p>

  <ul>
    <li>Geo-replication</li>
    <li>Disaster recovery</li>
    <li>Feeding edge clusters into a central, aggregate cluster</li>
    <li>Physical isolation of clusters (such as production vs. testing)</li>
    <li>Cloud migration or hybrid cloud deployments</li>
    <li>Legal and compliance requirements</li>
  </ul>

  <p>
    Administrators can set up such inter-cluster data flows with Kafka's MirrorMaker (version 2), a tool to replicate data between different Kafka environments in a streaming manner. MirrorMaker is built on top of the Kafka Connect framework and supports features such as:
  </p>

  <ul>
    <li>Replicates topics (data plus configurations)</li>
    <li>Replicates consumer groups including offsets to migrate applications between clusters</li>
    <li>Replicates ACLs</li>
    <li>Preserves partitioning</li>
    <li>Automatically detects new topics and partitions</li>
    <li>Provides a wide range of metrics, such as end-to-end replication latency across multiple data centers/clusters</li>
    <li>Fault-tolerant and horizontally scalable operations</li>
  </ul>

  <p>
  <em>Note: Geo-replication with MirrorMaker replicates data across Kafka clusters. This inter-cluster replication is different from Kafka's <a href="#replication">intra-cluster replication</a>, which replicates data within the same Kafka cluster.</em>
  </p>

  <h4 class="anchor-heading"><a id="georeplication-flows" class="anchor-link"></a><a href="#georeplication-flows">What Are Replication Flows</a></h4>

  <p>
    With MirrorMaker, Kafka administrators can replicate topics, topic configurations, consumer groups and their offsets, and ACLs from one or more source Kafka clusters to one or more target Kafka clusters, i.e., across cluster environments. In a nutshell, MirrorMaker uses Connectors to consume from source clusters and produce to target clusters.
  </p>

  <p>
    These directional flows from source to target clusters are called replication flows. They are defined with the format <code>{source_cluster}->{target_cluster}</code> in the MirrorMaker configuration file as described later. Administrators can create complex replication topologies based on these flows.
  </p>

  <p>
    Here are some example patterns:
  </p>

  <ul>
    <li>Active/Active high availability deployments: <code>A->B, B->A</code></li>
    <li>Active/Passive or Active/Standby high availability deployments: <code>A->B</code></li>
    <li>Aggregation (e.g., from many clusters to one): <code>A->K, B->K, C->K</code></li>
    <li>Fan-out (e.g., from one to many clusters): <code>K->A, K->B, K->C</code></li>
    <li>Forwarding: <code>A->B, B->C, C->D</code></li>
  </ul>

  <p>
    By default, a flow replicates all topics and consumer groups (except excluded ones). However, each replication flow can be configured independently. For instance, you can define that only specific topics or consumer groups are replicated from the source cluster to the target cluster.
  </p>

  <p>
    Here is a first example on how to configure data replication from a <code>primary</code> cluster to a <code>secondary</code> cluster (an active/passive setup):
  </p>

<pre><code class="language-text"># Basic settings
clusters = primary, secondary
primary.bootstrap.servers = broker3-primary:9092
secondary.bootstrap.servers = broker5-secondary:9092

# Define replication flows
primary->secondary.enabled = true
primary->secondary.topics = foobar-topic, quux-.*</code></pre>


  <h4 class="anchor-heading"><a id="georeplication-mirrormaker" class="anchor-link"></a><a href="#georeplication-mirrormaker">Configuring Geo-Replication</a></h4>

  <p>
    The following sections describe how to configure and run a dedicated MirrorMaker cluster. If you want to run MirrorMaker within an existing Kafka Connect cluster or other supported deployment setups, please refer to <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-382%3A+MirrorMaker+2.0">KIP-382: MirrorMaker 2.0</a> and be aware that the names of configuration settings may vary between deployment modes.
  </p>

  <p>
    Beyond what's covered in the following sections, further examples and information on configuration settings are available at:
  </p>

  <ul>
	  <li><a href="https://github.com/apache/kafka/blob/trunk/connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorMakerConfig.java">MirrorMakerConfig</a>, <a href="https://github.com/apache/kafka/blob/trunk/connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorConnectorConfig.java">MirrorConnectorConfig</a></li>
	  <li><a href="https://github.com/apache/kafka/blob/trunk/connect/mirror/src/main/java/org/apache/kafka/connect/mirror/DefaultTopicFilter.java">DefaultTopicFilter</a> for topics, <a href="https://github.com/apache/kafka/blob/trunk/connect/mirror/src/main/java/org/apache/kafka/connect/mirror/DefaultGroupFilter.java">DefaultGroupFilter</a> for consumer groups</li>
	  <li>Example configuration settings in <a href="https://github.com/apache/kafka/blob/trunk/config/connect-mirror-maker.properties">connect-mirror-maker.properties</a>, <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-382%3A+MirrorMaker+2.0">KIP-382: MirrorMaker 2.0</a></li>
  </ul>

  <h5 class="anchor-heading"><a id="georeplication-config-syntax" class="anchor-link"></a><a href="#georeplication-config-syntax">Configuration File Syntax</a></h5>

  <p>
    The MirrorMaker configuration file is typically named <code>connect-mirror-maker.properties</code>. You can configure a variety of components in this file:
  </p>

  <ul>
    <li>MirrorMaker settings: global settings including cluster definitions (aliases), plus custom settings per replication flow</li>
    <li>Kafka Connect and connector settings</li>
    <li>Kafka producer, consumer, and admin client settings</li>
  </ul>

  <p>
    Example: Define MirrorMaker settings (explained in more detail later).
  </p>

<pre><code class="language-text"># Global settings
clusters = us-west, us-east   # defines cluster aliases
us-west.bootstrap.servers = broker3-west:9092
us-east.bootstrap.servers = broker5-east:9092

topics = .*   # all topics to be replicated by default

# Specific replication flow settings (here: flow from us-west to us-east)
us-west->us-east.enabled = true
us-west->us.east.topics = foo.*, bar.*  # override the default above</code></pre>

  <p>
    MirrorMaker is based on the Kafka Connect framework. Any Kafka Connect, source connector, and sink connector settings as described in the <a href="#connectconfigs">documentation chapter on Kafka Connect</a> can be used directly in the MirrorMaker configuration, without having to change or prefix the name of the configuration setting.
  </p>

  <p>
    Example: Define custom Kafka Connect settings to be used by MirrorMaker.
  </p>

<pre><code class="language-text"># Setting Kafka Connect defaults for MirrorMaker
tasks.max = 5</code></pre>

  <p>
  Most of the default Kafka Connect settings work well for MirrorMaker out-of-the-box, with the exception of <code>tasks.max</code>. In order to evenly distribute the workload across more than one MirrorMaker process, it is recommended to set <code>tasks.max</code> to at least <code>2</code> (preferably higher) depending on the available hardware resources and the total number of topic-partitions to be replicated.
  </p>

  <p>
  You can further customize MirrorMaker's Kafka Connect settings <em>per source or target cluster</em> (more precisely, you can specify Kafka Connect worker-level configuration settings "per connector"). Use the format of <code>{cluster}.{config_name}</code> in the MirrorMaker configuration file.
  </p>

  <p>
    Example: Define custom connector settings for the <code>us-west</code> cluster.
  </p>

<pre><code class="language-text"># us-west custom settings
us-west.offset.storage.topic = my-mirrormaker-offsets</code></pre>

  <p>
    MirrorMaker internally uses the Kafka producer, consumer, and admin clients. Custom settings for these clients are often needed. To override the defaults, use the following format in the MirrorMaker configuration file:
  </p>

  <ul>
    <li><code>{source}.consumer.{consumer_config_name}</code></li>
    <li><code>{target}.producer.{producer_config_name}</code></li>
    <li><code>{source_or_target}.admin.{admin_config_name}</code></li>
  </ul>

  <p>
    Example: Define custom producer, consumer, admin client settings.
  </p>

<pre><code class="language-text"># us-west cluster (from which to consume)
us-west.consumer.isolation.level = read_committed
us-west.admin.bootstrap.servers = broker57-primary:9092

# us-east cluster (to which to produce)
us-east.producer.compression.type = gzip
us-east.producer.buffer.memory = 32768
us-east.admin.bootstrap.servers = broker8-secondary:9092</code></pre>

  <h5 class="anchor-heading"><a id="georeplication-exactly_once" class="anchor-link"></a><a href="#georeplication-exactly_once">Exactly once</a></h5>

  <p>
    Exactly-once semantics are supported for dedicated MirrorMaker clusters as of version 3.5.0.</p>
  
  <p>
    For new MirrorMaker clusters, set the <code>exactly.once.source.support</code> property to enabled for all targeted Kafka clusters that should be written to with exactly-once semantics. For example, to enable exactly-once for writes to cluster <code>us-east</code>, the following configuration can be used:
  </p>

<pre><code class="language-text">us-east.exactly.once.source.support = enabled</code></pre>
  
  <p>
    For existing MirrorMaker clusters, a two-step upgrade is necessary. Instead of immediately setting the <code>exactly.once.source.support</code> property to enabled, first set it to <code>preparing</code> on all nodes in the cluster. Once this is complete, it can be set to <code>enabled</code> on all nodes in the cluster, in a second round of restarts.
  </p>
  
  <p>
    In either case, it is also necessary to enable intra-cluster communication between the MirrorMaker nodes, as described in  <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-710%3A+Full+support+for+distributed+mode+in+dedicated+MirrorMaker+2.0+clusters">KIP-710</a>. To do this, the <code>dedicated.mode.enable.internal.rest</code> property must be set to <code>true</code>. In addition, many of the REST-related  <a href="https://kafka.apache.org/documentation/#connectconfigs">configuration properties available for Kafka Connect</a> can be specified the MirrorMaker config. For example, to enable intra-cluster communication in MirrorMaker cluster with each node listening on port 8080 of their local machine, the following should be added to the MirrorMaker config file:
  </p>

<pre><code class="language-text">dedicated.mode.enable.internal.rest = true
listeners = http://localhost:8080</code></pre>
  
  <p><b>
    Note that, if intra-cluster communication is enabled in production environments, it is highly recommended to secure the REST servers brought up by each MirrorMaker node. See the <a href="https://kafka.apache.org/documentation/#connectconfigs">configuration properties for Kafka Connect</a> for information on how this can be accomplished.
  </b></p>
  
  <p>
    It is also recommended to filter records from aborted transactions out from replicated data when running MirrorMaker. To do this, ensure that the consumer used to read from source clusters is configured with <code>isolation.level</code> set to <code>read_committed</code>. If replicating data from cluster <code>us-west</code>, this can be done for all replication flows that read from that cluster by adding the following to the MirrorMaker config file:
  </p>

<pre><code class="language-text">us-west.consumer.isolation.level = read_committed</code></pre>

  <p>
    As a final note, under the hood, MirrorMaker uses Kafka Connect source connectors to replicate data. For more information on exactly-once support for these kinds of connectors, see the <a href="https://kafka.apache.org/documentation/#connect_exactlyoncesource">relevant docs page</a>.
  </p>

  <h5 class="anchor-heading"><a id="georeplication-flow-create" class="anchor-link"></a><a href="#georeplication-flow-create">Creating and Enabling Replication Flows</a></h5>

  <p>
    To define a replication flow, you must first define the respective source and target Kafka clusters in the MirrorMaker configuration file.
  </p>

  <ul>
    <li><code>clusters</code> (required): comma-separated list of Kafka cluster "aliases"</li>
    <li><code>{clusterAlias}.bootstrap.servers</code> (required): connection information for the specific cluster; comma-separated list of "bootstrap" Kafka brokers
  </ul>

  <p>
    Example: Define two cluster aliases <code>primary</code> and <code>secondary</code>, including their connection information.
  </p>

<pre><code class="language-text">clusters = primary, secondary
primary.bootstrap.servers = broker10-primary:9092,broker-11-primary:9092
secondary.bootstrap.servers = broker5-secondary:9092,broker6-secondary:9092</code></pre>

  <p>
    Secondly, you must explicitly enable individual replication flows with <code>{source}->{target}.enabled = true</code> as needed. Remember that flows are directional: if you need two-way (bidirectional) replication, you must enable flows in both directions.
  </p>

<pre><code class="language-text"># Enable replication from primary to secondary
primary->secondary.enabled = true</code></pre>

  <p>
    By default, a replication flow will replicate all but a few special topics and consumer groups from the source cluster to the target cluster, and automatically detect any newly created topics and groups. The names of replicated topics in the target cluster will be prefixed with the name of the source cluster (see section further below). For example, the topic <code>foo</code> in the source cluster <code>us-west</code> would be replicated to a topic named <code>us-west.foo</code> in the target cluster <code>us-east</code>.
  </p>

  <p>
    The subsequent sections explain how to customize this basic setup according to your needs.
  </p>

  <h5 class="anchor-heading"><a id="georeplication-flow-configure" class="anchor-link"></a><a href="#georeplication-flow-configure">Configuring Replication Flows</a></h5>

  <p>
The configuration of a replication flow is a combination of top-level default settings (e.g., <code>topics</code>), on top of which flow-specific settings, if any, are applied (e.g., <code>us-west->us-east.topics</code>). To change the top-level defaults, add the respective top-level setting to the MirrorMaker configuration file. To override the defaults for a specific replication flow only, use the syntax format <code>{source}->{target}.{config.name}</code>.
  </p>

  <p>
    The most important settings are:
  </p>

  <ul>
    <li><code>topics</code>: list of topics or a regular expression that defines which topics in the source cluster to replicate (default: <code>topics = .*</code>)
    <li><code>topics.exclude</code>: list of topics or a regular expression to subsequently exclude topics that were matched by the <code>topics</code> setting (default: <code>topics.exclude = .*[\-\.]internal, .*\.replica, __.*</code>)
    <li><code>groups</code>: list of topics or regular expression that defines which consumer groups in the source cluster to replicate (default: <code>groups = .*</code>)
    <li><code>groups.exclude</code>: list of topics or a regular expression to subsequently exclude consumer groups that were matched by the <code>groups</code> setting (default: <code>groups.exclude = console-consumer-.*, connect-.*, __.*</code>)
    <li><code>{source}->{target}.enable</code>: set to <code>true</code> to enable the replication flow (default: <code>false</code>)
  </ul>

  <p>
    Example:
  </p>

<pre><code class="language-text"># Custom top-level defaults that apply to all replication flows
topics = .*
groups = consumer-group1, consumer-group2

# Don't forget to enable a flow!
us-west->us-east.enabled = true

# Custom settings for specific replication flows
us-west->us-east.topics = foo.*
us-west->us-east.groups = bar.*
us-west->us-east.emit.heartbeats = false</code></pre>

  <p>
    Additional configuration settings are supported which can be left with their default values in most cases. See <a href="/documentation/#mirrormakerconfigs">MirrorMaker Configs</a>.
  </p>

  <h5 class="anchor-heading"><a id="georeplication-flow-secure" class="anchor-link"></a><a href="#georeplication-flow-secure">Securing Replication Flows</a></h5>

  <p>
    MirrorMaker supports the same <a href="#connectconfigs">security settings as Kafka Connect</a>, so please refer to the linked section for further information.
  </p>

  <p>
    Example: Encrypt communication between MirrorMaker and the <code>us-east</code> cluster.
  </p>

<pre><code class="language-text">us-east.security.protocol=SSL
us-east.ssl.truststore.location=/path/to/truststore.jks
us-east.ssl.truststore.password=my-secret-password
us-east.ssl.keystore.location=/path/to/keystore.jks
us-east.ssl.keystore.password=my-secret-password
us-east.ssl.key.password=my-secret-password</code></pre>

  <h5 class="anchor-heading"><a id="georeplication-topic-naming" class="anchor-link"></a><a href="#georeplication-topic-naming">Custom Naming of Replicated Topics in Target Clusters</a></h5>

  <p>
    Replicated topics in a target cluster—sometimes called <em>remote</em> topics—are renamed according to a replication policy. MirrorMaker uses this policy to ensure that events (aka records, messages) from different clusters are not written to the same topic-partition. By default as per <a href="https://github.com/apache/kafka/blob/trunk/connect/mirror-client/src/main/java/org/apache/kafka/connect/mirror/DefaultReplicationPolicy.java">DefaultReplicationPolicy</a>, the names of replicated topics in the target clusters have the format <code>{source}.{source_topic_name}</code>:
  </p>

<pre><code class="language-text">us-west         us-east
=========       =================
                bar-topic
foo-topic  -->  us-west.foo-topic</code></pre>

  <p>
    You can customize the separator (default: <code>.</code>) with the <code>replication.policy.separator</code> setting:
  </p>

<pre><code class="language-text"># Defining a custom separator
us-west->us-east.replication.policy.separator = _</code></pre>

  <p>
    If you need further control over how replicated topics are named, you can implement a custom <code>ReplicationPolicy</code> and override <code>replication.policy.class</code> (default is <code>DefaultReplicationPolicy</code>) in the MirrorMaker configuration.
  </p>

  <h5 class="anchor-heading"><a id="georeplication-config-conflicts" class="anchor-link"></a><a href="#georeplication-config-conflicts">Preventing Configuration Conflicts</a></h5>

  <p>
    MirrorMaker processes share configuration via their target Kafka clusters. This behavior may cause conflicts when configurations differ among MirrorMaker processes that operate against the same target cluster.
  </p>

  <p>
    For example, the following two MirrorMaker processes would be racy:
  </p>

<pre><code class="language-text"># Configuration of process 1
A->B.enabled = true
A->B.topics = foo

# Configuration of process 2
A->B.enabled = true
A->B.topics = bar</code></pre>

  <p>
    In this case, the two processes will share configuration via cluster <code>B</code>, which causes a conflict. Depending on which of the two processes is the elected "leader", the result will be that either the topic <code>foo</code> or the topic <code>bar</code> is replicated, but not both.
  </p>

  <p>
    It is therefore important to keep the MirrorMaker configuration consistent across replication flows to the same target cluster. This can be achieved, for example, through automation tooling or by using a single, shared MirrorMaker configuration file for your entire organization.
  </p>

  <h5 class="anchor-heading"><a id="georeplication-best-practice" class="anchor-link"></a><a href="#georeplication-best-practice">Best Practice: Consume from Remote, Produce to Local</a></h5>

  <p>
To minimize latency ("producer lag"), it is recommended to locate MirrorMaker processes as close as possible to their target clusters, i.e., the clusters that it produces data to. That's because Kafka producers typically struggle more with unreliable or high-latency network connections than Kafka consumers.
  </p>

<pre><code class="language-text">First DC          Second DC
==========        =========================
primary --------- MirrorMaker --> secondary
(remote)                           (local)</code></pre>

  <p>
To run such a "consume from remote, produce to local" setup, run the MirrorMaker processes close to and preferably in the same location as the target clusters, and explicitly set these "local" clusters in the <code>--clusters</code> command line parameter (blank-separated list of cluster aliases):
  </p>

<pre><code class="language-bash"># Run in secondary's data center, reading from the remote `primary` cluster
$ bin/connect-mirror-maker.sh connect-mirror-maker.properties --clusters secondary</code></pre>

The <code>--clusters secondary</code> tells the MirrorMaker process that the given cluster(s) are nearby, and prevents it from replicating data or sending configuration to clusters at other, remote locations.

  <h5 class="anchor-heading"><a id="georeplication-example-active-passive" class="anchor-link"></a><a href="#georeplication-example-active-passive">Example: Active/Passive High Availability Deployment</a></h5>

  <p>
The following example shows the basic settings to replicate topics from a primary to a secondary Kafka environment, but not from the secondary back to the primary. Please be aware that most production setups will need further configuration, such as security settings.
  </p>

<pre><code class="language-text"># Unidirectional flow (one-way) from primary to secondary cluster
primary.bootstrap.servers = broker1-primary:9092
secondary.bootstrap.servers = broker2-secondary:9092

primary->secondary.enabled = true
secondary->primary.enabled = false

primary->secondary.topics = foo.*  # only replicate some topics</code></pre>

  <h5 class="anchor-heading"><a id="georeplication-example-active-active" class="anchor-link"></a><a href="#georeplication-example-active-active">Example: Active/Active High Availability Deployment</a></h5>

  <p>
    The following example shows the basic settings to replicate topics between two clusters in both ways. Please be aware that most production setups will need further configuration, such as security settings.
  </p>

<pre><code class="language-text"># Bidirectional flow (two-way) between us-west and us-east clusters
clusters = us-west, us-east
us-west.bootstrap.servers = broker1-west:9092,broker2-west:9092
Us-east.bootstrap.servers = broker3-east:9092,broker4-east:9092

us-west->us-east.enabled = true
us-east->us-west.enabled = true</code></pre>

  <p>
    <em>Note on preventing replication "loops" (where topics will be originally replicated from A to B, then the replicated topics will be replicated yet again from B to A, and so forth)</em>: As long as you define the above flows in the same MirrorMaker configuration file, you do not need to explicitly add <code>topics.exclude</code> settings to prevent replication loops between the two clusters.
  </p>

  <h5 class="anchor-heading"><a id="georeplication-example-multi-cluster" class="anchor-link"></a><a href="#georeplication-example-multi-cluster">Example: Multi-Cluster Geo-Replication</a></h5>

  <p>
    Let's put all the information from the previous sections together in a larger example. Imagine there are three data centers (west, east, north), with two Kafka clusters in each data center (e.g., <code>west-1</code>, <code>west-2</code>). The example in this section shows how to configure MirrorMaker (1) for Active/Active replication within each data center, as well as (2) for Cross Data Center Replication (XDCR).
  </p>

  <p>
    First, define the source and target clusters along with their replication flows in the configuration:
  </p>

<pre><code class="language-text"># Basic settings
clusters: west-1, west-2, east-1, east-2, north-1, north-2
west-1.bootstrap.servers = ...
west-2.bootstrap.servers = ...
east-1.bootstrap.servers = ...
east-2.bootstrap.servers = ...
north-1.bootstrap.servers = ...
north-2.bootstrap.servers = ...

# Replication flows for Active/Active in West DC
west-1->west-2.enabled = true
west-2->west-1.enabled = true

# Replication flows for Active/Active in East DC
east-1->east-2.enabled = true
east-2->east-1.enabled = true

# Replication flows for Active/Active in North DC
north-1->north-2.enabled = true
north-2->north-1.enabled = true

# Replication flows for XDCR via west-1, east-1, north-1
west-1->east-1.enabled  = true
west-1->north-1.enabled = true
east-1->west-1.enabled  = true
east-1->north-1.enabled = true
north-1->west-1.enabled = true
north-1->east-1.enabled = true</code></pre>

  <p>
    Then, in each data center, launch one or more MirrorMaker as follows:
  </p>

<pre><code class="language-bash"># In West DC:
$ bin/connect-mirror-maker.sh connect-mirror-maker.properties --clusters west-1 west-2

# In East DC:
$ bin/connect-mirror-maker.sh connect-mirror-maker.properties --clusters east-1 east-2

# In North DC:
$ bin/connect-mirror-maker.sh connect-mirror-maker.properties --clusters north-1 north-2</code></pre>

  <p>
    With this configuration, records produced to any cluster will be replicated within the data center, as well as across to other data centers. By providing the <code>--clusters</code> parameter, we ensure that each MirrorMaker process produces data to nearby clusters only.
  </p>

  <p>
    <em>Note:</em> The <code>--clusters</code> parameter is, technically, not required here. MirrorMaker will work fine without it. However, throughput may suffer from "producer lag" between data centers, and you may incur unnecessary data transfer costs.
  </p>

<h4 class="anchor-heading"><a id="georeplication-starting" class="anchor-link"></a><a href="#georeplication-starting">Starting Geo-Replication</a></h4>

  <p>
    You can run as few or as many MirrorMaker processes (think: nodes, servers) as needed. Because MirrorMaker is based on Kafka Connect, MirrorMaker processes that are configured to replicate the same Kafka clusters run in a distributed setup: They will find each other, share configuration (see section below), load balance their work, and so on. If, for example, you want to increase the throughput of replication flows, one option is to run additional MirrorMaker processes in parallel.
  </p>

  <p>
    To start a MirrorMaker process, run the command:
  </p>

<pre><code class="language-bash">$ bin/connect-mirror-maker.sh connect-mirror-maker.properties</code></pre>

  <p>
    After startup, it may take a few minutes until a MirrorMaker process first begins to replicate data.
  </p>

  <p>
    Optionally, as described previously, you can set the parameter <code>--clusters</code> to ensure that the MirrorMaker process produces data to nearby clusters only.
  </p>

<pre><code class="language-bash"># Note: The cluster alias us-west must be defined in the configuration file
$ bin/connect-mirror-maker.sh connect-mirror-maker.properties \
    --clusters us-west
</code></pre>

  <p>
    <em>Note when testing replication of consumer groups:</em> By default, MirrorMaker does not replicate consumer groups created by the  <code>kafka-console-consumer.sh</code> tool, which you might use to test your MirrorMaker setup on the command line. If you do want to replicate these consumer groups as well, set the <code>groups.exclude</code> configuration accordingly (default: <code>groups.exclude = console-consumer-.*, connect-.*, __.*</code>). Remember to update the configuration again once you completed your testing.
  </p>

<h4 class="anchor-heading"><a id="georeplication-stopping" class="anchor-link"></a><a href="#georeplication-stopping">Stopping Geo-Replication</a></h4>

  <p>
    You can stop a running MirrorMaker process by sending a SIGTERM signal with the command:
  </p>

<pre><code class="language-bash">$ kill &lt;MirrorMaker pid&gt;</code></pre>

<h4 class="anchor-heading"><a id="georeplication-apply-config-changes" class="anchor-link"></a><a href="#georeplication-apply-config-changes">Applying Configuration Changes</a></h4>

  <p>
    To make configuration changes take effect, the MirrorMaker process(es) must be restarted.
  </p>

<h4 class="anchor-heading"><a id="georeplication-monitoring" class="anchor-link"></a><a href="#georeplication-monitoring">Monitoring Geo-Replication</a></h4>

  <p>
    It is recommended to monitor MirrorMaker processes to ensure all defined replication flows are up and running correctly. MirrorMaker is built on the Connect framework and inherits all of Connect's metrics, such <code>source-record-poll-rate</code>. In addition, MirrorMaker produces its own metrics under the <code>kafka.connect.mirror</code> metric group. Metrics are tagged with the following properties:
  </p>

  <ul>
    <li><code>source</code>: alias of source cluster (e.g., <code>primary</code>)</li>
    <li><code>target</code>: alias of target cluster (e.g., <code>secondary</code>)</li>
    <li><code>topic</code>:  replicated topic on target cluster</li>
    <li><code>partition</code>: partition being replicated</li>
  </ul>

  <p>
    Metrics are tracked for each replicated topic. The source cluster can be inferred from the topic name. For example, replicating <code>topic1</code> from <code>primary->secondary</code> will yield metrics like:
  </p>

  <ul>
    <li><code>target=secondary</code>
    <li><code>topic=primary.topic1</code>
    <li><code>partition=1</code>
  </ul>

  <p>
    The following metrics are emitted:
  </p>

<pre><code class="language-text"># MBean: kafka.connect.mirror:type=MirrorSourceConnector,target=([-.w]+),topic=([-.w]+),partition=([0-9]+)
record-count            # number of records replicated source -> target
record-age-ms           # age of records when they are replicated
record-age-ms-min
record-age-ms-max
record-age-ms-avg
replication-latency-ms  # time it takes records to propagate source->target
replication-latency-ms-min
replication-latency-ms-max
replication-latency-ms-avg
byte-rate               # average number of bytes/sec in replicated records

# MBean: kafka.connect.mirror:type=MirrorCheckpointConnector,source=([-.w]+),target=([-.w]+)

checkpoint-latency-ms   # time it takes to replicate consumer offsets
checkpoint-latency-ms-min
checkpoint-latency-ms-max
checkpoint-latency-ms-avg</code></pre>

  <p>
    These metrics do not differentiate between created-at and log-append timestamps.
  </p>


  <h3 class="anchor-heading"><a id="multitenancy" class="anchor-link"></a><a href="#multitenancy">6.4 Multi-Tenancy</a></h3>

  <h4 class="anchor-heading"><a id="multitenancy-overview" class="anchor-link"></a><a href="#multitenancy-overview">Multi-Tenancy Overview</a></h4>

  <p>
    As a highly scalable event streaming platform, Kafka is used by many users as their central nervous system, connecting in real-time a wide range of different systems and applications from various teams and lines of businesses. Such multi-tenant cluster environments command proper control and management to ensure the peaceful coexistence of these different needs. This section highlights features and best practices to set up such shared environments, which should help you operate clusters that meet SLAs/OLAs and that minimize potential collateral damage caused by "noisy neighbors".
  </p>

  <p>
    Multi-tenancy is a many-sided subject, including but not limited to:
  </p>

  <ul>
    <li>Creating user spaces for tenants (sometimes called namespaces)</li>
    <li>Configuring topics with data retention policies and more</li>
    <li>Securing topics and clusters with encryption, authentication, and authorization</li>
    <li>Isolating tenants with quotas and rate limits</li>
    <li>Monitoring and metering</li>
    <li>Inter-cluster data sharing (cf. geo-replication)</li>
  </ul>

  <h4 class="anchor-heading"><a id="multitenancy-topic-naming" class="anchor-link"></a><a href="#multitenancy-topic-naming">Creating User Spaces (Namespaces) For Tenants With Topic Naming</a></h4>

  <p>
    Kafka administrators operating a multi-tenant cluster typically need to define user spaces for each tenant. For the purpose of this section, "user spaces" are a collection of topics, which are grouped together under the management of a single entity or user.
  </p>

  <p>
    In Kafka, the main unit of data is the topic. Users can create and name each topic. They can also delete them, but it is not possible to rename a topic directly. Instead, to rename a topic, the user must create a new topic, move the messages from the original topic to the new, and then delete the original. With this in mind, it is recommended to define logical spaces, based on an hierarchical topic naming structure. This setup can then be combined with security features, such as prefixed ACLs, to isolate different spaces and tenants, while also minimizing the administrative overhead for securing the data in the cluster.
  </p>

  <p>
    These logical user spaces can be grouped in different ways, and the concrete choice depends on how your organization prefers to use your Kafka clusters. The most common groupings are as follows.
  </p>

  <p>
    <em>By team or organizational unit:</em> Here, the team is the main aggregator. In an organization where teams are the main user of the Kafka infrastructure, this might be the best grouping.
  </p>

  <p>
    Example topic naming structure:
  </p>

  <ul>
    <li><code>&lt;organization&gt;.&lt;team&gt;.&lt;dataset&gt;.&lt;event-name&gt;</code><br />(e.g., "acme.infosec.telemetry.logins")</li>
  </ul>

  <p>
    <em>By project or product:</em> Here, a team manages more than one project. Their credentials will be different for each project, so all the controls and settings will always be project related.
  </p>

  <p>
    Example topic naming structure:
  </p>

  <ul>
    <li><code>&lt;project&gt;.&lt;product&gt;.&lt;event-name&gt;</code><br />(e.g., "mobility.payments.suspicious")</li>
  </ul>

  <p>
    Certain information should normally not be put in a topic name, such as information that is likely to change over time (e.g., the name of the intended consumer) or that is a technical detail or metadata that is available elsewhere (e.g., the topic's partition count and other configuration settings).
  </p>

  <p>
  To enforce a topic naming structure, several options are available:
  </p>

  <ul>
    <li>Use <a href="#security_authz">prefix ACLs</a> (cf. <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-290%3A+Support+for+Prefixed+ACLs">KIP-290</a>) to enforce a common prefix for topic names. For example, team A may only be permitted to create topics whose names start with <code>payments.teamA.</code>.</li>
    <li>Define a custom <code>CreateTopicPolicy</code> (cf. <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-108%3A+Create+Topic+Policy">KIP-108</a> and the setting <a href="#brokerconfigs_create.topic.policy.class.name">create.topic.policy.class.name</a>) to enforce strict naming patterns. These policies provide the most flexibility and can cover complex patterns and rules to match an organization's needs.</li>
    <li>Disable topic creation for normal users by denying it with an ACL, and then rely on an external process to create topics on behalf of users (e.g., scripting or your favorite automation toolkit).</li>
    <li>It may also be useful to disable the Kafka feature to auto-create topics on demand by setting <code>auto.create.topics.enable=false</code> in the broker configuration. Note that you should not rely solely on this option.</li>
  </ul>


  <h4 class="anchor-heading"><a id="multitenancy-topic-configs" class="anchor-link"></a><a href="#multitenancy-topic-configs">Configuring Topics: Data Retention And More</a></h4>

  <p>
    Kafka's configuration is very flexible due to its fine granularity, and it supports a plethora of <a href="#topicconfigs">per-topic configuration settings</a> to help administrators set up multi-tenant clusters. For example, administrators often need to define data retention policies to control how much and/or for how long data will be stored in a topic, with settings such as <a href="#retention.bytes">retention.bytes</a> (size) and <a href="#retention.ms">retention.ms</a> (time). This limits storage consumption within the cluster, and helps complying with legal requirements such as GDPR.
  </p>

  <h4 class="anchor-heading"><a id="multitenancy-security" class="anchor-link"></a><a href="#multitenancy-security">Securing Clusters and Topics: Authentication, Authorization, Encryption</a></h4>

  <p>
  Because the documentation has a dedicated chapter on <a href="#security">security</a> that applies to any Kafka deployment, this section focuses on additional considerations for multi-tenant environments.
  </p>

  <p>
Security settings for Kafka fall into three main categories, which are similar to how administrators would secure other client-server data systems, like relational databases and traditional messaging systems.
  </p>

  <ol>
    <li><strong>Encryption</strong> of data transferred between Kafka brokers and Kafka clients, between brokers, between brokers and ZooKeeper nodes, and between brokers and other, optional tools.</li>
    <li><strong>Authentication</strong> of connections from Kafka clients and applications to Kafka brokers, as well as connections from Kafka brokers to ZooKeeper nodes.</li>
    <li><strong>Authorization</strong> of client operations such as creating, deleting, and altering the configuration of topics; writing events to or reading events from a topic; creating and deleting ACLs. Administrators can also define custom policies to put in place additional restrictions, such as a <code>CreateTopicPolicy</code> and <code>AlterConfigPolicy</code> (see <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-108%3A+Create+Topic+Policy">KIP-108</a> and the settings <a href="#brokerconfigs_create.topic.policy.class.name">create.topic.policy.class.name</a>, <a href="#brokerconfigs_alter.config.policy.class.name">alter.config.policy.class.name</a>).</li>
  </ol>

  <p>
  When securing a multi-tenant Kafka environment, the most common administrative task is the third category (authorization), i.e., managing the user/client permissions that grant or deny access to certain topics and thus to the data stored by users within a cluster. This task is performed predominantly through the <a href="#security_authz">setting of access control lists (ACLs)</a>. Here, administrators of multi-tenant environments in particular benefit from putting a hierarchical topic naming structure in place as described in a previous section, because they can conveniently control access to topics through prefixed ACLs (<code>--resource-pattern-type Prefixed</code>). This significantly minimizes the administrative overhead of securing topics in multi-tenant environments: administrators can make their own trade-offs between higher developer convenience (more lenient permissions, using fewer and broader ACLs) vs. tighter security (more stringent permissions, using more and narrower ACLs).
  </p>

  <p>
    In the following example, user Alice—a new member of ACME corporation's InfoSec team—is granted write permissions to all topics whose names start with "acme.infosec.", such as "acme.infosec.telemetry.logins" and "acme.infosec.syslogs.events".
  </p>

<pre><code class="language-bash"># Grant permissions to user Alice
$ bin/kafka-acls.sh \
    --bootstrap-server localhost:9092 \
    --add --allow-principal User:Alice \
    --producer \
    --resource-pattern-type prefixed --topic acme.infosec.</code></pre>

  <p>
    You can similarly use this approach to isolate different customers on the same shared cluster.
  </p>

  <h4 class="anchor-heading"><a id="multitenancy-isolation" class="anchor-link"></a><a href="#multitenancy-isolation">Isolating Tenants: Quotas, Rate Limiting, Throttling</a></h4>

  <p>
  Multi-tenant clusters should generally be configured with <a href="#design_quotas">quotas</a>, which protect against users (tenants) eating up too many cluster resources, such as when they attempt to write or read very high volumes of data, or create requests to brokers at an excessively high rate. This may cause network saturation, monopolize broker resources, and impact other clients—all of which you want to avoid in a shared environment.
  </p>

  <p>
  <strong>Client quotas:</strong> Kafka supports different types of (per-user principal) client quotas. Because a client's quotas apply irrespective of which topics the client is writing to or reading from, they are a convenient and effective tool to allocate resources in a multi-tenant cluster. <a href="#design_quotascpu">Request rate quotas</a>, for example, help to limit a user's impact on broker CPU usage by limiting the time a broker spends on the <a href="/protocol.html">request handling path</a> for that user, after which throttling kicks in. In many situations, isolating users with request rate quotas has a bigger impact in multi-tenant clusters than setting incoming/outgoing network bandwidth quotas, because excessive broker CPU usage for processing requests reduces the effective bandwidth the broker can serve. Furthermore, administrators can also define quotas on topic operations—such as create, delete, and alter—to prevent Kafka clusters from being overwhelmed by highly concurrent topic operations (see <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-599%3A+Throttle+Create+Topic%2C+Create+Partition+and+Delete+Topic+Operations">KIP-599</a> and the quota type <code>controller_mutation_rate</code>).
  </p>

  <p>
    <strong>Server quotas:</strong> Kafka also supports different types of broker-side quotas. For example, administrators can set a limit on the rate with which the <a href="#brokerconfigs_max.connection.creation.rate">broker accepts new connections</a>, set the <a href="#brokerconfigs_max.connections">maximum number of connections per broker</a>, or set the maximum number of connections allowed <a href="#brokerconfigs_max.connections.per.ip">from a specific IP address</a>.
  </p>

  <p>
  For more information, please refer to the <a href="#design_quotas">quota overview</a> and <a href="#quotas">how to set quotas</a>.
  </p>

  <h4 class="anchor-heading"><a id="multitenancy-monitoring" class="anchor-link"></a><a href="#multitenancy-monitoring">Monitoring and Metering</a></h4>

  <p>
  <a href="#monitoring">Monitoring</a> is a broader subject that is covered <a href="#monitoring">elsewhere</a> in the documentation. Administrators of any Kafka environment, but especially multi-tenant ones, should set up monitoring according to these instructions. Kafka supports a wide range of metrics, such as the rate of failed authentication attempts, request latency, consumer lag, total number of consumer groups, metrics on the quotas described in the previous section, and many more.
  </p>

  <p>
  For example, monitoring can be configured to track the size of topic-partitions (with the JMX metric <code>kafka.log.Log.Size.&lt;TOPIC-NAME&gt;</code>), and thus the total size of data stored in a topic. You can then define alerts when tenants on shared clusters are getting close to using too much storage space.
  </p>

  <h4 class="anchor-heading"><a id="multitenancy-georeplication" class="anchor-link"></a><a href="#multitenancy-georeplication">Multi-Tenancy and Geo-Replication</a></h4>

  <p>
    Kafka lets you share data across different clusters, which may be located in different geographical regions, data centers, and so on. Apart from use cases such as disaster recovery, this functionality is useful when a multi-tenant setup requires inter-cluster data sharing. See the section <a href="#georeplication">Geo-Replication (Cross-Cluster Data Mirroring)</a> for more information.
  </p>

  <h4 class="anchor-heading"><a id="multitenancy-more" class="anchor-link"></a><a href="#multitenancy-more">Further considerations</a></h4>

  <p>
    <strong>Data contracts:</strong> You may need to define data contracts between the producers and the consumers of data in a cluster, using event schemas. This ensures that events written to Kafka can always be read properly again, and prevents malformed or corrupt events being written. The best way to achieve this is to deploy a so-called schema registry alongside the cluster. (Kafka does not include a schema registry, but there are third-party implementations available.) A schema registry manages the event schemas and maps the schemas to topics, so that producers know which topics are accepting which types (schemas) of events, and consumers know how to read and parse events in a topic. Some registry implementations provide further functionality, such as schema evolution, storing a history of all schemas, and schema compatibility settings.
  </p>


  <h3 class="anchor-heading"><a id="config" class="anchor-link"></a><a href="#config">6.5 Kafka Configuration</a></h3>

  <h4 class="anchor-heading"><a id="clientconfig" class="anchor-link"></a><a href="#clientconfig">Important Client Configurations</a></h4>

  The most important producer configurations are:
  <ul>
      <li>acks</li>
      <li>compression</li>
      <li>batch size</li>
  </ul>
  The most important consumer configuration is the fetch size.
  <p>
  All configurations are documented in the <a href="#configuration">configuration</a> section.
  <p>
  <h4 class="anchor-heading"><a id="prodconfig" class="anchor-link"></a><a href="#prodconfig">A Production Server Config</a></h4>
  Here is an example production server configuration:
  <pre><code class="language-text"># ZooKeeper
zookeeper.connect=[list of ZooKeeper servers]

# Log configuration
num.partitions=8
default.replication.factor=3
log.dir=[List of directories. Kafka should have its own dedicated disk(s) or SSD(s).]

# Other configurations
broker.id=[An integer. Start with 0 and increment by 1 for each new broker.]
listeners=[list of listeners]
auto.create.topics.enable=false
min.insync.replicas=2
queued.max.requests=[number of concurrent requests]</code></pre>

  Our client configuration varies a fair amount between different use cases.

  <h3 class="anchor-heading"><a id="java" class="anchor-link"></a><a href="#java">6.6 Java Version</a></h3>

  Java 8, Java 11, and Java 17 are supported.
  <p>
  Note that Java 8 support project-wide has been deprecated since Apache Kafka 3.0 and Java 11 support for the broker and tools
  has been deprecated since Apache Kafka 3.7. Both will be removed in Apache Kafka 4.0.
  <p>
  Java 11 and later versions perform significantly better if TLS is enabled, so they are highly recommended (they also include a number of other
  performance improvements: G1GC, CRC32C, Compact Strings, Thread-Local Handshakes and more).
  <p>
  From a security perspective, we recommend the latest released patch version as older freely available versions have disclosed security vulnerabilities.
  <p>
  Typical arguments for running Kafka with OpenJDK-based Java implementations (including Oracle JDK) are:

  <pre><code class="language-text">-Xmx6g -Xms6g -XX:MetaspaceSize=96m -XX:+UseG1GC
-XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:G1HeapRegionSize=16M
-XX:MinMetaspaceFreeRatio=50 -XX:MaxMetaspaceFreeRatio=80 -XX:+ExplicitGCInvokesConcurrent</code></pre>

  For reference, here are the stats for one of LinkedIn's busiest clusters (at peak) that uses said Java arguments:
  <ul>
      <li>60 brokers</li>
      <li>50k partitions (replication factor 2)</li>
      <li>800k messages/sec in</li>
      <li>300 MB/sec inbound, 1 GB/sec+ outbound</li>
  </ul>

  All of the brokers in that cluster have a 90% GC pause time of about 21ms with less than 1 young GC per second.

  <h3 class="anchor-heading"><a id="hwandos" class="anchor-link"></a><a href="#hwandos">6.7 Hardware and OS</a></h3>
  We are using dual quad-core Intel Xeon machines with 24GB of memory.
  <p>
  You need sufficient memory to buffer active readers and writers. You can do a back-of-the-envelope estimate of memory needs by assuming you want to be able to buffer for 30 seconds and compute your memory need as write_throughput*30.
  <p>
  The disk throughput is important. We have 8x7200 rpm SATA drives. In general disk throughput is the performance bottleneck, and more disks is better. Depending on how you configure flush behavior you may or may not benefit from more expensive disks (if you force flush often then higher RPM SAS drives may be better).

  <h4 class="anchor-heading"><a id="os" class="anchor-link"></a><a href="#os">OS</a></h4>
  Kafka should run well on any unix system and has been tested on Linux and Solaris.
  <p>
  We have seen a few issues running on Windows and Windows is not currently a well supported platform though we would be happy to change that.
  <p>
  It is unlikely to require much OS-level tuning, but there are three potentially important OS-level configurations:
  <ul>
      <li>File descriptor limits: Kafka uses file descriptors for log segments and open connections. If a broker hosts many partitions, consider that the broker needs at least (number_of_partitions)*(partition_size/segment_size) to track all log segments in addition to the number of connections the broker makes. We recommend at least 100000 allowed file descriptors for the broker processes as a starting point. Note: The mmap() function adds an extra reference to the file associated with the file descriptor fildes which is not removed by a subsequent close() on that file descriptor. This reference is removed when there are no more mappings to the file.
      <li>Max socket buffer size: can be increased to enable high-performance data transfer between data centers as <a href="https://www.psc.edu/index.php/networking/641-tcp-tune">described here</a>.
      <li>Maximum number of memory map areas a process may have (aka vm.max_map_count). <a href="https://kernel.org/doc/Documentation/sysctl/vm.txt">See the Linux kernel documentation</a>. You should keep an eye at this OS-level property when considering the maximum number of partitions a broker may have. By default, on a number of Linux systems, the value of vm.max_map_count is somewhere around 65535. Each log segment, allocated per partition, requires a pair of index/timeindex files, and each of these files consumes 1 map area. In other words, each log segment uses 2 map areas. Thus, each partition requires minimum 2 map areas, as long as it hosts a single log segment. That is to say, creating 50000 partitions on a broker will result allocation of 100000 map areas and likely cause broker crash with OutOfMemoryError (Map failed) on a system with default vm.max_map_count. Keep in mind that the number of log segments per partition varies depending on the segment size, load intensity, retention policy and, generally, tends to be more than one.
  </ul>
  <p>

  <h4 class="anchor-heading"><a id="diskandfs" class="anchor-link"></a><a href="#diskandfs">Disks and Filesystem</a></h4>
  We recommend using multiple drives to get good throughput and not sharing the same drives used for Kafka data with application logs or other OS filesystem activity to ensure good latency. You can either RAID these drives together into a single volume or format and mount each drive as its own directory. Since Kafka has replication the redundancy provided by RAID can also be provided at the application level. This choice has several tradeoffs.
  <p>
  If you configure multiple data directories partitions will be assigned round-robin to data directories. Each partition will be entirely in one of the data directories. If data is not well balanced among partitions this can lead to load imbalance between disks.
  <p>
  RAID can potentially do better at balancing load between disks (although it doesn't always seem to) because it balances load at a lower level. The primary downside of RAID is that it is usually a big performance hit for write throughput and reduces the available disk space.
  <p>
  Another potential benefit of RAID is the ability to tolerate disk failures. However our experience has been that rebuilding the RAID array is so I/O intensive that it effectively disables the server, so this does not provide much real availability improvement.

  <h4 class="anchor-heading"><a id="appvsosflush" class="anchor-link"></a><a href="#appvsosflush">Application vs. OS Flush Management</a></h4>
  Kafka always immediately writes all data to the filesystem and supports the ability to configure the flush policy that controls when data is forced out of the OS cache and onto disk using the flush. This flush policy can be controlled to force data to disk after a period of time or after a certain number of messages has been written. There are several choices in this configuration.
  <p>
  Kafka must eventually call fsync to know that data was flushed. When recovering from a crash for any log segment not known to be fsync'd Kafka will check the integrity of each message by checking its CRC and also rebuild the accompanying offset index file as part of the recovery process executed on startup.
  <p>
  Note that durability in Kafka does not require syncing data to disk, as a failed node will always recover from its replicas.
  <p>
  We recommend using the default flush settings which disable application fsync entirely. This means relying on the background flush done by the OS and Kafka's own background flush. This provides the best of all worlds for most uses: no knobs to tune, great throughput and latency, and full recovery guarantees. We generally feel that the guarantees provided by replication are stronger than sync to local disk, however the paranoid still may prefer having both and application level fsync policies are still supported.
  <p>
  The drawback of using application level flush settings is that it is less efficient in its disk usage pattern (it gives the OS less leeway to re-order writes) and it can introduce latency as fsync in most Linux filesystems blocks writes to the file whereas the background flushing does much more granular page-level locking.
  <p>
  In general you don't need to do any low-level tuning of the filesystem, but in the next few sections we will go over some of this in case it is useful.

  <h4 class="anchor-heading"><a id="linuxflush" class="anchor-link"></a><a href="#linuxflush">Understanding Linux OS Flush Behavior</a></h4>

  In Linux, data written to the filesystem is maintained in <a href="https://en.wikipedia.org/wiki/Page_cache">pagecache</a> until it must be written out to disk (due to an application-level fsync or the OS's own flush policy). The flushing of data is done by a set of background threads called pdflush (or in post 2.6.32 kernels "flusher threads").
  <p>
  Pdflush has a configurable policy that controls how much dirty data can be maintained in cache and for how long before it must be written back to disk.
  This policy is described <a href="https://web.archive.org/web/20160518040713/http://www.westnet.com/~gsmith/content/linux-pdflush.htm">here</a>.
  When Pdflush cannot keep up with the rate of data being written it will eventually cause the writing process to block incurring latency in the writes to slow down the accumulation of data.
  <p>
  You can see the current state of OS memory usage by doing
  <pre><code class="language-bash">$ cat /proc/meminfo</code></pre>
  The meaning of these values are described in the link above.
  <p>
  Using pagecache has several advantages over an in-process cache for storing data that will be written out to disk:
  <ul>
    <li>The I/O scheduler will batch together consecutive small writes into bigger physical writes which improves throughput.
    <li>The I/O scheduler will attempt to re-sequence writes to minimize movement of the disk head which improves throughput.
    <li>It automatically uses all the free memory on the machine
  </ul>

  <h4 class="anchor-heading"><a id="filesystems" class="anchor-link"></a><a href="#filesystems">Filesystem Selection</a></h4>
  <p>Kafka uses regular files on disk, and as such it has no hard dependency on a specific filesystem. The two filesystems which have the most usage, however, are EXT4 and XFS. Historically, EXT4 has had more usage, but recent improvements to the XFS filesystem have shown it to have better performance characteristics for Kafka's workload with no compromise in stability.</p>
  <p>Comparison testing was performed on a cluster with significant message loads, using a variety of filesystem creation and mount options. The primary metric in Kafka that was monitored was the "Request Local Time", indicating the amount of time append operations were taking. XFS resulted in much better local times (160ms vs. 250ms+ for the best EXT4 configuration), as well as lower average wait times. The XFS performance also showed less variability in disk performance.</p>
  <h5 class="anchor-heading"><a id="generalfs" class="anchor-link"></a><a href="#generalfs">General Filesystem Notes</a></h5>
  For any filesystem used for data directories, on Linux systems, the following options are recommended to be used at mount time:
  <ul>
    <li>noatime: This option disables updating of a file's atime (last access time) attribute when the file is read. This can eliminate a significant number of filesystem writes, especially in the case of bootstrapping consumers. Kafka does not rely on the atime attributes at all, so it is safe to disable this.</li>
  </ul>
  <h5 class="anchor-heading"><a id="xfs" class="anchor-link"></a><a href="#xfs">XFS Notes</a></h5>
  The XFS filesystem has a significant amount of auto-tuning in place, so it does not require any change in the default settings, either at filesystem creation time or at mount. The only tuning parameters worth considering are:
  <ul>
    <li>largeio: This affects the preferred I/O size reported by the stat call. While this can allow for higher performance on larger disk writes, in practice it had minimal or no effect on performance.</li>
    <li>nobarrier: For underlying devices that have battery-backed cache, this option can provide a little more performance by disabling periodic write flushes. However, if the underlying device is well-behaved, it will report to the filesystem that it does not require flushes, and this option will have no effect.</li>
  </ul>
  <h5 class="anchor-heading"><a id="ext4" class="anchor-link"></a><a href="#ext4">EXT4 Notes</a></h5>
  EXT4 is a serviceable choice of filesystem for the Kafka data directories, however getting the most performance out of it will require adjusting several mount options. In addition, these options are generally unsafe in a failure scenario, and will result in much more data loss and corruption. For a single broker failure, this is not much of a concern as the disk can be wiped and the replicas rebuilt from the cluster. In a multiple-failure scenario, such as a power outage, this can mean underlying filesystem (and therefore data) corruption that is not easily recoverable. The following options can be adjusted:
  <ul>
    <li>data=writeback: Ext4 defaults to data=ordered which puts a strong order on some writes. Kafka does not require this ordering as it does very paranoid data recovery on all unflushed log. This setting removes the ordering constraint and seems to significantly reduce latency.
    <li>Disabling journaling: Journaling is a tradeoff: it makes reboots faster after server crashes but it introduces a great deal of additional locking which adds variance to write performance. Those who don't care about reboot time and want to reduce a major source of write latency spikes can turn off journaling entirely.
    <li>commit=num_secs: This tunes the frequency with which ext4 commits to its metadata journal. Setting this to a lower value reduces the loss of unflushed data during a crash. Setting this to a higher value will improve throughput.
    <li>nobh: This setting controls additional ordering guarantees when using data=writeback mode. This should be safe with Kafka as we do not depend on write ordering and improves throughput and latency.
    <li>delalloc: Delayed allocation means that the filesystem avoid allocating any blocks until the physical write occurs. This allows ext4 to allocate a large extent instead of smaller pages and helps ensure the data is written sequentially. This feature is great for throughput. It does seem to involve some locking in the filesystem which adds a bit of latency variance.
    <li>fast_commit: Added in Linux 5.10, <a href="https://lwn.net/Articles/842385/">fast_commit</a> is a lighter-weight journaling method which can be used with data=ordered journaling mode. Enabling it seems to significantly reduce latency.
  </ul>

  <h4 class="anchor-heading"><a id="replace_disk" class="anchor-link"></a><a href="#replace_disk">Replace KRaft Controller Disk</a></h4>
  <p>When Kafka is configured to use KRaft, the controllers store the cluster metadata in the directory specified in <code>metadata.log.dir</code> -- or the first log directory, if <code>metadata.log.dir</code> is not configured. See the documentation for <code>metadata.log.dir</code> for details.</p>

  <p>If the data in the cluster metadata directory is lost either because of hardware failure or the hardware needs to be replaced, care should be taken when provisioning the new controller node. The new controller node should not be formatted and started until the majority of the controllers have all of the committed data. To determine if the majority of the controllers have the committed data, run the <code>kafka-metadata-quorum.sh</code> tool to describe the replication status:</p>

  <pre><code class="language-bash">$ bin/kafka-metadata-quorum.sh --bootstrap-server localhost:9092 describe --replication
NodeId  LogEndOffset    Lag     LastFetchTimestamp      LastCaughtUpTimestamp   Status
1       25806           0       1662500992757           1662500992757           Leader
...     ...             ...     ...                     ...                     ...</code></pre>

  <p>Check and wait until the <code>Lag</code> is small for a majority of the controllers. If the leader's end offset is not increasing, you can wait until the lag is 0 for a majority; otherwise, you can pick the latest leader end offset and wait until all replicas have reached it. Check and wait until the <code>LastFetchTimestamp</code> and <code>LastCaughtUpTimestamp</code> are close to each other for the majority of the controllers. At this point it is safer to format the controller's metadata log directory. This can be done by running the <code>kafka-storage.sh</code> command.</p>

  <pre><code class="language-bash">$ bin/kafka-storage.sh format --cluster-id uuid --config server_properties</code></pre>

  <p>It is possible for the <code>bin/kafka-storage.sh format</code> command above to fail with a message like <code>Log directory ... is already formatted</code>. This can happen when combined mode is used and only the metadata log directory was lost but not the others. In that case and only in that case, can you run the <code>kafka-storage.sh format</code> command with the <code>--ignore-formatted</code> option.</p>

  <p>Start the KRaft controller after formatting the log directories.</p>

  <pre><code class="language-bash">$ bin/kafka-server-start.sh server_properties</code></pre>

  <h3 class="anchor-heading"><a id="monitoring" class="anchor-link"></a><a href="#monitoring">6.8 Monitoring</a></h3>

  Kafka uses Yammer Metrics for metrics reporting in the server. The Java clients use Kafka Metrics, a built-in metrics registry that minimizes transitive dependencies pulled into client applications. Both expose metrics via JMX and can be configured to report stats using pluggable stats reporters to hook up to your monitoring system.
  <p>
  All Kafka rate metrics have a corresponding cumulative count metric with suffix <code>-total</code>. For example,
  <code>records-consumed-rate</code> has a corresponding metric named <code>records-consumed-total</code>.
  <p>
  The easiest way to see the available metrics is to fire up jconsole and point it at a running kafka client or server; this will allow browsing all metrics with JMX.

  <h4 class="anchor-heading"><a id="remote_jmx" class="anchor-link"></a><a href="#remote_jmx">Security Considerations for Remote Monitoring using JMX</a></h4>
  Apache Kafka disables remote JMX by default. You can enable remote monitoring using JMX by setting the environment variable
  <code>JMX_PORT</code> for processes started using the CLI or standard Java system properties to enable remote JMX programmatically.
  You must enable security when enabling remote JMX in production scenarios to ensure that unauthorized users cannot monitor or
  control your broker or application as well as the platform on which these are running. Note that authentication is disabled for
  JMX by default in Kafka and security configs must be overridden for production deployments by setting the environment variable
  <code>KAFKA_JMX_OPTS</code> for processes started using the CLI or by setting appropriate Java system properties. See
  <a href="https://docs.oracle.com/javase/8/docs/technotes/guides/management/agent.html">Monitoring and Management Using JMX Technology</a>
  for details on securing JMX.
  <p>
  We do graphing and alerting on the following metrics:
  <table class="data-table">
  <tbody><tr>
        <th>Description</th>
        <th>Mbean name</th>
        <th>Normal value</th>
      </tr>
      <tr>
        <td>Message in rate</td>
        <td>kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec,topic=([-.\w]+)</td>
        <td>Incoming message rate per topic. Omitting 'topic=(...)' will yield the all-topic rate.</td>
      </tr>
      <tr>
        <td>Byte in rate from clients</td>
        <td>kafka.server:type=BrokerTopicMetrics,name=BytesInPerSec,topic=([-.\w]+)</td>
        <td>Byte in (from the clients) rate per topic. Omitting 'topic=(...)' will yield the all-topic rate.</td>
      </tr>
      <tr>
        <td>Byte in rate from other brokers</td>
        <td>kafka.server:type=BrokerTopicMetrics,name=ReplicationBytesInPerSec</td>
        <td>Byte in (from the other brokers) rate across all topics.</td>
      </tr>
      <tr>
        <td>Controller Request rate from Broker</td>
        <td>kafka.controller:type=ControllerChannelManager,name=RequestRateAndQueueTimeMs,brokerId=([0-9]+)</td>
        <td>The rate (requests per second) at which the ControllerChannelManager takes requests from the
            queue of the given broker. And the time it takes for a request to stay in this queue before
            it is taken from the queue.</td>
      </tr>
      <tr>
        <td>Controller Event queue size</td>
        <td>kafka.controller:type=ControllerEventManager,name=EventQueueSize</td>
        <td>Size of the ControllerEventManager's queue.</td>
      </tr>
      <tr>
        <td>Controller Event queue time</td>
        <td>kafka.controller:type=ControllerEventManager,name=EventQueueTimeMs</td>
        <td>Time that takes for any event (except the Idle event) to wait in the ControllerEventManager's
            queue before being processed</td>
      </tr>
      <tr>
        <td>Request rate</td>
        <td>kafka.network:type=RequestMetrics,name=RequestsPerSec,request={Produce|FetchConsumer|FetchFollower},version=([0-9]+)</td>
        <td></td>
      </tr>
      <tr>
        <td>Error rate</td>
        <td>kafka.network:type=RequestMetrics,name=ErrorsPerSec,request=([-.\w]+),error=([-.\w]+)</td>
        <td>Number of errors in responses counted per-request-type, per-error-code. If a response contains
          multiple errors, all are counted. error=NONE indicates successful responses.</td>
      </tr>
      <tr>
        <td>Produce request rate</td>
        <td>kafka.server:type=BrokerTopicMetrics,name=TotalProduceRequestsPerSec,topic=([-.\w]+)</td>
        <td>Produce request rate per topic. Omitting 'topic=(...)' will yield the all-topic rate.</td>
      </tr>
      <tr>
        <td>Fetch request rate</td>
        <td>kafka.server:type=BrokerTopicMetrics,name=TotalFetchRequestsPerSec,topic=([-.\w]+)</td>
        <td>Fetch request (from clients or followers) rate per topic. Omitting 'topic=(...)' will yield the all-topic rate.</td>
      </tr>
      <tr>
        <td>Failed produce request rate</td>
        <td>kafka.server:type=BrokerTopicMetrics,name=FailedProduceRequestsPerSec,topic=([-.\w]+)</td>
        <td>Failed Produce request rate per topic. Omitting 'topic=(...)' will yield the all-topic rate.</td>
      </tr>
      <tr>
        <td>Failed fetch request rate</td>
        <td>kafka.server:type=BrokerTopicMetrics,name=FailedFetchRequestsPerSec,topic=([-.\w]+)</td>
        <td>Failed Fetch request (from clients or followers) rate per topic. Omitting 'topic=(...)' will yield the all-topic rate.</td>
      </tr>
      <tr>
        <td>Request size in bytes</td>
        <td>kafka.network:type=RequestMetrics,name=RequestBytes,request=([-.\w]+)</td>
        <td>Size of requests for each request type.</td>
      </tr>
      <tr>
        <td>Temporary memory size in bytes</td>
        <td>kafka.network:type=RequestMetrics,name=TemporaryMemoryBytes,request={Produce|Fetch}</td>
        <td>Temporary memory used for message format conversions and decompression.</td>
      </tr>
      <tr>
        <td>Message conversion time</td>
        <td>kafka.network:type=RequestMetrics,name=MessageConversionsTimeMs,request={Produce|Fetch}</td>
        <td>Time in milliseconds spent on message format conversions.</td>
      </tr>
      <tr>
        <td>Message conversion rate</td>
        <td>kafka.server:type=BrokerTopicMetrics,name={Produce|Fetch}MessageConversionsPerSec,topic=([-.\w]+)</td>
        <td>Message format conversion rate, for Produce or Fetch requests, per topic. Omitting 'topic=(...)' will yield the all-topic rate.</td>
      </tr>
      <tr>
        <td>Request Queue Size</td>
        <td>kafka.network:type=RequestChannel,name=RequestQueueSize</td>
        <td>Size of the request queue.</td>
      </tr>
      <tr>
        <td>Byte out rate to clients</td>
        <td>kafka.server:type=BrokerTopicMetrics,name=BytesOutPerSec,topic=([-.\w]+)</td>
        <td>Byte out (to the clients) rate per topic. Omitting 'topic=(...)' will yield the all-topic rate.</td>
      </tr>
      <tr>
        <td>Byte out rate to other brokers</td>
        <td>kafka.server:type=BrokerTopicMetrics,name=ReplicationBytesOutPerSec</td>
        <td>Byte out (to the other brokers) rate across all topics</td>
      </tr>
      <tr>
        <td>Rejected byte rate</td>
        <td>kafka.server:type=BrokerTopicMetrics,name=BytesRejectedPerSec,topic=([-.\w]+)</td>
        <td>Rejected byte rate per topic, due to the record batch size being greater than max.message.bytes configuration. Omitting 'topic=(...)' will yield the all-topic rate.</td>
      </tr>
      <tr>
        <td>Message validation failure rate due to no key specified for compacted topic</td>
        <td>kafka.server:type=BrokerTopicMetrics,name=NoKeyCompactedTopicRecordsPerSec</td>
        <td>0</td>
      </tr>
      <tr>
        <td>Message validation failure rate due to invalid magic number</td>
        <td>kafka.server:type=BrokerTopicMetrics,name=InvalidMagicNumberRecordsPerSec</td>
        <td>0</td>
      </tr>
      <tr>
        <td>Message validation failure rate due to incorrect crc checksum</td>
        <td>kafka.server:type=BrokerTopicMetrics,name=InvalidMessageCrcRecordsPerSec</td>
        <td>0</td>
      </tr>
      <tr>
        <td>Message validation failure rate due to non-continuous offset or sequence number in batch</td>
        <td>kafka.server:type=BrokerTopicMetrics,name=InvalidOffsetOrSequenceRecordsPerSec</td>
        <td>0</td>
      </tr>
      <tr>
        <td>Log flush rate and time</td>
        <td>kafka.log:type=LogFlushStats,name=LogFlushRateAndTimeMs</td>
        <td></td>
      </tr>
      <tr>
        <td># of offline log directories</td>
        <td>kafka.log:type=LogManager,name=OfflineLogDirectoryCount</td>
        <td>0</td>
      </tr>
      <tr>
        <td>Leader election rate</td>
        <td>kafka.controller:type=ControllerStats,name=LeaderElectionRateAndTimeMs</td>
        <td>non-zero when there are broker failures</td>
      </tr>
      <tr>
        <td>Unclean leader election rate</td>
        <td>kafka.controller:type=ControllerStats,name=UncleanLeaderElectionsPerSec</td>
        <td>0</td>
      </tr>
      <tr>
        <td>Is controller active on broker</td>
        <td>kafka.controller:type=KafkaController,name=ActiveControllerCount</td>
        <td>only one broker in the cluster should have 1</td>
      </tr>
      <tr>
        <td>Pending topic deletes</td>
        <td>kafka.controller:type=KafkaController,name=TopicsToDeleteCount</td>
        <td></td>
      </tr>
      <tr>
        <td>Pending replica deletes</td>
        <td>kafka.controller:type=KafkaController,name=ReplicasToDeleteCount</td>
        <td></td>
      </tr>
      <tr>
        <td>Ineligible pending topic deletes</td>
        <td>kafka.controller:type=KafkaController,name=TopicsIneligibleToDeleteCount</td>
        <td></td>
      </tr>
      <tr>
        <td>Ineligible pending replica deletes</td>
        <td>kafka.controller:type=KafkaController,name=ReplicasIneligibleToDeleteCount</td>
        <td></td>
      </tr>
      <tr>
        <td># of under replicated partitions (|ISR| &lt |all replicas|)</td>
        <td>kafka.server:type=ReplicaManager,name=UnderReplicatedPartitions</td>
        <td>0</td>
      </tr>
      <tr>
        <td># of under minIsr partitions (|ISR| &lt min.insync.replicas)</td>
        <td>kafka.server:type=ReplicaManager,name=UnderMinIsrPartitionCount</td>
        <td>0</td>
      </tr>
      <tr>
        <td># of at minIsr partitions (|ISR| = min.insync.replicas)</td>
        <td>kafka.server:type=ReplicaManager,name=AtMinIsrPartitionCount</td>
        <td>0</td>
      </tr>
      <tr>
        <td>Producer Id counts</td>
        <td>kafka.server:type=ReplicaManager,name=ProducerIdCount</td>
        <td>Count of all producer ids created by transactional and idempotent producers in each replica on the broker</td>
      </tr>
      <tr>
        <td>Partition counts</td>
        <td>kafka.server:type=ReplicaManager,name=PartitionCount</td>
        <td>mostly even across brokers</td>
      </tr>
      <tr>
        <td>Offline Replica counts</td>
        <td>kafka.server:type=ReplicaManager,name=OfflineReplicaCount</td>
        <td>0</td>
      </tr>
      <tr>
        <td>Leader replica counts</td>
        <td>kafka.server:type=ReplicaManager,name=LeaderCount</td>
        <td>mostly even across brokers</td>
      </tr>
      <tr>
        <td>ISR shrink rate</td>
        <td>kafka.server:type=ReplicaManager,name=IsrShrinksPerSec</td>
        <td>If a broker goes down, ISR for some of the partitions will
    shrink. When that broker is up again, ISR will be expanded
    once the replicas are fully caught up. Other than that, the
    expected value for both ISR shrink rate and expansion rate is 0.</td>
      </tr>
      <tr>
        <td>ISR expansion rate</td>
        <td>kafka.server:type=ReplicaManager,name=IsrExpandsPerSec</td>
        <td>See above</td>
      </tr>
      <tr>
        <td>Failed ISR update rate</td>
        <td>kafka.server:type=ReplicaManager,name=FailedIsrUpdatesPerSec</td>
        <td>0</td>
      </tr>
      <tr>
        <td>Max lag in messages btw follower and leader replicas</td>
        <td>kafka.server:type=ReplicaFetcherManager,name=MaxLag,clientId=Replica</td>
        <td>lag should be proportional to the maximum batch size of a produce request.</td>
      </tr>
      <tr>
        <td>Lag in messages per follower replica</td>
        <td>kafka.server:type=FetcherLagMetrics,name=ConsumerLag,clientId=([-.\w]+),topic=([-.\w]+),partition=([0-9]+)</td>
        <td>lag should be proportional to the maximum batch size of a produce request.</td>
      </tr>
      <tr>
        <td>Requests waiting in the producer purgatory</td>
        <td>kafka.server:type=DelayedOperationPurgatory,name=PurgatorySize,delayedOperation=Produce</td>
        <td>non-zero if ack=-1 is used</td>
      </tr>
      <tr>
        <td>Requests waiting in the fetch purgatory</td>
        <td>kafka.server:type=DelayedOperationPurgatory,name=PurgatorySize,delayedOperation=Fetch</td>
        <td>size depends on fetch.wait.max.ms in the consumer</td>
      </tr>
      <tr>
        <td>Request total time</td>
        <td>kafka.network:type=RequestMetrics,name=TotalTimeMs,request={Produce|FetchConsumer|FetchFollower}</td>
        <td>broken into queue, local, remote and response send time</td>
      </tr>
      <tr>
        <td>Time the request waits in the request queue</td>
        <td>kafka.network:type=RequestMetrics,name=RequestQueueTimeMs,request={Produce|FetchConsumer|FetchFollower}</td>
        <td></td>
      </tr>
      <tr>
        <td>Time the request is processed at the leader</td>
        <td>kafka.network:type=RequestMetrics,name=LocalTimeMs,request={Produce|FetchConsumer|FetchFollower}</td>
        <td></td>
      </tr>
      <tr>
        <td>Time the request waits for the follower</td>
        <td>kafka.network:type=RequestMetrics,name=RemoteTimeMs,request={Produce|FetchConsumer|FetchFollower}</td>
        <td>non-zero for produce requests when ack=-1</td>
      </tr>
      <tr>
          <td>Time the request waits in the response queue</td>
          <td>kafka.network:type=RequestMetrics,name=ResponseQueueTimeMs,request={Produce|FetchConsumer|FetchFollower}</td>
          <td></td>
      </tr>
      <tr>
        <td>Time to send the response</td>
        <td>kafka.network:type=RequestMetrics,name=ResponseSendTimeMs,request={Produce|FetchConsumer|FetchFollower}</td>
        <td></td>
      </tr>
      <tr>
        <td>Number of messages the consumer lags behind the producer by. Published by the consumer, not broker.</td>
        <td>kafka.consumer:type=consumer-fetch-manager-metrics,client-id={client-id} Attribute: records-lag-max</td>
        <td></td>
      </tr>
      <tr>
        <td>The average fraction of time the network processors are idle</td>
        <td>kafka.network:type=SocketServer,name=NetworkProcessorAvgIdlePercent</td>
        <td>between 0 and 1, ideally &gt 0.3</td>
      </tr>
      <tr>
        <td>The number of connections disconnected on a processor due to a client not re-authenticating and then using the connection beyond its expiration time for anything other than re-authentication</td>
        <td>kafka.server:type=socket-server-metrics,listener=[SASL_PLAINTEXT|SASL_SSL],networkProcessor=&lt;#&gt;,name=expired-connections-killed-count</td>
        <td>ideally 0 when re-authentication is enabled, implying there are no longer any older, pre-2.2.0 clients connecting to this (listener, processor) combination</td>
      </tr>
      <tr>
        <td>The total number of connections disconnected, across all processors, due to a client not re-authenticating and then using the connection beyond its expiration time for anything other than re-authentication</td>
        <td>kafka.network:type=SocketServer,name=ExpiredConnectionsKilledCount</td>
        <td>ideally 0 when re-authentication is enabled, implying there are no longer any older, pre-2.2.0 clients connecting to this broker</td>
      </tr>
      <tr>
        <td>The average fraction of time the request handler threads are idle</td>
        <td>kafka.server:type=KafkaRequestHandlerPool,name=RequestHandlerAvgIdlePercent</td>
        <td>between 0 and 1, ideally &gt 0.3</td>
      </tr>
      <tr>
        <td>Bandwidth quota metrics per (user, client-id), user or client-id</td>
        <td>kafka.server:type={Produce|Fetch},user=([-.\w]+),client-id=([-.\w]+)</td>
        <td>Two attributes. throttle-time indicates the amount of time in ms the client was throttled. Ideally = 0.
            byte-rate indicates the data produce/consume rate of the client in bytes/sec.
            For (user, client-id) quotas, both user and client-id are specified. If per-client-id quota is applied to the client, user is not specified. If per-user quota is applied, client-id is not specified.</td>
      </tr>
      <tr>
        <td>Request quota metrics per (user, client-id), user or client-id</td>
        <td>kafka.server:type=Request,user=([-.\w]+),client-id=([-.\w]+)</td>
        <td>Two attributes. throttle-time indicates the amount of time in ms the client was throttled. Ideally = 0.
            request-time indicates the percentage of time spent in broker network and I/O threads to process requests from client group.
            For (user, client-id) quotas, both user and client-id are specified. If per-client-id quota is applied to the client, user is not specified. If per-user quota is applied, client-id is not specified.</td>
      </tr>
      <tr>
        <td>Requests exempt from throttling</td>
        <td>kafka.server:type=Request</td>
        <td>exempt-throttle-time indicates the percentage of time spent in broker network and I/O threads to process requests
            that are exempt from throttling.</td>
      </tr>
      <tr>
        <td>ZooKeeper client request latency</td>
        <td>kafka.server:type=ZooKeeperClientMetrics,name=ZooKeeperRequestLatencyMs</td>
        <td>Latency in milliseconds for ZooKeeper requests from broker.</td>
      </tr>
      <tr>
        <td>ZooKeeper connection status</td>
        <td>kafka.server:type=SessionExpireListener,name=SessionState</td>
        <td>Connection status of broker's ZooKeeper session which may be one of
            Disconnected|SyncConnected|AuthFailed|ConnectedReadOnly|SaslAuthenticated|Expired.</td>
      </tr>
      <tr>
        <td>Max time to load group metadata</td>
        <td>kafka.server:type=group-coordinator-metrics,name=partition-load-time-max</td>
        <td>maximum time, in milliseconds, it took to load offsets and group metadata from the consumer offset partitions loaded in the last 30 seconds (including time spent waiting for the loading task to be scheduled)</td>
      </tr>
      <tr>
        <td>Avg time to load group metadata</td>
        <td>kafka.server:type=group-coordinator-metrics,name=partition-load-time-avg</td>
        <td>average time, in milliseconds, it took to load offsets and group metadata from the consumer offset partitions loaded in the last 30 seconds (including time spent waiting for the loading task to be scheduled)</td>
      </tr>
      <tr>
        <td>Max time to load transaction metadata</td>
        <td>kafka.server:type=transaction-coordinator-metrics,name=partition-load-time-max</td>
        <td>maximum time, in milliseconds, it took to load transaction metadata from the consumer offset partitions loaded in the last 30 seconds (including time spent waiting for the loading task to be scheduled)</td>
      </tr>
      <tr>
        <td>Avg time to load transaction metadata</td>
        <td>kafka.server:type=transaction-coordinator-metrics,name=partition-load-time-avg</td>
        <td>average time, in milliseconds, it took to load transaction metadata from the consumer offset partitions loaded in the last 30 seconds (including time spent waiting for the loading task to be scheduled)</td>
      </tr>
      <tr>
        <td>Rate of transactional verification errors</td>
        <td>kafka.server:type=AddPartitionsToTxnManager,name=VerificationFailureRate</td>
        <td>Rate of verifications that returned in failure either from the AddPartitionsToTxn API response or through errors in the AddPartitionsToTxnManager. In steady state 0, but transient errors are expected during rolls and reassignments of the transactional state partition.</td>
      </tr>
      <tr>
        <td>Time to verify a transactional request</td>
        <td>kafka.server:type=AddPartitionsToTxnManager,name=VerificationTimeMs</td>
        <td>The amount of time queueing while a possible previous request is in-flight plus the round trip to the transaction coordinator to verify (or not verify)</td>
      </tr>
      <tr>
        <td>Consumer Group Offset Count</td>
        <td>kafka.server:type=GroupMetadataManager,name=NumOffsets</td>
        <td>Total number of committed offsets for Consumer Groups</td>
      </tr>
      <tr>
        <td>Consumer Group Count</td>
        <td>kafka.server:type=GroupMetadataManager,name=NumGroups</td>
        <td>Total number of Consumer Groups</td>
      </tr>
      <tr>
        <td>Consumer Group Count, per State</td>
        <td>kafka.server:type=GroupMetadataManager,name=NumGroups[PreparingRebalance,CompletingRebalance,Empty,Stable,Dead]</td>
        <td>The number of Consumer Groups in each state: PreparingRebalance, CompletingRebalance, Empty, Stable, Dead</td>
      </tr>
      <tr>
        <td>Number of reassigning partitions</td>
        <td>kafka.server:type=ReplicaManager,name=ReassigningPartitions</td>
        <td>The number of reassigning leader partitions on a broker.</td>
      </tr>
      <tr>
        <td>Outgoing byte rate of reassignment traffic</td>
        <td>kafka.server:type=BrokerTopicMetrics,name=ReassignmentBytesOutPerSec</td>
        <td>0; non-zero when a partition reassignment is in progress.</td>
      </tr>
      <tr>
        <td>Incoming byte rate of reassignment traffic</td>
        <td>kafka.server:type=BrokerTopicMetrics,name=ReassignmentBytesInPerSec</td>
        <td>0; non-zero when a partition reassignment is in progress.</td>
      </tr>
      <tr>
        <td>Size of a partition on disk (in bytes)</td>
        <td>kafka.log:type=Log,name=Size,topic=([-.\w]+),partition=([0-9]+)</td>
        <td>The size of a partition on disk, measured in bytes.</td>
      </tr>
      <tr>
        <td>Number of log segments in a partition</td>
        <td>kafka.log:type=Log,name=NumLogSegments,topic=([-.\w]+),partition=([0-9]+)</td>
        <td>The number of log segments in a partition.</td>
      </tr>
      <tr>
        <td>First offset in a partition</td>
        <td>kafka.log:type=Log,name=LogStartOffset,topic=([-.\w]+),partition=([0-9]+)</td>
        <td>The first offset in a partition.</td>
      </tr>
      <tr>
        <td>Last offset in a partition</td>
        <td>kafka.log:type=Log,name=LogEndOffset,topic=([-.\w]+),partition=([0-9]+)</td>
        <td>The last offset in a partition.</td>
      </tr>
  </tbody></table>

<h4 class="anchor-heading"><a id="tiered_storage_monitoring" class="anchor-link"></a><a href="#tiered_storage_monitoring">Tiered Storage Monitoring</a></h4>
  The following set of metrics are available for monitoring of the tiered storage feature:<br/><br/>
  <table class="data-table">
    <tbody>
      <tr>
        <th>Metric/Attribute name</th>
        <th>Description</th>
        <th>Mbean name</th>
      </tr>
      <tr>
        <td>Remote Fetch Bytes Per Sec</td>
        <td>Rate of bytes read from remote storage per topic. Omitting 'topic=(...)' will yield the all-topic rate</td>
        <td>kafka.server:type=BrokerTopicMetrics,name=RemoteFetchBytesPerSec,topic=([-.\w]+)</td>
      </tr>
      <tr>
        <td>Remote Fetch Requests Per Sec</td>
        <td>Rate of read requests from remote storage per topic. Omitting 'topic=(...)' will yield the all-topic rate</td>
        <td>kafka.server:type=BrokerTopicMetrics,name=RemoteFetchRequestsPerSec,topic=([-.\w]+)</td>
      </tr>
      <tr>
        <td>Remote Fetch Errors Per Sec</td>
        <td>Rate of read errors from remote storage per topic. Omitting 'topic=(...)' will yield the all-topic rate</td>
        <td>kafka.server:type=BrokerTopicMetrics,name=RemoteFetchErrorsPerSec,topic=([-.\w]+)</td>
      </tr>
      <tr>
        <td>Remote Copy Bytes Per Sec</td>
        <td>Rate of bytes copied to remote storage per topic. Omitting 'topic=(...)' will yield the all-topic rate</td>
        <td>kafka.server:type=BrokerTopicMetrics,name=RemoteCopyBytesPerSec,topic=([-.\w]+)</td>
      </tr>
      <tr>
        <td>Remote Copy Requests Per Sec</td>
        <td>Rate of write requests to remote storage per topic. Omitting 'topic=(...)' will yield the all-topic rate</td>
        <td>kafka.server:type=BrokerTopicMetrics,name=RemoteCopyRequestsPerSec,topic=([-.\w]+)</td>
      </tr>
      <tr>
        <td>Remote Copy Errors Per Sec</td>
        <td>Rate of write errors from remote storage per topic. Omitting 'topic=(...)' will yield the all-topic rate</td>
        <td>kafka.server:type=BrokerTopicMetrics,name=RemoteCopyErrorsPerSec,topic=([-.\w]+)</td>
      </tr>
      <tr>
        <td>Remote Copy Lag Bytes</td>
        <td>Bytes which are eligible for tiering, but are not in remote storage yet. Omitting 'topic=(...)' will yield the all-topic sum</td>
        <td>kafka.server:type=BrokerTopicMetrics,name=RemoteCopyLagBytes,topic=([-.\w]+)</td>
      </tr>
      <tr>
        <td>Remote Copy Lag Segments</td>
        <td>Segments which are eligible for tiering, but are not in remote storage yet. Omitting 'topic=(...)' will yield the all-topic count</td>
        <td>kafka.server:type=BrokerTopicMetrics,name=RemoteCopyLagSegments,topic=([-.\w]+)</td>
      </tr>
      <tr>
        <td>Remote Delete Requests Per Sec</td>
        <td>Rate of delete requests to remote storage per topic. Omitting 'topic=(...)' will yield the all-topic rate</td>
        <td>kafka.server:type=BrokerTopicMetrics,name=RemoteDeleteRequestsPerSec,topic=([-.\w]+)</td>
      </tr>
      <tr>
        <td>Remote Delete Errors Per Sec</td>
        <td>Rate of delete errors from remote storage per topic. Omitting 'topic=(...)' will yield the all-topic rate</td>
        <td>kafka.server:type=BrokerTopicMetrics,name=RemoteDeleteErrorsPerSec,topic=([-.\w]+)</td>
      </tr>
      <tr>
        <td>Remote Delete Lag Bytes</td>
        <td>Tiered bytes which are eligible for deletion, but have not been deleted yet. Omitting 'topic=(...)' will yield the all-topic sum</td>
        <td>kafka.server:type=BrokerTopicMetrics,name=RemoteDeleteLagBytes,topic=([-.\w]+)</td>
      </tr>
      <tr>
        <td>Remote Delete Lag Segments</td>
        <td>Tiered segments which are eligible for deletion, but have not been deleted yet. Omitting 'topic=(...)' will yield the all-topic count</td>
        <td>kafka.server:type=BrokerTopicMetrics,name=RemoteDeleteLagSegments,topic=([-.\w]+)</td>
      </tr>
      <tr>
        <td>Build Remote Log Aux State Requests Per Sec</td>
        <td>Rate of requests for rebuilding the auxiliary state from remote storage per topic. Omitting 'topic=(...)' will yield the all-topic rate</td>
        <td>kafka.server:type=BrokerTopicMetrics,name=BuildRemoteLogAuxStateRequestsPerSec,topic=([-.\w]+)</td>
      </tr>
      <tr>
        <td>Build Remote Log Aux State Errors Per Sec</td>
        <td>Rate of errors for rebuilding the auxiliary state from remote storage per topic. Omitting 'topic=(...)' will yield the all-topic rate</td>
        <td>kafka.server:type=BrokerTopicMetrics,name=BuildRemoteLogAuxStateErrorsPerSec,topic=([-.\w]+)</td>
      </tr>
      <tr>
        <td>Remote Log Size Computation Time</td>
        <td>The amount of time needed to compute the size of the remote log. Omitting 'topic=(...)' will yield the all-topic time</td>
        <td>kafka.server:type=BrokerTopicMetrics,name=RemoteLogSizeComputationTime,topic=([-.\w]+)</td>
      </tr>
      <tr>
        <td>Remote Log Size Bytes</td>
        <td>The total size of a remote log in bytes. Omitting 'topic=(...)' will yield the all-topic sum</td>
        <td>kafka.server:type=BrokerTopicMetrics,name=RemoteLogSizeBytes,topic=([-.\w]+)</td>
      </tr>
      <tr>
        <td>Remote Log Metadata Count</td>
        <td>The total number of metadata entries for remote storage. Omitting 'topic=(...)' will yield the all-topic count</td>
        <td>kafka.server:type=BrokerTopicMetrics,name=RemoteLogMetadataCount,topic=([-.\w]+)</td>
      </tr>
      <tr>
        <td>Delayed Remote Fetch Expires Per Sec</td>
        <td>The number of expired remote fetches per second. Omitting 'topic=(...)' will yield the all-topic rate</td>
        <td>kafka.server:type=DelayedRemoteFetchMetrics,name=ExpiresPerSec,topic=([-.\w]+)</td>
      </tr>
      <tr>
        <td>RemoteLogReader Task Queue Size</td>
        <td>Size of the queue holding remote storage read tasks</td>
        <td>org.apache.kafka.storage.internals.log:type=RemoteStorageThreadPool,name=RemoteLogReaderTaskQueueSize</td>
      </tr>
      <tr>
        <td>RemoteLogReader Avg Idle Percent</td>
        <td>Average idle percent of thread pool for processing remote storage read tasks</td>
        <td>org.apache.kafka.storage.internals.log:type=RemoteStorageThreadPool,name=RemoteLogReaderAvgIdlePercent</td>
      </tr>
      <tr>
        <td>RemoteLogManager Tasks Avg Idle Percent</td>
        <td>Average idle percent of thread pool for copying data to remote storage</td>
        <td>kafka.log.remote:type=RemoteLogManager,name=RemoteLogManagerTasksAvgIdlePercent</td>
      </tr>
    </tbody>
  </table>

<h4 class="anchor-heading"><a id="kraft_monitoring" class="anchor-link"></a><a href="#kraft_monitoring">KRaft Monitoring Metrics</a></h4>
The set of metrics that allow monitoring of the KRaft quorum and the metadata log.<br>
Note that some exposed metrics depend on the role of the node as defined by <code>process.roles</code>
<h5 class="anchor-heading"><a id="kraft_quorum_monitoring" class="anchor-link"></a><a href="#kraft_quorum_monitoring">KRaft Quorum Monitoring Metrics</a></h5>
These metrics are reported on both Controllers and Brokers in a KRaft Cluster
<table class="data-table">
  <tbody>
  <tr>
    <th>Metric/Attribute name</th>
    <th>Description</th>
    <th>Mbean name</th>
  </tr>
  <tr>
    <td>Current State</td>
    <td>The current state of this member; possible values are leader, candidate, voted, follower, unattached, observer.</td>
    <td>kafka.server:type=raft-metrics,name=current-state</td>
  </tr>
  <tr>
    <td>Current Leader</td>
    <td>The current quorum leader's id; -1 indicates unknown.</td>
    <td>kafka.server:type=raft-metrics,name=current-leader</td>
  </tr>
  <tr>
    <td>Current Voted</td>
    <td>The current voted leader's id; -1 indicates not voted for anyone.</td>
    <td>kafka.server:type=raft-metrics,name=current-vote</td>
  </tr>
  <tr>
    <td>Current Epoch</td>
    <td>The current quorum epoch.</td>
    <td>kafka.server:type=raft-metrics,name=current-epoch</td>
  </tr>
  <tr>
    <td>High Watermark</td>
    <td>The high watermark maintained on this member; -1 if it is unknown.</td>
    <td>kafka.server:type=raft-metrics,name=high-watermark</td>
  </tr>
  <tr>
    <td>Log End Offset</td>
    <td>The current raft log end offset.</td>
    <td>kafka.server:type=raft-metrics,name=log-end-offset</td>
  </tr>
  <tr>
    <td>Number of Unknown Voter Connections</td>
    <td>Number of unknown voters whose connection information is not cached. This value of this metric is always 0.</td>
    <td>kafka.server:type=raft-metrics,name=number-unknown-voter-connections</td>
  </tr>
  <tr>
    <td>Average Commit Latency</td>
    <td>The average time in milliseconds to commit an entry in the raft log.</td>
    <td>kafka.server:type=raft-metrics,name=commit-latency-avg</td>
  </tr>
  <tr>
    <td>Maximum Commit Latency</td>
    <td>The maximum time in milliseconds to commit an entry in the raft log.</td>
    <td>kafka.server:type=raft-metrics,name=commit-latency-max</td>
  </tr>
  <tr>
    <td>Average Election Latency</td>
    <td>The average time in milliseconds spent on electing a new leader.</td>
    <td>kafka.server:type=raft-metrics,name=election-latency-avg</td>
  </tr>
  <tr>
    <td>Maximum Election Latency</td>
    <td>The maximum time in milliseconds spent on electing a new leader.</td>
    <td>kafka.server:type=raft-metrics,name=election-latency-max</td>
  </tr>
  <tr>
    <td>Fetch Records Rate</td>
    <td>The average number of records fetched from the leader of the raft quorum.</td>
    <td>kafka.server:type=raft-metrics,name=fetch-records-rate</td>
  </tr>
  <tr>
    <td>Append Records Rate</td>
    <td>The average number of records appended per sec by the leader of the raft quorum.</td>
    <td>kafka.server:type=raft-metrics,name=append-records-rate</td>
  </tr>
  <tr>
    <td>Average Poll Idle Ratio</td>
    <td>The average fraction of time the client's poll() is idle as opposed to waiting for the user code to process records.</td>
    <td>kafka.server:type=raft-metrics,name=poll-idle-ratio-avg</td>
  </tr>
  <tr>
    <td>Current Metadata Version</td>
    <td>Outputs the feature level of the current effective metadata version.</td>
    <td>kafka.server:type=MetadataLoader,name=CurrentMetadataVersion</td>
  </tr>
  <tr>
    <td>Metadata Snapshot Load Count</td>
    <td>The total number of times we have loaded a KRaft snapshot since the process was started.</td>
    <td>kafka.server:type=MetadataLoader,name=HandleLoadSnapshotCount</td>
  </tr>
  <tr>
    <td>Latest Metadata Snapshot Size</td>
    <td>The total size in bytes of the latest snapshot that the node has generated. If none have been generated yet, this is the size
    of the latest snapshot that was loaded. If no snapshots have been generated or loaded, this is 0.</td>
    <td>kafka.server:type=SnapshotEmitter,name=LatestSnapshotGeneratedBytes</td>
  </tr>
  <tr>
    <td>Latest Metadata Snapshot Age</td>
    <td>The interval in milliseconds since the latest snapshot that the node has generated. 
    If none have been generated yet, this is approximately the time delta since the process was started.</td>
    <td>kafka.server:type=SnapshotEmitter,name=LatestSnapshotGeneratedAgeMs</td>
  </tr>
  </tbody>
</table>
<h5 class="anchor-heading"><a id="kraft_controller_monitoring" class="anchor-link"></a><a href="#kraft_controller_monitoring">KRaft Controller Monitoring Metrics</a></h5>
<table class="data-table">
  <tbody>
  <tr>
    <th>Metric/Attribute name</th>
    <th>Description</th>
    <th>Mbean name</th>
  </tr>
  <tr>
    <td>Active Controller Count</td>
    <td>The number of Active Controllers on this node. Valid values are '0' or '1'.</td>
    <td>kafka.controller:type=KafkaController,name=ActiveControllerCount</td>
  </tr>
  <tr>
    <td>Event Queue Time Ms</td>
    <td>A Histogram of the time in milliseconds that requests spent waiting in the Controller Event Queue.</td>
    <td>kafka.controller:type=ControllerEventManager,name=EventQueueTimeMs</td>
  </tr>
  <tr>
    <td>Event Queue Processing Time Ms</td>
    <td>A Histogram of the time in milliseconds that requests spent being processed in the Controller Event Queue.</td>
    <td>kafka.controller:type=ControllerEventManager,name=EventQueueProcessingTimeMs</td>
  </tr>
  <tr>
    <td>Fenced Broker Count</td>
    <td>The number of fenced brokers as observed by this Controller.</td>
    <td>kafka.controller:type=KafkaController,name=FencedBrokerCount</td>
  </tr>
  <tr>
    <td>Active Broker Count</td>
    <td>The number of active brokers as observed by this Controller.</td>
    <td>kafka.controller:type=KafkaController,name=ActiveBrokerCount</td>
  </tr>
  <tr>
    <td>Migrating ZK Broker Count</td>
    <td>The number of brokers registered with the Controller that haven't yet migrated to KRaft mode.</td>
    <td>kafka.controller:type=KafkaController,name=MigratingZkBrokerCount</td>
  </tr>
  <tr>
    <td>ZK Migrating State</td>
    <td>
      <ul style="list-style: none; padding-left: 0; margin: 0;">
        <li>0 - NONE, cluster created in KRaft mode;</li>
        <li>4 - ZK, Migration has not started, controller is a ZK controller;</li>
        <li>2 - PRE_MIGRATION, the KRaft Controller is waiting for all ZK brokers to register in migration mode;</li>
        <li>1 - MIGRATION, ZK metadata has been migrated, but some broker is still running in ZK mode;</li>
        <li>3 - POST_MIGRATION, the cluster migration is complete;</li>
      </ul>
    </td>
    <td>kafka.controller:type=KafkaController,name=ZkMigrationState</td>
  </tr>
  <tr>
    <td>Global Topic Count</td>
    <td>The number of global topics as observed by this Controller.</td>
    <td>kafka.controller:type=KafkaController,name=GlobalTopicCount</td>
  </tr>
  <tr>
    <td>Global Partition Count</td>
    <td>The number of global partitions as observed by this Controller.</td>
    <td>kafka.controller:type=KafkaController,name=GlobalPartitionCount</td>
  </tr>
  <tr>
    <td>Offline Partition Count</td>
    <td>The number of offline topic partitions (non-internal) as observed by this Controller.</td>
    <td>kafka.controller:type=KafkaController,name=OfflinePartitionsCount</td>
  </tr>
  <tr>
    <td>Preferred Replica Imbalance Count</td>
    <td>The count of topic partitions for which the leader is not the preferred leader.</td>
    <td>kafka.controller:type=KafkaController,name=PreferredReplicaImbalanceCount</td>
  </tr>
  <tr>
    <td>Metadata Error Count</td>
    <td>The number of times this controller node has encountered an error during metadata log processing.</td>
    <td>kafka.controller:type=KafkaController,name=MetadataErrorCount</td>
  </tr>
  <tr>
    <td>Last Applied Record Offset</td>
    <td>The offset of the last record from the cluster metadata partition that was applied by the Controller.</td>
    <td>kafka.controller:type=KafkaController,name=LastAppliedRecordOffset</td>
  </tr>
  <tr>
    <td>Last Committed Record Offset</td>
    <td>The offset of the last record committed to this Controller.</td>
    <td>kafka.controller:type=KafkaController,name=LastCommittedRecordOffset</td>
  </tr>
  <tr>
    <td>Last Applied Record Timestamp</td>
    <td>The timestamp of the last record from the cluster metadata partition that was applied by the Controller.</td>
    <td>kafka.controller:type=KafkaController,name=LastAppliedRecordTimestamp</td>
  </tr>
  <tr>
    <td>Last Applied Record Lag Ms</td>
    <td>The difference between now and the timestamp of the last record from the cluster metadata partition that was applied by the controller.
    For active Controllers the value of this lag is always zero.</td>
    <td>kafka.controller:type=KafkaController,name=LastAppliedRecordLagMs</td>
  </tr>
  <tr>
    <td>ZooKeeper Write Behind Lag</td>
    <td>The amount of lag in records that ZooKeeper is behind relative to the highest committed record in the metadata log. 
    This metric will only be reported by the active KRaft controller.</td>
    <td>kafka.controller:type=KafkaController,name=ZkWriteBehindLag</td>
  </tr>
  <tr>
    <td>ZooKeeper Metadata Snapshot Write Time</td>
    <td>The number of milliseconds the KRaft controller took reconciling a snapshot into ZooKeeper.</td>
    <td>kafka.controller:type=KafkaController,name=ZkWriteSnapshotTimeMs</td>
  </tr>
  <tr>
    <td>ZooKeeper Metadata Delta Write Time</td>
    <td>The number of milliseconds the KRaft controller took writing a delta into ZK.</td>
    <td>kafka.controller:type=KafkaController,name=ZkWriteDeltaTimeMs</td>
  </tr>
  <tr>
    <td>Timed-out Broker Heartbeat Count</td>
    <td>The number of broker heartbeats that timed out on this controller since the process was started. Note that only 
    active controllers handle heartbeats, so only they will see increases in this metric.</td>
    <td>kafka.controller:type=KafkaController,name=TimedOutBrokerHeartbeatCount</td>
  </tr>
  <tr>
    <td>Number Of Operations Started In Event Queue</td>
    <td>The total number of controller event queue operations that were started. This includes deferred operations.</td>
    <td>kafka.controller:type=KafkaController,name=EventQueueOperationsStartedCount</td>
  </tr>
  <tr>
    <td>Number of Operations Timed Out In Event Queue</td>
    <td>The total number of controller event queue operations that timed out before they could be performed.</td>
    <td>kafka.controller:type=KafkaController,name=EventQueueOperationsTimedOutCount</td>
  </tr>
  <tr>
    <td>Number Of New Controller Elections</td>
    <td>Counts the number of times this node has seen a new controller elected. A transition to the "no leader" state 
    is not counted here. If the same controller as before becomes active, that still counts.</td>
    <td>kafka.controller:type=KafkaController,name=NewActiveControllersCount</td>
  </tr>
  </tbody>
</table>
  <h5 class="anchor-heading"><a id="kraft_broker_monitoring" class="anchor-link"></a><a href="#kraft_broker_monitoring">KRaft Broker Monitoring Metrics</a></h5>
  <table class="data-table">
    <tbody>
    <tr>
      <th>Metric/Attribute name</th>
      <th>Description</th>
      <th>Mbean name</th>
    </tr>
  <tr>
    <td>Last Applied Record Offset</td>
    <td>The offset of the last record from the cluster metadata partition that was applied by the broker</td>
    <td>kafka.server:type=broker-metadata-metrics,name=last-applied-record-offset</td>
  </tr>
  <tr>
    <td>Last Applied Record Timestamp</td>
    <td>The timestamp of the last record from the cluster metadata partition that was applied by the broker.</td>
    <td>kafka.server:type=broker-metadata-metrics,name=last-applied-record-timestamp</td>
  </tr>
  <tr>
    <td>Last Applied Record Lag Ms</td>
    <td>The difference between now and the timestamp of the last record from the cluster metadata partition that was applied by the broker</td>
    <td>kafka.server:type=broker-metadata-metrics,name=last-applied-record-lag-ms</td>
  </tr>
  <tr>
    <td>Metadata Load Error Count</td>
    <td>The number of errors encountered by the BrokerMetadataListener while loading the metadata log and generating a new MetadataDelta based on it.</td>
    <td>kafka.server:type=broker-metadata-metrics,name=metadata-load-error-count</td>
  </tr>
  <tr>
    <td>Metadata Apply Error Count</td>
    <td>The number of errors encountered by the BrokerMetadataPublisher while applying a new MetadataImage based on the latest MetadataDelta.</td>
    <td>kafka.server:type=broker-metadata-metrics,name=metadata-apply-error-count</td>
  </tr>
  </tbody>
</table>
  <h4><a id="selector_monitoring" href="#selector_monitoring">Common monitoring metrics for producer/consumer/connect/streams</a></h4>

  The following metrics are available on producer/consumer/connector/streams instances. For specific metrics, please see following sections.

  <table class="data-table">
    <tbody>
      <tr>
        <th>Metric/Attribute name</th>
        <th>Description</th>
        <th>Mbean name</th>
      </tr>
      <tr>
        <td>connection-close-rate</td>
        <td>Connections closed per second in the window.</td>
        <td>kafka.[producer|consumer|connect]:type=[producer|consumer|connect]-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>connection-close-total</td>
        <td>Total connections closed in the window.</td>
        <td>kafka.[producer|consumer|connect]:type=[producer|consumer|connect]-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>connection-creation-rate</td>
        <td>New connections established per second in the window.</td>
        <td>kafka.[producer|consumer|connect]:type=[producer|consumer|connect]-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>connection-creation-total</td>
        <td>Total new connections established in the window.</td>
        <td>kafka.[producer|consumer|connect]:type=[producer|consumer|connect]-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>network-io-rate</td>
        <td>The average number of network operations (reads or writes) on all connections per second.</td>
        <td>kafka.[producer|consumer|connect]:type=[producer|consumer|connect]-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>network-io-total</td>
        <td>The total number of network operations (reads or writes) on all connections.</td>
        <td>kafka.[producer|consumer|connect]:type=[producer|consumer|connect]-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>outgoing-byte-rate</td>
        <td>The average number of outgoing bytes sent per second to all servers.</td>
        <td>kafka.[producer|consumer|connect]:type=[producer|consumer|connect]-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>outgoing-byte-total</td>
        <td>The total number of outgoing bytes sent to all servers.</td>
        <td>kafka.[producer|consumer|connect]:type=[producer|consumer|connect]-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>request-rate</td>
        <td>The average number of requests sent per second.</td>
        <td>kafka.[producer|consumer|connect]:type=[producer|consumer|connect]-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>request-total</td>
        <td>The total number of requests sent.</td>
        <td>kafka.[producer|consumer|connect]:type=[producer|consumer|connect]-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>request-size-avg</td>
        <td>The average size of all requests in the window.</td>
        <td>kafka.[producer|consumer|connect]:type=[producer|consumer|connect]-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>request-size-max</td>
        <td>The maximum size of any request sent in the window.</td>
        <td>kafka.[producer|consumer|connect]:type=[producer|consumer|connect]-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>incoming-byte-rate</td>
        <td>Bytes/second read off all sockets.</td>
        <td>kafka.[producer|consumer|connect]:type=[producer|consumer|connect]-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>incoming-byte-total</td>
        <td>Total bytes read off all sockets.</td>
        <td>kafka.[producer|consumer|connect]:type=[producer|consumer|connect]-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>response-rate</td>
        <td>Responses received per second.</td>
        <td>kafka.[producer|consumer|connect]:type=[producer|consumer|connect]-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>response-total</td>
        <td>Total responses received.</td>
        <td>kafka.[producer|consumer|connect]:type=[producer|consumer|connect]-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>select-rate</td>
        <td>Number of times the I/O layer checked for new I/O to perform per second.</td>
        <td>kafka.[producer|consumer|connect]:type=[producer|consumer|connect]-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>select-total</td>
        <td>Total number of times the I/O layer checked for new I/O to perform.</td>
        <td>kafka.[producer|consumer|connect]:type=[producer|consumer|connect]-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>io-wait-time-ns-avg</td>
        <td>The average length of time the I/O thread spent waiting for a socket ready for reads or writes in nanoseconds.</td>
        <td>kafka.[producer|consumer|connect]:type=[producer|consumer|connect]-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>io-wait-time-ns-total</td>
        <td>The total time the I/O thread spent waiting in nanoseconds.</td>
        <td>kafka.[producer|consumer|connect]:type=[producer|consumer|connect]-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>io-waittime-total</td>
        <td><b>*Deprecated*</b> The total time the I/O thread spent waiting in nanoseconds. Replacement is <code>io-wait-time-ns-total</code>.</td>
        <td>kafka.[producer|consumer|connect]:type=[producer|consumer|connect]-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>io-wait-ratio</td>
        <td>The fraction of time the I/O thread spent waiting.</td>
        <td>kafka.[producer|consumer|connect]:type=[producer|consumer|connect]-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>io-time-ns-avg</td>
        <td>The average length of time for I/O per select call in nanoseconds.</td>
        <td>kafka.[producer|consumer|connect]:type=[producer|consumer|connect]-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>io-time-ns-total</td>
        <td>The total time the I/O thread spent doing I/O in nanoseconds.</td>
        <td>kafka.[producer|consumer|connect]:type=[producer|consumer|connect]-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>iotime-total</td>
        <td><b>*Deprecated*</b> The total time the I/O thread spent doing I/O in nanoseconds. Replacement is <code>io-time-ns-total</code>.</td>
        <td>kafka.[producer|consumer|connect]:type=[producer|consumer|connect]-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>io-ratio</td>
        <td>The fraction of time the I/O thread spent doing I/O.</td>
        <td>kafka.[producer|consumer|connect]:type=[producer|consumer|connect]-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>connection-count</td>
        <td>The current number of active connections.</td>
        <td>kafka.[producer|consumer|connect]:type=[producer|consumer|connect]-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>successful-authentication-rate</td>
        <td>Connections per second that were successfully authenticated using SASL or SSL.</td>
        <td>kafka.[producer|consumer|connect]:type=[producer|consumer|connect]-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>successful-authentication-total</td>
        <td>Total connections that were successfully authenticated using SASL or SSL.</td>
        <td>kafka.[producer|consumer|connect]:type=[producer|consumer|connect]-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>failed-authentication-rate</td>
        <td>Connections per second that failed authentication.</td>
        <td>kafka.[producer|consumer|connect]:type=[producer|consumer|connect]-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>failed-authentication-total</td>
        <td>Total connections that failed authentication.</td>
        <td>kafka.[producer|consumer|connect]:type=[producer|consumer|connect]-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>successful-reauthentication-rate</td>
        <td>Connections per second that were successfully re-authenticated using SASL.</td>
        <td>kafka.[producer|consumer|connect]:type=[producer|consumer|connect]-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>successful-reauthentication-total</td>
        <td>Total connections that were successfully re-authenticated using SASL.</td>
        <td>kafka.[producer|consumer|connect]:type=[producer|consumer|connect]-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>reauthentication-latency-max</td>
        <td>The maximum latency in ms observed due to re-authentication.</td>
        <td>kafka.[producer|consumer|connect]:type=[producer|consumer|connect]-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>reauthentication-latency-avg</td>
        <td>The average latency in ms observed due to re-authentication.</td>
        <td>kafka.[producer|consumer|connect]:type=[producer|consumer|connect]-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>failed-reauthentication-rate</td>
        <td>Connections per second that failed re-authentication.</td>
        <td>kafka.[producer|consumer|connect]:type=[producer|consumer|connect]-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>failed-reauthentication-total</td>
        <td>Total connections that failed re-authentication.</td>
        <td>kafka.[producer|consumer|connect]:type=[producer|consumer|connect]-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>successful-authentication-no-reauth-total</td>
        <td>Total connections that were successfully authenticated by older, pre-2.2.0 SASL clients that do not support re-authentication. May only be non-zero.</td>
        <td>kafka.[producer|consumer|connect]:type=[producer|consumer|connect]-metrics,client-id=([-.\w]+)</td>
      </tr>
    </tbody>
  </table>

  <h4><a id="common_node_monitoring" href="#common_node_monitoring">Common Per-broker metrics for producer/consumer/connect/streams</a></h4>

  The following metrics are available on producer/consumer/connector/streams instances. For specific metrics, please see following sections.

  <table class="data-table">
    <tbody>
      <tr>
        <th>Metric/Attribute name</th>
        <th>Description</th>
        <th>Mbean name</th>
      </tr>
      <tr>
        <td>outgoing-byte-rate</td>
        <td>The average number of outgoing bytes sent per second for a node.</td>
        <td>kafka.[producer|consumer|connect]:type=[consumer|producer|connect]-node-metrics,client-id=([-.\w]+),node-id=([0-9]+)</td>
      </tr>
      <tr>
        <td>outgoing-byte-total</td>
        <td>The total number of outgoing bytes sent for a node.</td>
        <td>kafka.[producer|consumer|connect]:type=[consumer|producer|connect]-node-metrics,client-id=([-.\w]+),node-id=([0-9]+)</td>
      </tr>
      <tr>
        <td>request-rate</td>
        <td>The average number of requests sent per second for a node.</td>
        <td>kafka.[producer|consumer|connect]:type=[consumer|producer|connect]-node-metrics,client-id=([-.\w]+),node-id=([0-9]+)</td>
      </tr>
      <tr>
        <td>request-total</td>
        <td>The total number of requests sent for a node.</td>
        <td>kafka.[producer|consumer|connect]:type=[consumer|producer|connect]-node-metrics,client-id=([-.\w]+),node-id=([0-9]+)</td>
      </tr>
      <tr>
        <td>request-size-avg</td>
        <td>The average size of all requests in the window for a node.</td>
        <td>kafka.[producer|consumer|connect]:type=[consumer|producer|connect]-node-metrics,client-id=([-.\w]+),node-id=([0-9]+)</td>
      </tr>
      <tr>
        <td>request-size-max</td>
        <td>The maximum size of any request sent in the window for a node.</td>
        <td>kafka.[producer|consumer|connect]:type=[consumer|producer|connect]-node-metrics,client-id=([-.\w]+),node-id=([0-9]+)</td>
      </tr>
      <tr>
        <td>incoming-byte-rate</td>
        <td>The average number of bytes received per second for a node.</td>
        <td>kafka.[producer|consumer|connect]:type=[consumer|producer|connect]-node-metrics,client-id=([-.\w]+),node-id=([0-9]+)</td>
      </tr>
      <tr>
        <td>incoming-byte-total</td>
        <td>The total number of bytes received for a node.</td>
        <td>kafka.[producer|consumer|connect]:type=[consumer|producer|connect]-node-metrics,client-id=([-.\w]+),node-id=([0-9]+)</td>
      </tr>
      <tr>
        <td>request-latency-avg</td>
        <td>The average request latency in ms for a node.</td>
        <td>kafka.[producer|consumer|connect]:type=[consumer|producer|connect]-node-metrics,client-id=([-.\w]+),node-id=([0-9]+)</td>
      </tr>
      <tr>
        <td>request-latency-max</td>
        <td>The maximum request latency in ms for a node.</td>
        <td>kafka.[producer|consumer|connect]:type=[consumer|producer|connect]-node-metrics,client-id=([-.\w]+),node-id=([0-9]+)</td>
      </tr>
      <tr>
        <td>response-rate</td>
        <td>Responses received per second for a node.</td>
        <td>kafka.[producer|consumer|connect]:type=[consumer|producer|connect]-node-metrics,client-id=([-.\w]+),node-id=([0-9]+)</td>
      </tr>
      <tr>
        <td>response-total</td>
        <td>Total responses received for a node.</td>
        <td>kafka.[producer|consumer|connect]:type=[consumer|producer|connect]-node-metrics,client-id=([-.\w]+),node-id=([0-9]+)</td>
      </tr>
    </tbody>
  </table>

  <h4 class="anchor-heading"><a id="producer_monitoring" class="anchor-link"></a><a href="#producer_monitoring">Producer monitoring</a></h4>

  The following metrics are available on producer instances.

  <table class="data-table">
  <tbody><tr>
        <th>Metric/Attribute name</th>
        <th>Description</th>
        <th>Mbean name</th>
      </tr>
        <tr>
        <td>waiting-threads</td>
        <td>The number of user threads blocked waiting for buffer memory to enqueue their records.</td>
        <td>kafka.producer:type=producer-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>buffer-total-bytes</td>
        <td>The maximum amount of buffer memory the client can use (whether or not it is currently used).</td>
        <td>kafka.producer:type=producer-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>buffer-available-bytes</td>
        <td>The total amount of buffer memory that is not being used (either unallocated or in the free list).</td>
        <td>kafka.producer:type=producer-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>buffer-exhausted-rate</td>
        <td>The average per-second number of record sends that are dropped due to buffer exhaustion</td>
        <td>kafka.producer:type=producer-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>buffer-exhausted-total</td>
        <td>The total number of record sends that are dropped due to buffer exhaustion</td>
        <td>kafka.producer:type=producer-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>bufferpool-wait-time</td>
        <td>The fraction of time an appender waits for space allocation.</td>
        <td>kafka.producer:type=producer-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>bufferpool-wait-ratio</td>
        <td>The fraction of time an appender waits for space allocation.</td>
        <td>kafka.producer:type=producer-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>bufferpool-wait-time-total</td>
        <td><b>*Deprecated*</b> The total time an appender waits for space allocation in nanoseconds. Replacement is <code>bufferpool-wait-time-ns-total</code></td>
        <td>kafka.producer:type=producer-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>bufferpool-wait-time-ns-total</td>
        <td>The total time an appender waits for space allocation in nanoseconds.</td>
        <td>kafka.producer:type=producer-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>flush-time-ns-total</td>
        <td>The total time the Producer spent in Producer.flush in nanoseconds.</td>
        <td>kafka.producer:type=producer-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>txn-init-time-ns-total</td>
        <td>The total time the Producer spent initializing transactions in nanoseconds (for EOS).</td>
        <td>kafka.producer:type=producer-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>txn-begin-time-ns-total</td>
        <td>The total time the Producer spent in beginTransaction in nanoseconds (for EOS).</td>
        <td>kafka.producer:type=producer-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>txn-send-offsets-time-ns-total</td>
        <td>The total time the Producer spent sending offsets to transactions in nanoseconds (for EOS).</td>
        <td>kafka.producer:type=producer-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>txn-commit-time-ns-total</td>
        <td>The total time the Producer spent committing transactions in nanoseconds (for EOS).</td>
        <td>kafka.producer:type=producer-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>txn-abort-time-ns-total</td>
        <td>The total time the Producer spent aborting transactions in nanoseconds (for EOS).</td>
        <td>kafka.producer:type=producer-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>metadata-wait-time-ns-total</td>
        <td>the total time in nanoseconds that has spent waiting for metadata from the Kafka broker</td>
        <td>kafka.producer:type=producer-metrics,client-id=([-.\w]+)</td>
      </tr>
  </tbody></table>

  <h5 class="anchor-heading"><a id="producer_sender_monitoring" class="anchor-link"></a><a href="#producer_sender_monitoring">Producer Sender Metrics</a></h5>

  <!--#include virtual="generated/producer_metrics.html" -->


  <h4 class="anchor-heading"><a id="consumer_monitoring" class="anchor-link"></a><a href="#consumer_monitoring">Consumer monitoring</a></h4>

  The following metrics are available on consumer instances.

  <table class="data-table">
    <tbody>
      <tr>
        <th>Metric/Attribute name</th>
        <th>Description</th>
        <th>Mbean name</th>
      </tr>
      <tr>
        <td>time-between-poll-avg</td>
        <td>The average delay between invocations of poll().</td>
        <td>kafka.consumer:type=consumer-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>time-between-poll-max</td>
        <td>The max delay between invocations of poll().</td>
        <td>kafka.consumer:type=consumer-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>last-poll-seconds-ago</td>
        <td>The number of seconds since the last poll() invocation.</td>
        <td>kafka.consumer:type=consumer-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>poll-idle-ratio-avg</td>
        <td>The average fraction of time the consumer's poll() is idle as opposed to waiting for the user code to process records.</td>
        <td>kafka.consumer:type=consumer-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>committed-time-ns-total</td>
        <td>The total time the Consumer spent in committed in nanoseconds.</td>
        <td>kafka.consumer:type=consumer-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>commit-sync-time-ns-total</td>
        <td>The total time the Consumer spent committing offsets in nanoseconds (for AOS).</td>
        <td>kafka.consumer:type=consumer-metrics,client-id=([-.\w]+)</td>
      </tr>
    </tbody>
  </table>

  <h5 class="anchor-heading"><a id="consumer_group_monitoring" class="anchor-link"></a><a href="#consumer_group_monitoring">Consumer Group Metrics</a></h5>
  <table class="data-table">
    <tbody>
      <tr>
        <th>Metric/Attribute name</th>
        <th>Description</th>
        <th>Mbean name</th>
      </tr>
      <tr>
        <td>commit-latency-avg</td>
        <td>The average time taken for a commit request</td>
        <td>kafka.consumer:type=consumer-coordinator-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>commit-latency-max</td>
        <td>The max time taken for a commit request</td>
        <td>kafka.consumer:type=consumer-coordinator-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>commit-rate</td>
        <td>The number of commit calls per second</td>
        <td>kafka.consumer:type=consumer-coordinator-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>commit-total</td>
        <td>The total number of commit calls</td>
        <td>kafka.consumer:type=consumer-coordinator-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>assigned-partitions</td>
        <td>The number of partitions currently assigned to this consumer</td>
        <td>kafka.consumer:type=consumer-coordinator-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>heartbeat-response-time-max</td>
        <td>The max time taken to receive a response to a heartbeat request</td>
        <td>kafka.consumer:type=consumer-coordinator-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>heartbeat-rate</td>
        <td>The average number of heartbeats per second</td>
        <td>kafka.consumer:type=consumer-coordinator-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>heartbeat-total</td>
        <td>The total number of heartbeats</td>
        <td>kafka.consumer:type=consumer-coordinator-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>join-time-avg</td>
        <td>The average time taken for a group rejoin</td>
        <td>kafka.consumer:type=consumer-coordinator-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>join-time-max</td>
        <td>The max time taken for a group rejoin</td>
        <td>kafka.consumer:type=consumer-coordinator-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>join-rate</td>
        <td>The number of group joins per second</td>
        <td>kafka.consumer:type=consumer-coordinator-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>join-total</td>
        <td>The total number of group joins</td>
        <td>kafka.consumer:type=consumer-coordinator-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>sync-time-avg</td>
        <td>The average time taken for a group sync</td>
        <td>kafka.consumer:type=consumer-coordinator-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>sync-time-max</td>
        <td>The max time taken for a group sync</td>
        <td>kafka.consumer:type=consumer-coordinator-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>sync-rate</td>
        <td>The number of group syncs per second</td>
        <td>kafka.consumer:type=consumer-coordinator-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>sync-total</td>
        <td>The total number of group syncs</td>
        <td>kafka.consumer:type=consumer-coordinator-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>rebalance-latency-avg</td>
        <td>The average time taken for a group rebalance</td>
        <td>kafka.consumer:type=consumer-coordinator-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>rebalance-latency-max</td>
        <td>The max time taken for a group rebalance</td>
        <td>kafka.consumer:type=consumer-coordinator-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>rebalance-latency-total</td>
        <td>The total time taken for group rebalances so far</td>
        <td>kafka.consumer:type=consumer-coordinator-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>rebalance-total</td>
        <td>The total number of group rebalances participated</td>
        <td>kafka.consumer:type=consumer-coordinator-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>rebalance-rate-per-hour</td>
        <td>The number of group rebalance participated per hour</td>
        <td>kafka.consumer:type=consumer-coordinator-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>failed-rebalance-total</td>
        <td>The total number of failed group rebalances</td>
        <td>kafka.consumer:type=consumer-coordinator-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>failed-rebalance-rate-per-hour</td>
        <td>The number of failed group rebalance event per hour</td>
        <td>kafka.consumer:type=consumer-coordinator-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>last-rebalance-seconds-ago</td>
        <td>The number of seconds since the last rebalance event</td>
        <td>kafka.consumer:type=consumer-coordinator-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>last-heartbeat-seconds-ago</td>
        <td>The number of seconds since the last controller heartbeat</td>
        <td>kafka.consumer:type=consumer-coordinator-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>partitions-revoked-latency-avg</td>
        <td>The average time taken by the on-partitions-revoked rebalance listener callback</td>
        <td>kafka.consumer:type=consumer-coordinator-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>partitions-revoked-latency-max</td>
        <td>The max time taken by the on-partitions-revoked rebalance listener callback</td>
        <td>kafka.consumer:type=consumer-coordinator-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>partitions-assigned-latency-avg</td>
        <td>The average time taken by the on-partitions-assigned rebalance listener callback</td>
        <td>kafka.consumer:type=consumer-coordinator-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>partitions-assigned-latency-max</td>
        <td>The max time taken by the on-partitions-assigned rebalance listener callback</td>
        <td>kafka.consumer:type=consumer-coordinator-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>partitions-lost-latency-avg</td>
        <td>The average time taken by the on-partitions-lost rebalance listener callback</td>
        <td>kafka.consumer:type=consumer-coordinator-metrics,client-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>partitions-lost-latency-max</td>
        <td>The max time taken by the on-partitions-lost rebalance listener callback</td>
        <td>kafka.consumer:type=consumer-coordinator-metrics,client-id=([-.\w]+)</td>
      </tr>
    </tbody>
  </table>

  <h5 class="anchor-heading"><a id="consumer_fetch_monitoring" class="anchor-link"></a><a href="#consumer_fetch_monitoring">Consumer Fetch Metrics</a></h5>

  <!--#include virtual="generated/consumer_metrics.html" -->

  <h4 class="anchor-heading"><a id="connect_monitoring" class="anchor-link"></a><a href="#connect_monitoring">Connect Monitoring</a></h4>

  A Connect worker process contains all the producer and consumer metrics as well as metrics specific to Connect.
  The worker process itself has a number of metrics, while each connector and task have additional metrics.

  <!--#include virtual="generated/connect_metrics.html" -->

  <h4 class="anchor-heading"><a id="kafka_streams_monitoring" class="anchor-link"></a><a href="#kafka_streams_monitoring">Streams Monitoring</a></h4>

  A Kafka Streams instance contains all the producer and consumer metrics as well as additional metrics specific to Streams.
  The metrics have three recording levels: <code>info</code>, <code>debug</code>, and <code>trace</code>.

  <p>
    Note that the metrics have a 4-layer hierarchy. At the top level there are client-level metrics for each started
    Kafka Streams client. Each client has stream threads, with their own metrics. Each stream thread has tasks, with their
    own metrics. Each task has a number of processor nodes, with their own metrics. Each task also has a number of state stores
    and record caches, all with their own metrics.
  </p>

  Use the following configuration option to specify which metrics
  you want collected:

<pre><code>metrics.recording.level="info"</code></pre>

<h5 class="anchor-heading"><a id="kafka_streams_client_monitoring" class="anchor-link"></a><a href="#kafka_streams_client_monitoring">Client Metrics</a></h5>
All of the following metrics have a recording level of <code>info</code>:
<table class="data-table">
  <tbody>
  <tr>
    <th>Metric/Attribute name</th>
    <th>Description</th>
    <th>Mbean name</th>
  </tr>
  <tr>
    <td>version</td>
    <td>The version of the Kafka Streams client.</td>
    <td>kafka.streams:type=stream-metrics,client-id=([-.\w]+)</td>
  </tr>
  <tr>
    <td>commit-id</td>
    <td>The version control commit ID of the Kafka Streams client.</td>
    <td>kafka.streams:type=stream-metrics,client-id=([-.\w]+)</td>
  </tr>
  <tr>
    <td>application-id</td>
    <td>The application ID of the Kafka Streams client.</td>
    <td>kafka.streams:type=stream-metrics,client-id=([-.\w]+)</td>
  </tr>
  <tr>
    <td>topology-description</td>
    <td>The description of the topology executed in the Kafka Streams client.</td>
    <td>kafka.streams:type=stream-metrics,client-id=([-.\w]+)</td>
  </tr>
  <tr>
    <td>state</td>
    <td>The state of the Kafka Streams client.</td>
    <td>kafka.streams:type=stream-metrics,client-id=([-.\w]+)</td>
  </tr>
  <tr>
      <td>failed-stream-threads</td>
      <td>The number of failed stream threads since the start of the Kafka Streams client.</td>
      <td>kafka.streams:type=stream-metrics,client-id=([-.\w]+)</td>
  </tr>
  </tbody>
</table>

<h5 class="anchor-heading"><a id="kafka_streams_thread_monitoring" class="anchor-link"></a><a href="#kafka_streams_thread_monitoring">Thread Metrics</a></h5>
All of the following metrics have a recording level of <code>info</code>:
<table class="data-table">
    <tbody>
      <tr>
        <th>Metric/Attribute name</th>
        <th>Description</th>
        <th>Mbean name</th>
      </tr>
      <tr>
        <td>commit-latency-avg</td>
        <td>The average execution time in ms, for committing, across all running tasks of this thread.</td>
        <td>kafka.streams:type=stream-thread-metrics,thread-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>commit-latency-max</td>
        <td>The maximum execution time in ms, for committing, across all running tasks of this thread.</td>
        <td>kafka.streams:type=stream-thread-metrics,thread-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>poll-latency-avg</td>
        <td>The average execution time in ms, for consumer polling.</td>
        <td>kafka.streams:type=stream-thread-metrics,thread-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>poll-latency-max</td>
        <td>The maximum execution time in ms, for consumer polling.</td>
        <td>kafka.streams:type=stream-thread-metrics,thread-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>process-latency-avg</td>
        <td>The average execution time in ms, for processing.</td>
        <td>kafka.streams:type=stream-thread-metrics,thread-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>process-latency-max</td>
        <td>The maximum execution time in ms, for processing.</td>
        <td>kafka.streams:type=stream-thread-metrics,thread-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>punctuate-latency-avg</td>
        <td>The average execution time in ms, for punctuating.</td>
        <td>kafka.streams:type=stream-thread-metrics,thread-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>punctuate-latency-max</td>
        <td>The maximum execution time in ms, for punctuating.</td>
        <td>kafka.streams:type=stream-thread-metrics,thread-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>commit-rate</td>
        <td>The average number of commits per second.</td>
        <td>kafka.streams:type=stream-thread-metrics,thread-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>commit-total</td>
        <td>The total number of commit calls.</td>
        <td>kafka.streams:type=stream-thread-metrics,thread-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>poll-rate</td>
        <td>The average number of consumer poll calls per second.</td>
        <td>kafka.streams:type=stream-thread-metrics,thread-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>poll-total</td>
        <td>The total number of consumer poll calls.</td>
        <td>kafka.streams:type=stream-thread-metrics,thread-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>process-rate</td>
        <td>The average number of processed records per second.</td>
        <td>kafka.streams:type=stream-thread-metrics,thread-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>process-total</td>
        <td>The total number of processed records.</td>
        <td>kafka.streams:type=stream-thread-metrics,thread-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>punctuate-rate</td>
        <td>The average number of punctuate calls per second.</td>
        <td>kafka.streams:type=stream-thread-metrics,thread-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>punctuate-total</td>
        <td>The total number of punctuate calls.</td>
        <td>kafka.streams:type=stream-thread-metrics,thread-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>task-created-rate</td>
        <td>The average number of tasks created per second.</td>
        <td>kafka.streams:type=stream-thread-metrics,thread-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>task-created-total</td>
        <td>The total number of tasks created.</td>
        <td>kafka.streams:type=stream-thread-metrics,thread-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>task-closed-rate</td>
        <td>The average number of tasks closed per second.</td>
        <td>kafka.streams:type=stream-thread-metrics,thread-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>task-closed-total</td>
        <td>The total number of tasks closed.</td>
        <td>kafka.streams:type=stream-thread-metrics,thread-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>blocked-time-ns-total</td>
        <td>The total time the thread spent blocked on kafka.</td>
        <td>kafka.streams:type=stream-thread-metrics,thread-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>thread-start-time</td>
        <td>The time that the thread was started.</td>
        <td>kafka.streams:type=stream-thread-metrics,thread-id=([-.\w]+)</td>
      </tr>
 </tbody>
</table>

<h5 class="anchor-heading"><a id="kafka_streams_task_monitoring" class="anchor-link"></a><a href="#kafka_streams_task_monitoring">Task Metrics</a></h5>
All of the following metrics have a recording level of <code>debug</code>, except for the dropped-records-* and
active-process-ratio metrics which have a recording level of <code>info</code>:
 <table class="data-table">
      <tbody>
      <tr>
        <th>Metric/Attribute name</th>
        <th>Description</th>
        <th>Mbean name</th>
      </tr>
      <tr>
        <td>process-latency-avg</td>
        <td>The average execution time in ns, for processing.</td>
        <td>kafka.streams:type=stream-task-metrics,thread-id=([-.\w]+),task-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>process-latency-max</td>
        <td>The maximum execution time in ns, for processing.</td>
        <td>kafka.streams:type=stream-task-metrics,thread-id=([-.\w]+),task-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>process-rate</td>
        <td>The average number of processed records per second across all source processor nodes of this task.</td>
        <td>kafka.streams:type=stream-task-metrics,thread-id=([-.\w]+),task-id=([-.\w]+)</td>
       </tr>
      <tr>
        <td>process-total</td>
        <td>The total number of processed records across all source processor nodes of this task.</td>
        <td>kafka.streams:type=stream-task-metrics,thread-id=([-.\w]+),task-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>record-lateness-avg</td>
        <td>The average observed lateness of records (stream time - record timestamp).</td>
        <td>kafka.streams:type=stream-task-metrics,thread-id=([-.\w]+),task-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>record-lateness-max</td>
        <td>The max observed lateness of records (stream time - record timestamp).</td>
        <td>kafka.streams:type=stream-task-metrics,thread-id=([-.\w]+),task-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>enforced-processing-rate</td>
        <td>The average number of enforced processings per second.</td>
        <td>kafka.streams:type=stream-task-metrics,thread-id=([-.\w]+),task-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>enforced-processing-total</td>
        <td>The total number enforced processings.</td>
        <td>kafka.streams:type=stream-task-metrics,thread-id=([-.\w]+),task-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>dropped-records-rate</td>
        <td>The average number of records dropped within this task.</td>
        <td>kafka.streams:type=stream-task-metrics,thread-id=([-.\w]+),task-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>dropped-records-total</td>
        <td>The total number of records dropped within this task.</td>
        <td>kafka.streams:type=stream-task-metrics,thread-id=([-.\w]+),task-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>active-process-ratio</td>
        <td>The fraction of time the stream thread spent on processing this task among all assigned active tasks.</td>
        <td>kafka.streams:type=stream-task-metrics,thread-id=([-.\w]+),task-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>input-buffer-bytes-total</td>
        <td>The total number of bytes accumulated by this task,</td>
        <td>kafka.streams:type=stream-task-metrics,thread-id=([-.\w]+),task-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>cache-size-bytes-total</td>
        <td>The cache size in bytes accumulated by this task.</td>
        <td>kafka.streams:type=stream-task-metrics,thread-id=([-.\w]+),task-id=([-.\w]+)</td>
      </tr>
 </tbody>
</table>

 <h5 class="anchor-heading"><a id="kafka_streams_node_monitoring" class="anchor-link"></a><a href="#kafka_streams_node_monitoring">Processor Node Metrics</a></h5>
 The following metrics are only available on certain types of nodes, i.e., the process-* metrics are only available for
 source processor nodes, the <code>suppression-emit-*</code> metrics are only available for suppression operation nodes,
 <code>emit-final-*</code> metrics are only available for windowed aggregations nodes, and the
 <code>record-e2e-latency-*</code> metrics are only available for source processor nodes and terminal nodes (nodes without successor
 nodes).
 All of the metrics have a recording level of <code>debug</code>, except for the <code>record-e2e-latency-*</code> metrics which have
 a recording level of <code>info</code>:
 <table class="data-table">
      <tbody>
      <tr>
        <th>Metric/Attribute name</th>
        <th>Description</th>
        <th>Mbean name</th>
      </tr>
      <tr>
        <td>bytes-consumed-total</td>
        <td>The total number of bytes consumed by a source processor node.</td>
        <td>kafka.streams:type=stream-topic-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),processor-node-id=([-.\w]+),topic=([-.\w]+)</td>
      </tr>
      <tr>
        <td>bytes-produced-total</td>
        <td>The total number of bytes produced by a sink processor node.</td>
        <td>kafka.streams:type=stream-topic-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),processor-node-id=([-.\w]+),topic=([-.\w]+)</td>
      </tr>
      <tr>
        <td>process-rate</td>
        <td>The average number of records processed by a source processor node per second.</td>
        <td>kafka.streams:type=stream-processor-node-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),processor-node-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>process-total</td>
        <td>The total number of records processed by a source processor node per second.</td>
        <td>kafka.streams:type=stream-processor-node-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),processor-node-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>suppression-emit-rate</td>
        <td>The rate at which records that have been emitted downstream from suppression operation nodes.</td>
        <td>kafka.streams:type=stream-processor-node-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),processor-node-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>suppression-emit-total</td>
        <td>The total number of records that have been emitted downstream from suppression operation nodes.</td>
        <td>kafka.streams:type=stream-processor-node-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),processor-node-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>emit-final-latency-max</td>
        <td>The max latency to emit final records when a record could be emitted.</td>
        <td>kafka.streams:type=stream-processor-node-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),processor-node-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>emit-final-latency-avg</td>
        <td>The avg latency to emit final records when a record could be emitted.</td>
        <td>kafka.streams:type=stream-processor-node-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),processor-node-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>emit-final-records-rate</td>
        <td>The rate of records emitted when records could be emitted.</td>
        <td>kafka.streams:type=stream-processor-node-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),processor-node-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>emit-final-records-total</td>
        <td>The total number of records emitted.</td>
        <td>kafka.streams:type=stream-processor-node-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),processor-node-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>record-e2e-latency-avg</td>
        <td>The average end-to-end latency of a record, measured by comparing the record timestamp with the system time when it has been fully processed by the node.</td>
        <td>kafka.streams:type=stream-processor-node-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),processor-node-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>record-e2e-latency-max</td>
        <td>The maximum end-to-end latency of a record, measured by comparing the record timestamp with the system time when it has been fully processed by the node.</td>
        <td>kafka.streams:type=stream-processor-node-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),processor-node-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>record-e2e-latency-min</td>
        <td>The minimum end-to-end latency of a record, measured by comparing the record timestamp with the system time when it has been fully processed by the node.</td>
        <td>kafka.streams:type=stream-processor-node-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),processor-node-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>records-consumed-total</td>
        <td>The total number of records consumed by a source processor node.</td>
        <td>kafka.streams:type=stream-topic-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),processor-node-id=([-.\w]+),topic=([-.\w]+)</td>
      </tr>
      <tr>
        <td>records-produced-total</td>
        <td>The total number of records produced by a sink processor node.</td>
        <td>kafka.streams:type=stream-topic-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),processor-node-id=([-.\w]+),topic=([-.\w]+)</td>
      </tr>
      </tbody>
 </table>

 <h5 class="anchor-heading"><a id="kafka_streams_store_monitoring" class="anchor-link"></a><a href="#kafka_streams_store_monitoring">State Store Metrics</a></h5>
All the following metrics have a recording level of <code>debug</code>, except for the <code>record-e2e-latency-*</code>
metrics which have a recording level <code>trace</code> and <code>num-open-iterators</code> which has recording level
<code>info</code>.
Note that the <code>store-scope</code> value is specified in <code>StoreSupplier#metricsScope()</code> for user's
customized state stores; for built-in state stores, currently we have:
  <ul>
    <li><code>in-memory-state</code></li>
    <li><code>in-memory-lru-state</code></li>
    <li><code>in-memory-window-state</code></li>
    <li><code>in-memory-suppression</code> (for suppression buffers)</li>
    <li><code>rocksdb-state</code> (for RocksDB backed key-value store)</li>
    <li><code>rocksdb-window-state</code> (for RocksDB backed window store)</li>
    <li><code>rocksdb-session-state</code> (for RocksDB backed session store)</li>
  </ul>
  Metrics suppression-buffer-size-avg, suppression-buffer-size-max, suppression-buffer-count-avg, and suppression-buffer-count-max
  are only available for suppression buffers. All other metrics are not available for suppression buffers.

  <table class="data-table">
      <tbody>
      <tr>
        <th>Metric/Attribute name</th>
        <th>Description</th>
        <th>Mbean name</th>
      </tr>
      <tr>
        <td>put-latency-avg</td>
        <td>The average put execution time in ns.</td>
        <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>put-latency-max</td>
        <td>The maximum put execution time in ns.</td>
        <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>put-if-absent-latency-avg</td>
        <td>The average put-if-absent execution time in ns.</td>
        <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>put-if-absent-latency-max</td>
        <td>The maximum put-if-absent execution time in ns.</td>
        <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>get-latency-avg</td>
        <td>The average get execution time in ns.</td>
        <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>get-latency-max</td>
        <td>The maximum get execution time in ns.</td>
        <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>delete-latency-avg</td>
        <td>The average delete execution time in ns.</td>
        <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>delete-latency-max</td>
        <td>The maximum delete execution time in ns.</td>
        <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>put-all-latency-avg</td>
        <td>The average put-all execution time in ns.</td>
        <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>put-all-latency-max</td>
        <td>The maximum put-all execution time in ns.</td>
        <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>all-latency-avg</td>
        <td>The average execution time in ns, from iterator create to close time.</td>
        <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>all-latency-max, from iterator create to close time.</td>
        <td>The maximum all operation execution time in ns.</td>
        <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>range-latency-avg, from iterator create to close time.</td>
        <td>The average range execution time in ns, from iterator create to close time.</td>
        <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>range-latency-max, from iterator create to close time.</td>
        <td>The maximum range execution time in ns.</td>
        <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>prefix-scan-latency-avg</td>
        <td>The average prefix-scan execution time in ns, from iterator create to close time.</td>
        <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>prefix-scan-latency-max</td>
        <td>The maximum prefix-scan execution time in ns, from iterator create to close time.</td>
        <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
      </tr>
       <tr>
        <td>flush-latency-avg</td>
        <td>The average flush execution time in ns.</td>
        <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>flush-latency-max</td>
        <td>The maximum flush execution time in ns.</td>
        <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>restore-latency-avg</td>
        <td>The average restore execution time in ns.</td>
        <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>restore-latency-max</td>
        <td>The maximum restore execution time in ns.</td>
        <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>put-rate</td>
        <td>The average put rate for this store.</td>
        <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>put-if-absent-rate</td>
        <td>The average put-if-absent rate for this store.</td>
        <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>get-rate</td>
        <td>The average get rate for this store.</td>
        <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>delete-rate</td>
        <td>The average delete rate for this store.</td>
        <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>put-all-rate</td>
        <td>The average put-all rate for this store.</td>
        <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>all-rate</td>
        <td>The average all operation rate for this store.</td>
        <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>range-rate</td>
        <td>The average range rate for this store.</td>
        <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>prefix-scan-rate</td>
        <td>The average prefix-scan rate for this store.</td>
        <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>flush-rate</td>
        <td>The average flush rate for this store.</td>
        <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>restore-rate</td>
        <td>The average restore rate for this store.</td>
        <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>suppression-buffer-size-avg</td>
        <td>The average total size, in bytes, of the buffered data over the sampling window.</td>
        <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),in-memory-suppression-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>suppression-buffer-size-max</td>
        <td>The maximum total size, in bytes, of the buffered data over the sampling window.</td>
        <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),in-memory-suppression-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>suppression-buffer-count-avg</td>
        <td>The average number of records buffered over the sampling window.</td>
        <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),in-memory-suppression-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>suppression-buffer-count-max</td>
        <td>The maximum number of records buffered over the sampling window.</td>
        <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),in-memory-suppression-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>record-e2e-latency-avg</td>
        <td>The average end-to-end latency of a record, measured by comparing the record timestamp with the system time when it has been fully processed by the node.</td>
        <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>record-e2e-latency-max</td>
        <td>The maximum end-to-end latency of a record, measured by comparing the record timestamp with the system time when it has been fully processed by the node.</td>
        <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>record-e2e-latency-min</td>
        <td>The minimum end-to-end latency of a record, measured by comparing the record timestamp with the system time when it has been fully processed by the node.</td>
        <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>num-open-iterators</td>
        <td>The current number of iterators on the store that have been created, but not yet closed.</td>
        <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>iterator-duration-avg</td>
        <td>The average time spent between creating an iterator and closing it, in nanoseconds.</td>
        <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>iterator-duration-max</td>
        <td>The maximum time spent between creating an iterator and closing it, in nanoseconds.</td>
        <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>oldest-iterator-open-since-ms</td>
        <td>The UNIX timestamp the oldest still open iterator was created, in milliseconds.</td>
        <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
      </tr>
    </tbody>
 </table>

  <h5 class="anchor-heading"><a id="kafka_streams_rocksdb_monitoring" class="anchor-link"></a><a href="#kafka_streams_rocksdb_monitoring">RocksDB Metrics</a></h5>
  RocksDB metrics are grouped into statistics-based metrics and properties-based metrics.
  The former are recorded from statistics that a RocksDB state store collects whereas the latter are recorded from
  properties that RocksDB exposes.
  Statistics collected by RocksDB provide cumulative measurements over time, e.g. bytes written to the state store.
  Properties exposed by RocksDB provide current measurements, e.g., the amount of memory currently used.
  Note that the <code>store-scope</code> for built-in RocksDB state stores are currently the following:
  <ul>
    <li><code>rocksdb-state</code> (for RocksDB backed key-value store)</li>
    <li><code>rocksdb-window-state</code> (for RocksDB backed window store)</li>
    <li><code>rocksdb-session-state</code> (for RocksDB backed session store)</li>
  </ul>

  <strong>RocksDB Statistics-based Metrics:</strong>
  All of the following statistics-based metrics have a recording level of <code>debug</code> because collecting
  statistics in <a href="https://github.com/facebook/rocksdb/wiki/Statistics#stats-level-and-performance-costs">RocksDB
  may have an impact on performance</a>.
  Statistics-based metrics are collected every minute from the RocksDB state stores.
  If a state store consists of multiple RocksDB instances, as is the case for WindowStores and SessionStores,
  each metric reports an aggregation over the RocksDB instances of the state store.

  <table class="data-table">
    <tbody>
    <tr>
      <th>Metric/Attribute name</th>
      <th>Description</th>
      <th>Mbean name</th>
    </tr>
    <tr>
      <td>bytes-written-rate</td>
      <td>The average number of bytes written per second to the RocksDB state store.</td>
      <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
    </tr>
    <tr>
      <td>bytes-written-total</td>
      <td>The total number of bytes written to the RocksDB state store.</td>
      <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
    </tr>
    <tr>
      <td>bytes-read-rate</td>
      <td>The average number of bytes read per second from the RocksDB state store.</td>
      <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
    </tr>
    <tr>
      <td>bytes-read-total</td>
      <td>The total number of bytes read from the RocksDB state store.</td>
      <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
    </tr>
    <tr>
      <td>memtable-bytes-flushed-rate</td>
      <td>The average number of bytes flushed per second from the memtable to disk.</td>
      <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
    </tr>
    <tr>
      <td>memtable-bytes-flushed-total</td>
      <td>The total number of bytes flushed from the memtable to disk.</td>
      <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
    </tr>
    <tr>
      <td>memtable-hit-ratio</td>
      <td>The ratio of memtable hits relative to all lookups to the memtable.</td>
      <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
    </tr>
    <tr>
      <td>memtable-flush-time-avg</td>
      <td>The average duration of memtable flushes to disc in ms.</td>
      <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
    </tr>
    <tr>
      <td>memtable-flush-time-min</td>
      <td>The minimum duration of memtable flushes to disc in ms.</td>
      <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
    </tr>
    <tr>
      <td>memtable-flush-time-max</td>
      <td>The maximum duration of memtable flushes to disc in ms.</td>
      <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
    </tr>
    <tr>
      <td>block-cache-data-hit-ratio</td>
      <td>The ratio of block cache hits for data blocks relative to all lookups for data blocks to the block cache.</td>
      <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
    </tr>
    <tr>
      <td>block-cache-index-hit-ratio</td>
      <td>The ratio of block cache hits for index blocks relative to all lookups for index blocks to the block cache.</td>
      <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
    </tr>
    <tr>
      <td>block-cache-filter-hit-ratio</td>
      <td>The ratio of block cache hits for filter blocks relative to all lookups for filter blocks to the block cache.</td>
      <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
    </tr>
    <tr>
      <td>write-stall-duration-avg</td>
      <td>The average duration of write stalls in ms.</td>
      <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
    </tr>
    <tr>
      <td>write-stall-duration-total</td>
      <td>The total duration of write stalls in ms.</td>
      <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
    </tr>
    <tr>
      <td>bytes-read-compaction-rate</td>
      <td>The average number of bytes read per second during compaction.</td>
      <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
    </tr>
    <tr>
      <td>bytes-written-compaction-rate</td>
      <td>The average number of bytes written per second during compaction.</td>
      <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
    </tr>
    <tr>
      <td>compaction-time-avg</td>
      <td>The average duration of disc compactions in ms.</td>
      <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
    </tr>
    <tr>
      <td>compaction-time-min</td>
      <td>The minimum duration of disc compactions in ms.</td>
      <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
    </tr>
    <tr>
      <td>compaction-time-max</td>
      <td>The maximum duration of disc compactions in ms.</td>
      <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
    </tr>
    <tr>
      <td>number-open-files</td>
      <td>The number of current open files.</td>
      <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
    </tr>
    <tr>
      <td>number-file-errors-total</td>
      <td>The total number of file errors occurred.</td>
      <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
    </tr>
    </tbody>
  </table>

  <strong>RocksDB Properties-based Metrics:</strong>
  All of the following properties-based metrics have a recording level of <code>info</code> and are recorded when the
  metrics are accessed.
  If a state store consists of multiple RocksDB instances, as is the case for WindowStores and SessionStores,
  each metric reports the sum over all the RocksDB instances of the state store, except for the block cache metrics
  <code>block-cache-*</code>. The block cache metrics report the sum over all RocksDB instances if each instance uses its
  own block cache, and they report the recorded value from only one instance if a single block cache is shared
  among all instances.

  <table class="data-table">
    <tbody>
    <tr>
      <th>Metric/Attribute name</th>
      <th>Description</th>
      <th>Mbean name</th>
    </tr>
    <tr>
      <td>num-immutable-mem-table</td>
      <td>The number of immutable memtables that have not yet been flushed.</td>
      <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
    </tr>
    <tr>
      <td>cur-size-active-mem-table</td>
      <td>The approximate size of the active memtable in bytes.</td>
      <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
    </tr>
    <tr>
      <td>cur-size-all-mem-tables</td>
      <td>The approximate size of active and unflushed immutable memtables in bytes.</td>
      <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
    </tr>
    <tr>
      <td>size-all-mem-tables</td>
      <td>The approximate size of active, unflushed immutable, and pinned immutable memtables in bytes.</td>
      <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
    </tr>
    <tr>
      <td>num-entries-active-mem-table</td>
      <td>The number of entries in the active memtable.</td>
      <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
    </tr>
    <tr>
      <td>num-entries-imm-mem-tables</td>
      <td>The number of entries in the unflushed immutable memtables.</td>
      <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
    </tr>
    <tr>
      <td>num-deletes-active-mem-table</td>
      <td>The number of delete entries in the active memtable.</td>
      <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
    </tr>
    <tr>
      <td>num-deletes-imm-mem-tables</td>
      <td>The number of delete entries in the unflushed immutable memtables.</td>
      <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
    </tr>
    <tr>
      <td>mem-table-flush-pending</td>
      <td>This metric reports 1 if a memtable flush is pending, otherwise it reports 0.</td>
      <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
    </tr>
    <tr>
      <td>num-running-flushes</td>
      <td>The number of currently running flushes.</td>
      <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
    </tr>
    <tr>
      <td>compaction-pending</td>
      <td>This metric reports 1 if at least one compaction is pending, otherwise it reports 0.</td>
      <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
    </tr>
    <tr>
      <td>num-running-compactions</td>
      <td>The number of currently running compactions.</td>
      <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
    </tr>
    <tr>
      <td>estimate-pending-compaction-bytes</td>
      <td>The estimated total number of bytes a compaction needs to rewrite on disk to get all levels down to under
        target size (only valid for level compaction).</td>
      <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
    </tr>
    <tr>
      <td>total-sst-files-size</td>
      <td>The total size in bytes of all SST files.</td>
      <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
    </tr>
    <tr>
      <td>live-sst-files-size</td>
      <td>The total size in bytes of all SST files that belong to the latest LSM tree.</td>
      <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
    </tr>
    <tr>
      <td>num-live-versions</td>
      <td>Number of live versions of the LSM tree.</td>
      <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
    </tr>
    <tr>
      <td>block-cache-capacity</td>
      <td>The capacity of the block cache in bytes.</td>
      <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
    </tr>
    <tr>
      <td>block-cache-usage</td>
      <td>The memory size of the entries residing in block cache in bytes.</td>
      <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
    </tr>
    <tr>
      <td>block-cache-pinned-usage</td>
      <td>The memory size for the entries being pinned in the block cache in bytes.</td>
      <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
    </tr>
    <tr>
      <td>estimate-num-keys</td>
      <td>The estimated number of keys in the active and unflushed immutable memtables and storage.</td>
      <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
    </tr>
    <tr>
      <td>estimate-table-readers-mem</td>
      <td>The estimated memory in bytes used for reading SST tables, excluding memory used in block cache.</td>
      <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
    </tr>
    <tr>
      <td>background-errors</td>
      <td>The total number of background errors.</td>
      <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
    </tr>
    </tbody>
  </table>

  <h5 class="anchor-heading"><a id="kafka_streams_cache_monitoring" class="anchor-link"></a><a href="#kafka_streams_cache_monitoring">Record Cache Metrics</a></h5>
  All of the following metrics have a recording level of <code>debug</code>:

  <table class="data-table">
      <tbody>
      <tr>
        <th>Metric/Attribute name</th>
        <th>Description</th>
        <th>Mbean name</th>
      </tr>
      <tr>
        <td>hit-ratio-avg</td>
        <td>The average cache hit ratio defined as the ratio of cache read hits over the total cache read requests.</td>
        <td>kafka.streams:type=stream-record-cache-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),record-cache-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>hit-ratio-min</td>
        <td>The minimum cache hit ratio.</td>
        <td>kafka.streams:type=stream-record-cache-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),record-cache-id=([-.\w]+)</td>
      </tr>
      <tr>
        <td>hit-ratio-max</td>
        <td>The maximum cache hit ratio.</td>
        <td>kafka.streams:type=stream-record-cache-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),record-cache-id=([-.\w]+)</td>
      </tr>
    </tbody>
 </table>

  <h4 class="anchor-heading"><a id="others_monitoring" class="anchor-link"></a><a href="#others_monitoring">Others</a></h4>

  We recommend monitoring GC time and other stats and various server stats such as CPU utilization, I/O service time, etc.

  On the client side, we recommend monitoring the message/byte rate (global and per topic), request rate/size/time, and on the consumer side, max lag in messages among all partitions and min fetch request rate. For a consumer to keep up, max lag needs to be less than a threshold and min fetch rate needs to be larger than 0.

  <h3 class="anchor-heading"><a id="zk" class="anchor-link"></a><a href="#zk">6.9 ZooKeeper</a></h3>

  <h4 class="anchor-heading"><a id="zkversion" class="anchor-link"></a><a href="#zkversion">Stable version</a></h4>
  The current stable branch is 3.8. Kafka is regularly updated to include the latest release in the 3.8 series.
  
  <h4 class="anchor-heading"><a id="zk_depr" class="anchor-link"></a><a href="#zk_depr">ZooKeeper Deprecation</a></h4>
  <p>With the release of Apache Kafka 3.5, Zookeeper is now marked deprecated. Removal of ZooKeeper is planned in the next major release of Apache Kafka (version 4.0),
     which is scheduled to happen no sooner than April 2024. During the deprecation phase, ZooKeeper is still supported for metadata management of Kafka clusters,
     but it is not recommended for new deployments. There is a small subset of features that remain to be implemented in KRaft
     see <a href="#kraft_missing">current missing features</a> for more information.</p>

    <h5 class="anchor-heading"><a id="zk_depr_migration" class="anchor-link"></a><a href="#zk_drep_migration">Migration</a></h5>
    <p>Users are recommended to begin planning for migration to KRaft and also begin testing to provide any feedback. Refer to <a href="#kraft_zk_migration">ZooKeeper to KRaft Migration</a> for details on how to perform a live migration from ZooKeeper to KRaft and current limitations.</p>
	
    <h5 class="anchor-heading"><a id="zk_depr_3xsupport" class="anchor-link"></a><a href="#zk_depr_3xsupport">3.x and ZooKeeper Support</a></h5>
    <p>The final 3.x minor release, that supports ZooKeeper mode, will receive critical bug fixes and security fixes for 12 months after its release.</p>
	
    <h5 class="anchor-heading"><a id="zk_depr_timeline" class="anchor-link"></a><a href="#zk_depr_timeline">ZooKeeper and KRaft timeline</a></h5>
    <p>For details and updates on tentative timelines for ZooKeeper removal and planned KRaft feature releases, refer to <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-833%3A+Mark+KRaft+as+Production+Ready">KIP-833</a>.</p>

<h4 class="anchor-heading"><a id="zkops" class="anchor-link"></a><a href="#zkops">Operationalizing ZooKeeper</a></h4>
  Operationally, we do the following for a healthy ZooKeeper installation:
  <ul>
    <li>Redundancy in the physical/hardware/network layout: try not to put them all in the same rack, decent (but don't go nuts) hardware, try to keep redundant power and network paths, etc. A typical ZooKeeper ensemble has 5 or 7 servers, which tolerates 2 and 3 servers down, respectively. If you have a small deployment, then using 3 servers is acceptable, but keep in mind that you'll only be able to tolerate 1 server down in this case. </li>
    <li>I/O segregation: if you do a lot of write type traffic you'll almost definitely want the transaction logs on a dedicated disk group. Writes to the transaction log are synchronous (but batched for performance), and consequently, concurrent writes can significantly affect performance. ZooKeeper snapshots can be one such a source of concurrent writes, and ideally should be written on a disk group separate from the transaction log. Snapshots are written to disk asynchronously, so it is typically ok to share with the operating system and message log files. You can configure a server to use a separate disk group with the dataLogDir parameter.</li>
    <li>Application segregation: Unless you really understand the application patterns of other apps that you want to install on the same box, it can be a good idea to run ZooKeeper in isolation (though this can be a balancing act with the capabilities of the hardware).</li>
    <li>Use care with virtualization: It can work, depending on your cluster layout and read/write patterns and SLAs, but the tiny overheads introduced by the virtualization layer can add up and throw off ZooKeeper, as it can be very time sensitive</li>
    <li>ZooKeeper configuration: It's java, make sure you give it 'enough' heap space (We usually run them with 3-5G, but that's mostly due to the data set size we have here). Unfortunately we don't have a good formula for it, but keep in mind that allowing for more ZooKeeper state means that snapshots can become large, and large snapshots affect recovery time. In fact, if the snapshot becomes too large (a few gigabytes), then you may need to increase the initLimit parameter to give enough time for servers to recover and join the ensemble.</li>
    <li>Monitoring: Both JMX and the 4 letter words (4lw) commands are very useful, they do overlap in some cases (and in those cases we prefer the 4 letter commands, they seem more predictable, or at the very least, they work better with the LI monitoring infrastructure)</li>
    <li>Don't overbuild the cluster: large clusters, especially in a write heavy usage pattern, means a lot of intracluster communication (quorums on the writes and subsequent cluster member updates), but don't underbuild it (and risk swamping the cluster). Having more servers adds to your read capacity.</li>
  </ul>
  Overall, we try to keep the ZooKeeper system as small as will handle the load (plus standard growth capacity planning) and as simple as possible. We try not to do anything fancy with the configuration or application layout as compared to the official release as well as keep it as self contained as possible. For these reasons, we tend to skip the OS packaged versions, since it has a tendency to try to put things in the OS standard hierarchy, which can be 'messy', for want of a better way to word it.

  <h3 class="anchor-heading"><a id="kraft" class="anchor-link"></a><a href="#kraft">6.10 KRaft</a></h3>

  <h4 class="anchor-heading"><a id="kraft_config" class="anchor-link"></a><a href="#kraft_config">Configuration</a></h4>

  <h5 class="anchor-heading"><a id="kraft_role" class="anchor-link"></a><a href="#kraft_role">Process Roles</a></h5>

  <p>In KRaft mode each Kafka server can be configured as a controller, a broker, or both using the <code>process.roles</code> property. This property can have the following values:</p>

  <ul>
    <li>If <code>process.roles</code> is set to <code>broker</code>, the server acts as a broker.</li>
    <li>If <code>process.roles</code> is set to <code>controller</code>, the server acts as a controller.</li>
    <li>If <code>process.roles</code> is set to <code>broker,controller</code>, the server acts as both a broker and a controller.</li>
    <li>If <code>process.roles</code> is not set at all, it is assumed to be in ZooKeeper mode.</li>
  </ul>

  <p>Kafka servers that act as both brokers and controllers are referred to as "combined" servers. Combined servers are simpler to operate for small use cases like a development environment. The key disadvantage is that the controller will be less isolated from the rest of the system. For example, it is not possible to roll or scale the controllers separately from the brokers in combined mode. Combined mode is not recommended in critical deployment environments.</p>


  <h5 class="anchor-heading"><a id="kraft_voter" class="anchor-link"></a><a href="#kraft_voter">Controllers</a></h5>

  <p>In KRaft mode, specific Kafka servers are selected to be controllers (unlike the ZooKeeper-based mode, where any server can become the Controller). The servers selected to be controllers will participate in the metadata quorum. Each controller is either an active or a hot standby for the current active controller.</p>

  <p>A Kafka admin will typically select 3 or 5 servers for this role, depending on factors like cost and the number of concurrent failures your system should withstand without availability impact. A majority of the controllers must be alive in order to maintain availability. With 3 controllers, the cluster can tolerate 1 controller failure; with 5 controllers, the cluster can tolerate 2 controller failures.</p>

  <p>All of the servers in a Kafka cluster discover the quorum voters using the <code>controller.quorum.voters</code> property. This identifies the quorum controller servers that should be used. All the controllers must be enumerated. Each controller is identified with their <code>id</code>, <code>host</code> and <code>port</code> information. For example:</p>

  <pre><code class="language-bash">controller.quorum.voters=id1@host1:port1,id2@host2:port2,id3@host3:port3</code></pre>

  <p>If a Kafka cluster has 3 controllers named controller1, controller2 and controller3, then controller1 may have the following configuration:</p>

  <pre><code class="language-bash">process.roles=controller
node.id=1
listeners=CONTROLLER://controller1.example.com:9093
controller.quorum.voters=1@controller1.example.com:9093,2@controller2.example.com:9093,3@controller3.example.com:9093</code></pre>

  <p>Every broker and controller must set the <code>controller.quorum.voters</code> property. The node ID supplied in the <code>controller.quorum.voters</code> property must match the corresponding id on the controller servers. For example, on controller1, node.id must be set to 1, and so forth. Each node ID must be unique across all the servers in a particular cluster. No two servers can have the same node ID regardless of their <code>process.roles</code> values.

  <h4 class="anchor-heading"><a id="kraft_storage" class="anchor-link"></a><a href="#kraft_storage">Storage Tool</a></h4>
  <p></p>
  The <code>kafka-storage.sh random-uuid</code> command can be used to generate a cluster ID for your new cluster. This cluster ID must be used when formatting each server in the cluster with the <code>kafka-storage.sh format</code> command.

  <p>This is different from how Kafka has operated in the past. Previously, Kafka would format blank storage directories automatically, and also generate a new cluster ID automatically. One reason for the change is that auto-formatting can sometimes obscure an error condition. This is particularly important for the metadata log maintained by the controller and broker servers. If a majority of the controllers were able to start with an empty log directory, a leader might be able to be elected with missing committed data.</p>

  <h4 class="anchor-heading"><a id="kraft_debug" class="anchor-link"></a><a href="#kraft_debug">Debugging</a></h4>

  <h5 class="anchor-heading"><a id="kraft_metadata_tool" class="anchor-link"></a><a href="#kraft_metadata_tool">Metadata Quorum Tool</a></h5>

  <p>The <code>kafka-metadata-quorum</code> tool can be used to describe the runtime state of the cluster metadata partition. For example, the following command displays a summary of the metadata quorum:</p>

  <pre><code class="language-bash">$ bin/kafka-metadata-quorum.sh --bootstrap-server localhost:9092 describe --status
ClusterId:              fMCL8kv1SWm87L_Md-I2hg
LeaderId:               3002
LeaderEpoch:            2
HighWatermark:          10
MaxFollowerLag:         0
MaxFollowerLagTimeMs:   -1
CurrentVoters:          [3000,3001,3002]
CurrentObservers:       [0,1,2]</code></pre>

  <h5 class="anchor-heading"><a id="kraft_dump_log" class="anchor-link"></a><a href="#kraft_dump_log">Dump Log Tool</a></h5>

  <p>The <code>kafka-dump-log</code> tool can be used to debug the log segments and snapshots for the cluster metadata directory. The tool will scan the provided files and decode the metadata records. For example, this command decodes and prints the records in the first log segment:</p>

  <pre><code class="language-bash">$ bin/kafka-dump-log.sh --cluster-metadata-decoder --files metadata_log_dir/__cluster_metadata-0/00000000000000000000.log</code></pre>

  <p>This command decodes and prints the records in the a cluster metadata snapshot:</p>

  <pre><code class="language-bash">$ bin/kafka-dump-log.sh --cluster-metadata-decoder --files metadata_log_dir/__cluster_metadata-0/00000000000000000100-0000000001.checkpoint</code></pre>

  <h5 class="anchor-heading"><a id="kraft_shell_tool" class="anchor-link"></a><a href="#kraft_shell_tool">Metadata Shell</a></h5>

  <p>The <code>kafka-metadata-shell</code> tool can be used to interactively inspect the state of the cluster metadata partition:</p>

  <pre><code class="language-bash">$ bin/kafka-metadata-shell.sh --snapshot metadata_log_dir/__cluster_metadata-0/00000000000000000000.log
&gt;&gt; ls /
brokers  local  metadataQuorum  topicIds  topics
&gt;&gt; ls /topics
foo
&gt;&gt; cat /topics/foo/0/data
{
  "partitionId" : 0,
  "topicId" : "5zoAlv-xEh9xRANKXt1Lbg",
  "replicas" : [ 1 ],
  "isr" : [ 1 ],
  "removingReplicas" : null,
  "addingReplicas" : null,
  "leader" : 1,
  "leaderEpoch" : 0,
  "partitionEpoch" : 0
}
&gt;&gt; exit</code></pre>

  <h4 class="anchor-heading"><a id="kraft_deployment" class="anchor-link"></a><a href="#kraft_deployment">Deploying Considerations</a></h4>

  <ul>
    <li>Kafka server's <code>process.role</code> should be set to either <code>broker</code> or <code>controller</code> but not both. Combined mode can be used in development environments, but it should be avoided in critical deployment environments.</li>
    <li>For redundancy, a Kafka cluster should use 3 controllers. More than 3 controllers is not recommended in critical environments. In the rare case of a partial network failure it is possible for the cluster metadata quorum to become unavailable. This limitation will be addressed in a future release of Kafka.</li>
    <li>The Kafka controllers store all the metadata for the cluster in memory and on disk. We believe that for a typical Kafka cluster 5GB of main memory and 5GB of disk space on the metadata log director is sufficient.</li>
  </ul>

  <h4 class="anchor-heading"><a id="kraft_missing" class="anchor-link"></a><a href="#kraft_missing">Missing Features</a></h4>

  <p>The following features are not fully implemented in KRaft mode:</p>

  <ul>
    <li>Supporting JBOD configurations with multiple storage directories. Note that an Early Access release is supported in 3.7 as per <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-858%3A+Handle+JBOD+broker+disk+failure+in+KRaft">KIP-858</a>. Note that it is not yet recommended for use in production environments. Please refer to the <a href="https://cwiki.apache.org/confluence/display/KAFKA/Kafka+JBOD+in+KRaft+Early+Access+Release+Notes">release notes</a> to help us test it and provide feedback at <a href="https://issues.apache.org/jira/browse/KAFKA-16061">KAFKA-16061</a>.</li>
    <li>Modifying certain dynamic configurations on the standalone KRaft controller</li>
  </ul>

  <h4 class="anchor-heading"><a id="kraft_zk_migration" class="anchor-link"></a><a href="#kraft_zk_migration">ZooKeeper to KRaft Migration</a></h4>

  <h3>Terminology</h3>
  <ul>
    <li>Brokers that are in <b>ZK mode</b> store their metadata in Apache ZooKepeer. This is the old mode of handling metadata.</li>
    <li>Brokers that are in <b>KRaft mode</b> store their metadata in a KRaft quorum. This is the new and improved mode of handling metadata.</li>
    <li><b>Migration</b> is the process of moving cluster metadata from ZooKeeper into a KRaft quorum.</li>
  </ul>

  <h3>Migration Phases</h3>
  In general, the migration process passes through several phases.

  <ul>
    <li>In the <b>initial phase</b>, all the brokers are in ZK mode, and there is a ZK-based controller.</li>
    <li>During the <b>initial metadata load</b>, a KRaft quorum loads the metadata from ZooKeeper,</li>
    <li>In <b>hybrid phase</b>, some brokers are in ZK mode, but there is a KRaft controller.</li>
    <li>In <b>dual-write phase</b>, all brokers are KRaft, but the KRaft controller is continuing to write to ZK.</li>
    <li>When the migration has been <b>finalized</b>, we no longer write metadata to ZooKeeper.</li>
  </ul>

  <h3>Limitations</h3>
  <ul>
    <li>While a cluster is being migrated from ZK mode to KRaft mode, we do not support changing the <i>metadata
      version</i> (also known as the <i>inter.broker.protocol.version</i>.) Please do not attempt to do this during
      a migration, or you may break the cluster.</li>
    <li>After the migration has been finalized, it is not possible to revert back to ZooKeeper mode.</li>
    <li>
      During the migration, if a ZK broker is running with multiple log directories,
      any directory failure will cause the broker to shutdown.
      Brokers with broken log directories will only be able to migrate to KRaft once the directories are repaired.
      For further details refer to <a href="https://issues.apache.org/jira/browse/KAFKA-16431">KAFKA-16431</a>.
    </li>
    <li><a href="#kraft_missing">As noted above</a>, some features are not fully implemented in KRaft mode. If you are
      using one of those features, you will not be able to migrate to KRaft yet.</li>
  </ul>

  <h3>Preparing for migration</h3>
  <p>
    Before beginning the migration, the Kafka brokers must be upgraded to software version {{fullDotVersion}} and have the
    "inter.broker.protocol.version" configuration set to "{{dotVersion}}".
  </p>

  <p>
    It is recommended to enable TRACE level logging for the migration components while the migration is active. This can
    be done by adding the following log4j configuration to each KRaft controller's "log4j.properties" file.
  </p>

  <pre><code class="language-text">log4j.logger.org.apache.kafka.metadata.migration=TRACE</code></pre>

  <p>
    It is generally useful to enable DEBUG logging on the KRaft controllers and the ZK brokers during the migration.
  </p>

  <h3>Provisioning the KRaft controller quorum</h3>
  <p>
    Two things are needed before the migration can begin. First, the brokers must be configured to support the migration and second,
    a KRaft controller quorum must be deployed. The KRaft controllers should be provisioned with the same cluster ID as
    the existing Kafka cluster. This can be found by examining one of the "meta.properties" files in the data directories
    of the brokers, or by running the following command.
  </p>

  <pre><code class="language-bash">$ bin/zookeeper-shell.sh localhost:2181 get /cluster/id</code></pre>

  <p>
    The KRaft controller quorum should also be provisioned with the latest <code>metadata.version</code>.
    This is done automatically when you format the node with the <code>kafka-storage.sh</code> tool.
    For further instructions on KRaft deployment, please refer to <a href="#kraft_config">the above documentation</a>.
  </p>

  <p>
    In addition to the standard KRaft configuration, the KRaft controllers will need to enable support for the migration
    as well as provide ZooKeeper connection configuration.
  </p>

  <p>
    Here is a sample config for a KRaft controller that is ready for migration:
  </p>
  <pre><code class="language-text"># Sample KRaft cluster controller.properties listening on 9093
process.roles=controller
node.id=3000
controller.quorum.voters=3000@localhost:9093
controller.listener.names=CONTROLLER
listeners=CONTROLLER://:9093

# Enable the migration
zookeeper.metadata.migration.enable=true

# ZooKeeper client configuration
zookeeper.connect=localhost:2181

# The inter broker listener in brokers to allow KRaft controller send RPCs to brokers
inter.broker.listener.name=PLAINTEXT

# Other configs ...</code></pre>

  <p><em>Note: The KRaft cluster <code>node.id</code> values must be different from any existing ZK broker <code>broker.id</code>.
  In KRaft-mode, the brokers and controllers share the same Node ID namespace.</em></p>

  <h3>Enter Migration Mode on the Brokers</h3>
  <p>
    Once the KRaft controller quorum has been started, the brokers will need to be reconfigured and restarted. Brokers
    may be restarted in a rolling fashion to avoid impacting cluster availability. Each broker requires the
    following configuration to communicate with the KRaft controllers and to enable the migration.
  </p>

  <ul>
    <li><a href="#brokerconfigs_broker.id">broker.id</a>: Ensure <code>broker.id</code> is set to a non-negative integer even if <code>broker.id.generation.enable</code> is enabled (default is enabled). Additionally, ensure <code>broker.id</code> does not exceed <code>reserved.broker.max.id</code> to avoid failure.</li>
    <li><a href="#brokerconfigs_controller.quorum.voters">controller.quorum.voters</a></li>
    <li><a href="#brokerconfigs_controller.listener.names">controller.listener.names</a></li>
    <li>The controller.listener.name should also be added to <a href="#brokerconfigs_listener.security.protocol.map">listener.security.property.map</a></li>
    <li><a href="#brokerconfigs_zookeeper.metadata.migration.enable">zookeeper.metadata.migration.enable</a></li>
  </ul>

  <p>Here is a sample config for a broker that is ready for migration:</p>

  <pre><code class="language-text"># Sample ZK broker server.properties listening on 9092
broker.id=0
listeners=PLAINTEXT://:9092
advertised.listeners=PLAINTEXT://localhost:9092
listener.security.protocol.map=PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT

# Set the IBP
inter.broker.protocol.version={{dotVersion}}

# Enable the migration
zookeeper.metadata.migration.enable=true

# ZooKeeper client configuration
zookeeper.connect=localhost:2181

# KRaft controller quorum configuration
controller.quorum.voters=3000@localhost:9093
controller.listener.names=CONTROLLER</code></pre>

  <p>
    <em>Note: Once the final ZK broker has been restarted with the necessary configuration, the migration will automatically begin.</em>
    When the migration is complete, an INFO level log can be observed on the active controller:
  </p>

  <pre>Completed migration of metadata from Zookeeper to KRaft</pre>

  <h3>Migrating brokers to KRaft</h3>
  <p>
    Once the KRaft controller completes the metadata migration, the brokers will still be running
    in ZooKeeper mode. While the KRaft controller is in migration mode, it will continue sending
    controller RPCs to the ZooKeeper mode brokers. This includes RPCs like UpdateMetadata and
    LeaderAndIsr.
  </p>

  <p>
    To migrate the brokers to KRaft, they simply need to be reconfigured as KRaft brokers and restarted. Using the above
    broker configuration as an example, we would replace the <code>broker.id</code> with <code>node.id</code> and add
    <code>process.roles=broker</code>. It is important that the broker maintain the same Broker/Node ID when it is restarted.
    The zookeeper configurations should be removed at this point.
  </p>

  <p>
    If your broker has authorization configured via the <code>authorizer.class.name</code> property
    using <code>kafka.security.authorizer.AclAuthorizer</code>, this is also the time to change it
    to use <code>org.apache.kafka.metadata.authorizer.StandardAuthorizer</code> instead.
  </p>

  <pre><code class="language-text"># Sample KRaft broker server.properties listening on 9092
process.roles=broker
node.id=0
listeners=PLAINTEXT://:9092
advertised.listeners=PLAINTEXT://localhost:9092
listener.security.protocol.map=PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT

# Don't set the IBP, KRaft uses "metadata.version" feature flag
# inter.broker.protocol.version={{dotVersion}}

# Remove the migration enabled flag
# zookeeper.metadata.migration.enable=true

# Remove ZooKeeper client configuration
# zookeeper.connect=localhost:2181

# Keep the KRaft controller quorum configuration
controller.quorum.voters=3000@localhost:9093
controller.listener.names=CONTROLLER</code></pre>

  <p>
    Each broker is restarted with a KRaft configuration until the entire cluster is running in KRaft mode.
  </p>

  <h3>Finalizing the migration</h3>
  <p>
    Once all brokers have been restarted in KRaft mode, the last step to finalize the migration is to take the
    KRaft controllers out of migration mode. This is done by removing the "zookeeper.metadata.migration.enable"
    property from each of their configs and restarting them one at a time.
  </p>
  <p>
    Once the migration has been finalized, you can safely deprovision your ZooKeeper cluster, assuming you are
    not using it for anything else. After this point, it is no longer possible to revert to ZooKeeper mode.
  </p>

  <pre><code class="language-text"># Sample KRaft cluster controller.properties listening on 9093
process.roles=controller
node.id=3000
controller.quorum.voters=3000@localhost:9093
controller.listener.names=CONTROLLER
listeners=CONTROLLER://:9093

# Disable the migration
# zookeeper.metadata.migration.enable=true

# Remove ZooKeeper client configuration
# zookeeper.connect=localhost:2181

# Other configs ...</code></pre>

  <h3>Reverting to ZooKeeper mode During the Migration</h3>
  <p>
    While the cluster is still in migration mode, it is possible to revert to ZooKeeper mode.  The process
    to follow depends on how far the migration has progressed. In order to find out how to revert,
    select the <b>final</b> migration step that you have <b>completed</b> in this table.
  </p>
  <p>
    Note that the directions given here assume that each step was fully completed, and they were
    done in order. So, for example, we assume that if "Enter Migration Mode on the Brokers" was
    completed, "Provisioning the KRaft controller quorum" was also fully completed previously.
  </p>
  <p>
    If you did not fully complete any step, back out whatever you have done and then follow revert
    directions for the last fully completed step.
  </p>

  <table class="data-table">
      <tbody>
      <tr>
        <th>Final Migration Section Completed</th>
        <th>Directions for Reverting</th>
        <th>Notes</th>
      </tr>
      <tr>
        <td>Preparing for migration</td>
        <td>
          The preparation section does not involve leaving ZooKeeper mode. So there is nothing to do in the
          case of a revert.
        </td>
        <td>
        </td>
      </tr>
      <tr>
        <td>Provisioning the KRaft controller quorum</td>
        <td>
          <ul>
            <li>
              Deprovision the KRaft controller quorum.
            </li>
            <li>
              Then you are done.
            </li>
          </ul>
        </td>
        <td>
        </td>
      </tr>
      <tr>
        <td>Enter Migration Mode on the brokers</td>
        <td>
          <ul>
            <li>
              Deprovision the KRaft controller quorum.
            </li>
            <li>
              Using <code>zookeeper-shell.sh</code>, run <code>rmr /controller</code> so that one
              of the brokers can become the new old-style controller.
            </li>
            <li>
              On each broker, remove the <code>zookeeper.metadata.migration.enable</code>,
              <code>controller.listener.names</code>, and <code>controller.quorum.voters</code>
              configurations, and replace <code>node.id</code> with <code>broker.id</code>.
              Then perform a rolling restart of all brokers.
            </li>
            <li>
              Then you are done.
            </li>
          </ul>
        </td>
        <td>
          It is important to perform the <code>zookeeper-shell.sh</code> step <b>quickly</b>, to minimize the amount of
          time that the cluster lacks a controller. Until the <code> /controller</code> znode is deleted,
          you can also ignore any errors in the broker log about failing to connect to the Kraft controller.
          Those error logs should disappear after second roll to pure zookeeper mode.
        </td>
      </tr>
      <tr>
        <td>Migrating brokers to KRaft</td>
        <td>
          <ul>
            <li>
              On each broker, remove the <code>process.roles</code> configuration,
              replace the <code>node.id</code> with <code>broker.id</code> and
              restore the <code>zookeeper.connect</code> configuration to its previous value.
              If your cluster requires other ZooKeeper configurations for brokers, such as
              <code>zookeeper.ssl.protocol</code>, re-add those configurations as well.
              Then perform a rolling restart of all brokers.
            </li>
            <li>
              Deprovision the KRaft controller quorum.
            </li>
            <li>
              Using <code>zookeeper-shell.sh</code>, run <code>rmr /controller</code> so that one
              of the brokers can become the new old-style controller.
            </li>
            <li>
              On each broker, remove the <code>zookeeper.metadata.migration.enable</code>,
              <code>controller.listener.names</code>, and <code>controller.quorum.voters</code>
              configurations.
              Then perform a second rolling restart of all brokers.
            </li>
            <li>
              Then you are done.
            </li>
          </ul>
        </td>
        <td>
          <ul>
            <li>
              It is important to perform the <code>zookeeper-shell.sh</code> step <b>quickly</b>, to minimize the amount of
              time that the cluster lacks a controller. Until the <code> /controller</code> znode is deleted,
              you can also ignore any errors in the broker log about failing to connect to the Kraft controller.
              Those error logs should disappear after second roll to pure zookeeper mode.
            </li>
            <li>
              Make sure that on the first cluster roll, <code>zookeeper.metadata.migration.enable</code> remains set to
              </code>true</code>. <b>Do not set it to false until the second cluster roll.</b>
            </li>
          </ul>
        </td>
      </tr>
      <tr>
        <td>Finalizing the migration</td>
        <td>
          If you have finalized the ZK migration, then you cannot revert.
        </td>
        <td>
          Some users prefer to wait for a week or two before finalizing the migration. While this
          requires you to keep the ZooKeeper cluster running for a while longer, it may be helpful
          in validating KRaft mode in your cluster.
        </td>
      </tr>
    </tbody>
 </table>


<h3 class="anchor-heading"><a id="tiered_storage" class="anchor-link"></a><a href="#kraft">6.11 Tiered Storage</a></h3>

<h4 class="anchor-heading"><a id="tiered_storage_overview" class="anchor-link"></a><a href="#tiered_storage_overview">Tiered Storage Overview</a></h4>

<p>Kafka data is mostly consumed in a streaming fashion using tail reads. Tail reads leverage OS's page cache to serve the data instead of disk reads.
  Older data is typically read from the disk for backfill or failure recovery purposes and is infrequent.</p>

<p>In the tiered storage approach, Kafka cluster is configured with two tiers of storage - local and remote.
  The local tier is the same as the current Kafka that uses the local disks on the Kafka brokers to store the log segments.
  The new remote tier uses external storage systems, such as HDFS or S3, to store the completed log segments.
  Please check <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-405%3A+Kafka+Tiered+Storage">KIP-405</a> for more information.
</p>

<p><b>Note: Tiered storage is considered as an early access feature, and is not recommended for use in production environments</b></p>

<h4 class="anchor-heading"><a id="tiered_storage_config" class="anchor-link"></a><a href="#tiered_storage_config">Configuration</a></h4>

<h5 class="anchor-heading"><a id="tiered_storage_config_broker" class="anchor-link"></a><a href="#tiered_storage_config_broker">Broker Configurations</a></h5>

<p>By default, Kafka server will not enable tiered storage feature. <code>remote.log.storage.system.enable</code>
  is the property to control whether to enable tiered storage functionality in a broker or not. Setting it to "true" enables this feature.
</p>

<p><code>RemoteStorageManager</code> is an interface to provide the lifecycle of remote log segments and indexes. Kafka server
  doesn't provide out-of-the-box implementation of RemoteStorageManager. Configuring <code>remote.log.storage.manager.class.name</code>
  and <code>remote.log.storage.manager.class.path</code> to specify the implementation of RemoteStorageManager.
</p>

<p><code>RemoteLogMetadataManager</code> is an interface to provide the lifecycle of metadata about remote log segments with strongly consistent semantics.
  By default, Kafka provides an implementation with storage as an internal topic. This implementation can be changed by configuring
  <code>remote.log.metadata.manager.class.name</code> and <code>remote.log.metadata.manager.class.path</code>.
  When adopting the default kafka internal topic based implementation, <code>remote.log.metadata.manager.listener.name</code>
  is a mandatory property to specify which listener the clients created by the default RemoteLogMetadataManager implementation.
</p>


<h5 class="anchor-heading"><a id="tiered_storage_config_topic" class="anchor-link"></a><a href="#tiered_storage_config_topic">Topic Configurations</a></h5>

<p>After correctly configuring broker side configurations for tiered storage feature, there are still configurations in topic level needed to be set.
  <code>remote.storage.enable</code> is the switch to determine if a topic wants to use tiered storage or not. By default it is set to false.
  After enabling <code>remote.storage.enable</code> property, the next thing to consider is the log retention.
  When tiered storage is enabled for a topic, there are 2 additional log retention configurations to set:

<ul>
  <li><code>local.retention.ms</code></li>
  <li><code>retention.ms</code></li>
  <li><code>local.retention.bytes</code></li>
  <li><code>retention.bytes</code></li>
</ul>

  <p>The configuration prefixed with <code>local</code> are to specify the time/size the "local" log file can accept before moving to remote storage, and then get deleted.
  If unset, The value in <code>retention.ms</code> and <code>retention.bytes</code> will be used.</p>

<h4 class="anchor-heading"><a id="tiered_storage_config_ex" class="anchor-link"></a><a href="#tiered_storage_config_ex">Quick Start Example</a></h4>

<p>Apache Kafka doesn't provide an out-of-the-box RemoteStorageManager implementation. To have a preview of the tiered storage
  feature, the <a href="https://github.com/apache/kafka/blob/trunk/storage/src/test/java/org/apache/kafka/server/log/remote/storage/LocalTieredStorage.java">LocalTieredStorage</a>
  implemented for integration test can be used, which will create a temporary directory in local storage to simulate the remote storage.
</p>

<p>To adopt the `LocalTieredStorage`, the test library needs to be built locally</p>
<pre><code class="language-bash"># please checkout to the specific version tag you're using before building it
# ex: `git checkout {{fullDotVersion}}`
$ ./gradlew clean :storage:testJar</code></pre>
<p>After build successfully, there should be a `kafka-storage-x.x.x-test.jar` file under `storage/build/libs`.
Next, setting configurations in the broker side to enable tiered storage feature.</p>

<pre><code class="language-text"># Sample Zookeeper/Kraft broker server.properties listening on PLAINTEXT://:9092
remote.log.storage.system.enable=true

# Setting the listener for the clients in RemoteLogMetadataManager to talk to the brokers.
remote.log.metadata.manager.listener.name=PLAINTEXT

# Please provide the implementation info for remoteStorageManager.
# This is the mandatory configuration for tiered storage.
# Here, we use the `LocalTieredStorage` built above.
remote.log.storage.manager.class.name=org.apache.kafka.server.log.remote.storage.LocalTieredStorage
remote.log.storage.manager.class.path=/PATH/TO/kafka-storage-{{fullDotVersion}}-test.jar

# These 2 prefix are default values, but customizable
remote.log.storage.manager.impl.prefix=rsm.config.
remote.log.metadata.manager.impl.prefix=rlmm.config.

# Configure the directory used for `LocalTieredStorage`
# Note, please make sure the brokers need to have access to this directory
rsm.config.dir=/tmp/kafka-remote-storage

# This needs to be changed if number of brokers in the cluster is more than 1
rlmm.config.remote.log.metadata.topic.replication.factor=1

# Try to speed up the log retention check interval for testing
log.retention.check.interval.ms=1000</code></pre>

<p>Following <a href="#quickstart_startserver">quick start guide</a> to start up the kafka environment.
  Then, create a topic with tiered storage enabled with configs:

<pre><code class="language-bash"># remote.storage.enable=true -> enables tiered storage on the topic
# local.retention.ms=1000 -> The number of milliseconds to keep the local log segment before it gets deleted.
# Note that a local log segment is eligible for deletion only after it gets uploaded to remote.
# retention.ms=3600000 -> when segments exceed this time, the segments in remote storage will be deleted
# segment.bytes=1048576 -> for test only, to speed up the log segment rolling interval
# file.delete.delay.ms=10000 -> for test only, to speed up the local-log segment file delete delay

$ bin/kafka-topics.sh --create --topic tieredTopic --bootstrap-server localhost:9092 \
--config remote.storage.enable=true --config local.retention.ms=1000 --config retention.ms=3600000 \
--config segment.bytes=1048576 --config file.delete.delay.ms=1000</code></pre>

<p>Try to send messages to the `tieredTopic` topic to roll the log segment:</p>

<pre><code class="language-bash">$ bin/kafka-producer-perf-test.sh --topic tieredTopic --num-records 1200 --record-size 1024 --throughput -1 --producer-props bootstrap.servers=localhost:9092</code></pre>

<p>Then, after the active segment is rolled, the old segment should be moved to the remote storage and get deleted.
  This can be verified by checking the remote log directory configured above. For example:
</p>

<pre><code class="language-bash">$ ls /tmp/kafka-remote-storage/kafka-tiered-storage/tieredTopic-0-jF8s79t9SrG_PNqlwv7bAA
00000000000000000000-knnxbs3FSRyKdPcSAOQC-w.index
00000000000000000000-knnxbs3FSRyKdPcSAOQC-w.snapshot
00000000000000000000-knnxbs3FSRyKdPcSAOQC-w.leader_epoch_checkpoint
00000000000000000000-knnxbs3FSRyKdPcSAOQC-w.timeindex
00000000000000000000-knnxbs3FSRyKdPcSAOQC-w.log</code></pre>

<p>Lastly, we can try to consume some data from the beginning and print offset number, to make sure it will successfully fetch offset 0 from the remote storage.</p>

<pre><code class="language-bash">$ bin/kafka-console-consumer.sh --topic tieredTopic --from-beginning --max-messages 1 --bootstrap-server localhost:9092 --property print.offset=true</code></pre>

<p>Please note, if you want to disable tiered storage at the cluster level, you should delete the tiered storage enabled topics explicitly.
  Attempting to disable tiered storage at the cluster level without deleting the topics using tiered storage will result in an exception during startup.</p>

<pre><code class="language-bash">$ bin/kafka-topics.sh --delete --topic tieredTopic --bootstrap-server localhost:9092</code></pre>

<p>After topics are deleted, you're safe to set <code>remote.log.storage.system.enable=false</code> in the broker configuration.</p>

<h4 class="anchor-heading"><a id="tiered_storage_limitation" class="anchor-link"></a><a href="#tiered_storage_limitation">Limitations</a></h4>

<p>While the early access release of Tiered Storage offers the opportunity to try out this new feature, it is important to be aware of the following limitations:
<ul>
  <li>No support for clusters with multiple log directories (i.e. JBOD feature)</li>
  <li>No support for compacted topics</li>
  <li>Cannot disable tiered storage at the topic level</li>
  <li>Deleting tiered storage enabled topics is required before disabling tiered storage at the broker level</li>
  <li>Admin actions related to tiered storage feature are only supported on clients from version 3.0 onwards</li>
</ul>

<p>For more information, please check <a href="https://cwiki.apache.org/confluence/display/KAFKA/Kafka+Tiered+Storage+Early+Access+Release+Notes">Tiered Storage Early Access Release Note</a>.
</p>


</script>

<div class="p-ops"></div>