È possibile aggiungere partizioni a un argomento esistente in Kafka 0.8.2

Question

Ho un cluster Kafka in esecuzione con 2 partizioni. Stavo cercando un modo per aumentare il conteggio delle partizioni a 3. Tuttavia, non voglio perdere i messaggi esistenti nell'argomento. Ho provato a fermare Kafka, modificando il file server.properties per aumentare il numero di partizioni su 3 e riavviare Kafka. Tuttavia, questo non sembra cambiare nulla. Usando Kafka ConsumerOffsetChecker, continuo a vedere che sta usando solo 2 partizioni. La versione di Kafka che sto usando è 0.8.2.2. Nella versione 0.8.1, c'era uno script chiamato kafka-add-partitions.sh, che immagino potrebbe fare il trucco. Tuttavia, non vedo nessuno di questi script in 0.8.2. C'è un modo per realizzare questo? Ho fatto esperimenti con la creazione di un argomento completamente nuovo e per quello sembra utilizzare 3 partizioni in base alla modifica nel file server.properties. Tuttavia, per gli argomenti esistenti, non sembra preoccuparsene.

Answer 1

Sembra che è possibile utilizzare this script anziché:

bin/kafka-topics.sh --zookeeper zk_host:port/chroot --alter --topic my_topic_name 
    --partitions 40 

Nel codice sembra che fanno stessa cosa:

AdminUtils.createOrUpdateTopicPartitionAssignmentPathInZK(topic, partitionReplicaList, zkClient, true) 

kafka-topics.sh esegue this pezzo di codice, così come AddPartitionsCommand usato da script kafka-add-partition.

Tuttavia è necessario essere consapevoli di ri-partizionamento quando usando la chiave:

Essere consapevoli del fatto che un caso d'uso per le partizioni è semanticamente partizione dati e partizioni aggiungendo non cambia la compartimentazione dei dati esistenti quindi questo può disturbare i consumatori se si affidano a quella partizione . Cioè se i dati sono partizionati da hash(key) % number_of_partitions allora questo partizionamento sarà potenzialmente mischiato aggiungendo le partizioni, ma Kafka non tenterà di ridistribuire automaticamente i dati in alcun modo a .

Answer 2

Si prega di notare che nel mio caso il valore zk_host: port/chroot per il parametro --zookeeper ha gettato la seguente eccezione:

ERRORE java.lang.IllegalArgumentException: Argomento my_topic_name non esiste sul ZK percorso zk_host: porto/chroot.

Così, ho provato quanto segue e ha funzionato:

bin/kafka-topics.sh --alter --zookeeper zk_host:port --topic my_topic_name --partitions 10