Broker Log Management

As mentioned previously, Kafka brokers save their data as log segments in a directory. The logs are rotated depending on the size and time settings.

The most common log retention settings to adjust for your cluster are shown below. These are accessible in Cloudera Manager via the Kafka->Configuration tab.

• log.dirs: The location for the Kafka data (that is, topic directories and log segments).
• log.retention.{ms|minutes|hours}: The retention period for the entire log. Any older log segments are removed.
• log.retention.bytes: The retention size for the entire log.

There are many more variables available for fine-tuning broker log management. For more detailed information look at the relevant variables in the Apache Kafka documentation topic Broker Configs.

• log.dirs
• log.flush.*
• log.retention.*
• log.roll.*
• log.segment.*

Record Management

There are two pieces two record management, log segments and log cleaner.

As part of the general data storage, Kafka rolls logs periodically based on size and/or time limits. Once either limit is hit, a new log segment is created with the all new data being placed there, while older log segments should generally no longer change. This helps limit the risk of data loss or corruption to a single segment instead of the entire log.

• log.roll.{ms|hours}: The time period each log segment. Once the current segment is older than this value, it goes through log segment rotation.
• log.segment.bytes: The maximum size for a single log segment.

There is an alternative to simply removing log segments for a partition. There is another feature based on the log cleaner. When the log cleaner is enabled, then individual records in older log segments can be managed differently:

• log.cleaner.enable: This is a global setting in Kafka to enable the log cleaner.
• cleanup.policy: This is a per-topic property that is usually set at topic creation time. There are two valid values for this property, delete and compact.
• log.cleaner.min.compaction.lag.ms: This is the retention period for the “head” of the log. Only records outside of this retention period will be compacted by the log cleaner.

The compact policy, also called log compaction, assumes that the "most recent Kafka record is important." Some examples include tracking a current email address or tracking a current mailing address. With log compaction, older records with the same key are removed from a log segment and the latest one kept. This effectively removes some offsets from the partition.

Broker Garbage Collection

To set the JVM Garbage Collection log path on the brokers:

1. In Cloudera Manager, go to Kafka > Configurations.
2. Find the property Kafka Broker Environment Advanced Configuration Snippet (Safety Valve) and add the line:

   KAFKA_GC_LOG_OPTS="-Xloggc:/var/logs/kafka/gc-broker.log"

3. Restart the Kafka service to apply the new configuration.

To set the JVM Garbage log rotation on the brokers:

1. In Cloudera Manager, go to Kafka > Configurations.
2. Find the property Kafka Broker Environment Advanced Configuration Snippet (Safety Valve) and add the line (modified as appropriate):

   -XX:+UseGCLogFileRotation
   -XX:NumberOfGCLogFiles=10
   -XX:GCLogFileSize=100M
                   

3. Restart the Kafka service to apply the new configuration.

Using rsync to Copy Files from One Broker to Another

You can run rsync command to copies over all data from an old broker to a new broker, preserving modification times and permissions. Using rsync allows you to avoid having the re-replicate the data from the leader. You do need to ensure the disk structures match between the two brokers, or you have to verify the meta.properties file between the source and destination brokers (because there is one meta.properties file for each data directory).

Run command on destination broker:

rsync -avz
        src_broker:src_data_dir
        dest_data_dir

If you plan to change the broker ID, edit dest_data_dir/meta.properties.

kafka-topics Command

Use the kafka-topics command to generate a snapshot of topics in the Kafka cluster.

$ kafka-topics --zookeeper zkhost --describe

Topic: topic-a1 PartitionCount:3        ReplicationFactor:3     Configs:
Topic: topic-a1 Partition: 0    Leader: 64      Replicas: 64,62,63      Isr: 64,62,63
Topic: topic-a1 Partition: 1    Leader: 62      Replicas: 62,63,64      Isr: 62,63,64
Topic: topic-a1 Partition: 2    Leader: 63      Replicas: 63,64,62      Isr: 63,64,62
Topic: topic-a2 PartitionCount:1        ReplicationFactor:3     Configs:
Topic: topic-a2 Partition: 0    Leader: 64      Replicas: 64,62,63      Isr: 64,62,63

In this clean output, you can see each topic and the basic partition information. Note the following about the output:

• Partition count: The more partitions, the higher the possible parallelism among consumers and producers.
• Replication factor: Shows 1 for no redundancy and higher for more redundancy.
• Replicas and in-sync replicas (ISR): Shows which broker ID’s have the partitions and which replicas are current.

There are situations where this command shows an invalid value for the leader broker ID or the number of ISRs is fewer than the number of replicas. In those cases, there may be something wrong with those specific topics.

Note that it is possible to change topic configuration properties using this command. As mentioned earlier, increasing the partition count and/or the replication factor is not recommended.

kafka-configs Command

The kafka-configs command allows you to set and unset properties to topics. It is recommended that you use Cloudera Manager instead of this command to change properties on brokers, because this command bypasses any Cloudera Manager safety checks.

$ kafka-configs --zookeeper zkhost --entity-type topics --entity-name topic --alter --add-config property=value

$ kafka-configs --zookeeper zkhost --entity-type
            topics --entity-name topic --describe

$ kafka-configs --zookeeper zkhost --entity-type
            topics --entity-name topic --alter --delete-config property

The Apache Kafka documentation includes a complete list of topic properties.

kafka-console Command

The kafka-console command can be useful in a couple of ways:

• Acting as an independent consumer of particular topics. This can be useful to compare results against a consumer program that you’ve written.
• To test general topic consumption without the need to write any consumer code.

