Persist Demoted and Removed brokers

[morgangalpin]

Overview

Changes to the sets of demoted and removed brokers are retained in memory only. So when Cruise Control restarts, the information is lost and upon startup, load is again shifted back onto the demoted/removed brokers. The original intent was to add persistence for this data, as indicated by this old TODO in Executor.

The main goal of this change is to add a persistence mechanism that will serve to persist the sets of demoted/removed brokers, while also being general enough that other runtime data can be persisted as well in the future. A secondary goal of this change is to refactor the logic around manipulating demoted/removed brokers to reuse the common functionality, e.g. removing expired brokers from the sets.

Design

PersistedData package

Introduce a package to contain the layers of persisted data classes introduced below.

PersistedMap

An implementation of the standard Map interface that maintains a cached version of the map data in memory while also persisting any changes to the map. Upon creation, the map reads and initializes to the existing persisted state.

KafkaPersistedMap

A specialization of the PersistedMap class where the data is published to a kafka topic. Data in the map is eventually consistent as updates to the map data are read from the topic and used to update the cached data. This would also allow multiple instances of cruise control to share access to the persisted data.

ExecutorPersistedData

The Executor has two pieces of data that need persisting: the maps of demoted and removed brokers. Each map contains a mapping of broker id to the time they were demoted/removed. The data is read many times, but updated infrequently. This class would maintain both maps and ensure they're persisted and available for reading.

Executor

Needs to be updated to use an instance of ExecutorPersistedData as the store for both demoted and removed brokers. Also some refactoring to consolidate common functionality.

PersistedDataConfig

Introduces configuration for PersistedData. Each configuration will be prefixed with "persisted.data.".

Configuration Values

• persisted.data.backing.method: The method to use to store persisted data. This is the first "persisted.data" config to set which will determine which other configs of the series should be configured. At present, the only option is "kafka" which is also the default.
• persisted.data.kafka.topic.name: The name of the kafka topic to use to persist data when persisted.data.backing.method is set to "kafka". If the topic is not present, then it will be created. To be able to create the topic, the user that cruise control connects as needs to have CREATE permission at the cluster level. The user will also need READ, WRITE, DESCRIBE and ALTER permissions on this topic.
• persisted.data.kafka.topic.partition.count: The number of partitions to ensure are present for the kafka topic. Only applies when persisted.data.backing.method is set to "kafka". If the topic has fewer than this number of partitions, then partitions will be added.
• persisted.data.kafka.topic.replication.factor: The replication factor to use for the kafka topic. Only applies when persisted.data.backing.method is set to "kafka". Multiple partition replicas are desirable to ensure the topic is reasonably available.
• persisted.data.kafka.topic.additional.configs.map: The configs to apply to the kafka topic used to persist Cruise Control data. Only applies if persisted.data.backing.method is set to "kafka". This "list" should be a semicolon separated string of 'key=value' pairs. The keys and values need to be valid Kafka Topic configs. See: https://kafka.apache.org/documentation/#topicconfigs
• persisted.data.kafka.producer.additional.configs.map: The additional configs to use when creating the kafka producer to persist Cruise Control data. Only applies if persisted.data.backing.method is set to "kafka". This "list" should be a semicolon separated string of 'key=value' pairs. The keys and values need to be valid Kafka Producer configs. See: https://kafka.apache.org/documentation/#producerconfigs
• persisted.data.kafka.consumer.additional.configs.map: The additional configs to use when creating the kafka consumer to read persisted Cruise Control data. Only applies if persisted.data.backing.method is set to "kafka". This "list" should be a semicolon separated string of 'key=value' pairs. The keys and values need to be valid Kafka Consumer configs. See: https://kafka.apache.org/documentation/#consumerconfigs