Examples of usage:

$ kafka-console-consumer --bootstrap-server <broker1>,<broker2>... --topic <topic> --from-beginning
<record-earliest-offset>
<record-earliest-offset+1>

Note the following about the command:

• This command prints all records and keeps outputting as more records are written to the topic.
• If the kafka-console-consumer command is given no flags, it displays the full help message.
• In older versions of Kafka, it may have been necessary to use the --new-consumer flag. As of Apache Kafka version 0.10.2, this is no longer necessary.

kafka-console-producer Command

This command is used to write messages to a topic. It is typically not as useful as the console consumer, but it can be useful when the messages are in a text based format. In general, the usage will be something like:

$ cat file | kafka-console-producer args

kafka-consumer-groups Command

The basic usage of the kafka-consumer-groups command is:

$ kafka-consumer-groups.sh --bootstrap-server broker1,broker2... --describe --group GROUP_ID

This command is primarily useful for debugging consumer offset issues. The output from the command shows the log and consumer offsets for each partition connected to the consumer group corresponding to GROUP_ID. You can see at a glance which consumers are current with their partition and which ones are behind. From there, you can determine which partitions (and likely the corresponding brokers) are slow.

Beyond this debugging usage, there are other more advanced options to this command:

• --execute --reset-offsets SCENARIO_OPTION: Resets the offsets for a consumer group to a particular value based on the SCENARIO_OPTION flag given.

  Valid flags for SCENARIO_OPTION are:

  • --to-datetime
  • --by-period
  • --to-earliest
  • --to-latest
  • --shift-by
  • --from-file
  • --to-current

  You will likely want to set the --topic flag to restrict this change to a specific topic or a specific set of partitions within that topic.

This command can be used to reset all offsets on all topics; this is something you probably won’t ever want to do. It is highly recommended that you use this command carefully.

kafka-reassign-partitions Command

This script allows a great deal of control over partitions in a Kafka cluster. You can do either or both of the following:

• Change the ordering of the partition assignment list. This is usually done to control leader imbalances between brokers.
• Move replicas from one broker to another. The most common usage of this is after adding new brokers to the cluster.

To reassign partitions:

1. Create a list of topics you want to move.

   topics-to-move.json
   {"topics":  [{"topic": "foo1"},
                {"topic": "foo2"}],
    "version":1
   }

2. Use the --generate option in kafka-reassign-partitions.sh to list the distribution of partitions and replicas on your current brokers, followed by a list of suggested locations for partitions on your new broker.

   $ bin/kafka-reassign-partitions.sh --zookeeper localhost:2181
           --topics-to-move-json-file topics-to-move.json
           --broker-list "4"
           --generate
   
   Current partition replica assignment
   
   {"version":1,
    "partitions":[{"topic":"foo1","partition":2,"replicas":[1,2]},
                  {"topic":"foo1","partition":0,"replicas":[3,1]},
                  {"topic":"foo2","partition":2,"replicas":[1,2]},
                  {"topic":"foo2","partition":0,"replicas":[3,2]},
                  {"topic":"foo1","partition":1,"replicas":[2,3]},
                  {"topic":"foo2","partition":1,"replicas":[2,3]}]
   }
   
   {"version":1,
    "partitions":[{"topic":"foo1","partition":3,"replicas":[4]},
                  {"topic":"foo1","partition":1,"replicas":[4]},
                  {"topic":"foo2","partition":2,"replicas":[4]}]
   }

3. Revise the suggested list if required, and then save it as a JSON file.
4. Use the --execute option in kafka-reassign-partitions.sh to start the redistribution process, which can take several hours in some cases.

   $ bin/kafka-reassign-partitions.sh \
           --zookeeper localhost:2181 \
           --reassignment-json-file expand-cluster-reassignment.json
           --execute

5. Use the --verify option in kafka-reassign-partitions.sh to check the status of your partitions.

There are several caveats to using this command:

• It is highly recommended that you minimize the volume of replica changes. Say, instead of moving ten replicas with a single command, move two at a time in order to keep the cluster healthy.
• It is not possible to use this command to make an out-of-sync replica into the leader partition.
• Given the earlier restrictions, it is best to use this command only when all brokers and topics are healthy.

kafka-*-perf-test Scripts

The kafka-*-perf-test scripts can be used in several ways. In general, it is expected that these tools should be used on a test or development cluster.

• Measuring read and/or write throughput.
• Stress testing the cluster based on specific parameters (such as message size).
• Load testing for the purpose of evaluating specific metrics or determining the impact of cluster configuration changes.

The kafka-producer-perf-test script can either create a randomly generated byte record:

$ kafka-producer-perf-test --topic TOPIC --record-size SIZE_IN_BYTES

or randomly read from a set of provided records:

$ kafka-producer-perf-test --topic TOPIC --payload-delimiter DELIMITER --payload-file INPUT_FILE

where the INPUT_FILE is a concatenated set of pre-generated messages separated by DELIMITER. This script keeps producing messages or limited based on the --num-records flag.

The kafka-consumer-perf-test is:

$ kafka-consumer-perf-test --broker-list host1:port1,host2:port2,... --zookeeper zk1:port1,zk2:port2,... --topic TOPIC

The flags of most interest for this command are:

• --group gid: If you run more than one instance of this test, you will want to set different ids for each instance.
• --num-fetch-threads: Defaults to 1. Increase if higher throughput testing is needed.
• --from-latest: To start consuming from the latest offset. May be needed for certain types of testing.