Introdução

O Kafka, conhecido por sua alta performance em operações de leitura e escrita, realiza essas operações em Tópicos. A criação de um tópico é um processo crucial que envolve a alocação de partições e transições de estado. Embora o comando para criar um tópico seja simples:

```

bash kafka-topics.sh
--bootstrap-server localhost:9092
--create --topic meuTopico
--partitions 10
--replication-factor 2

  

Os bastidores dessa operação são mais complexos do que aparentam, abrangendo estratégias de distribuição de partições e gerenciamento de estado. Esta análise explora os detalhes internos da criação de tópicos no Kafka, baseada na versão 2.8.2.

Fluxo Geral de Criação de Tópicos
---------------------------------

A criação de um tópico pode ser dividida em duas fases principais:

1. **Orquestração pelo Controller:** O Controller do Kafka recebe a solicitação de criação, determina a estratégia de alocação de partições e persiste essa configuração no ZooKeeper.
2. **Execução e Alocação:** O Controller, ao detectar a alteração no ZooKeeper, dispara uma máquina de estados para executar a alocação real das réplicas nos brokers.
 
Após o usuário enviar a solicitação, o Controller define o plano de alocação de partições e o salva no ZooKeeper. Em seguida, o Controller retorna uma confirmação ao usuário. Paralelamente, um thread dedicado no Controller monitora o ZooKeeper. Ao identificar uma mudança, ele inicia assincronamente o processo de alocação de réplicas, definindo quais brokers atuarão como líderes e quais como seguidores.

Estratégia de Alocação de Partições
-----------------------------------

O processo de definição da estratégia de alocação de partições ocorre em três etapas:

1. O usuário envia a requisição de criação de tópico ao Controller.
2. O Controller elabora a estratégia de alocação de partições para o novo tópico.
3. O Controller persiste essa estratégia no ZooKeeper.
 
A etapa mais detalhada é a segunda: a estratégia de alocação. A lógica central reside no arquivo `scala/kafka/admin/AdminUtils.scala`, com distinções entre cenários com e sem racks. Focaremos na estratégia sem racks.

### Alocação de Líderes de Partição

A alocação dos líderes de partição é gerenciada pela função `assignReplicasToBrokersRackUnaware`. O método opera da seguinte forma:

 
    ```

private def assignReplicasToBrokersRackUnaware(
    nPartitions: Int,
    replicationFactor: Int,
    brokerList: Seq[Int],
    fixedStartIndex: Int = -1,
    startPartitionId: Int = -1
): Map[Int, Seq[Int]] = {
    val ret = mutable.Map[Int, Seq[Int]]()
    val brokerArray = brokerList.toArray
    // Índice inicial aleatório para o primeiro broker (líder)
    val startIndex = if (fixedStartIndex >= 0) fixedStartIndex else rand.nextInt(brokerArray.length)
    var currentPartitionId = math.max(0, startPartitionId)
    var nextReplicaShift = if (fixedStartIndex >= 0) fixedStartIndex else rand.nextInt(brokerArray.length)

    for (_ <- 0 until nPartitions) {
        // Ajusta o deslocamento para seguidores a cada 'nBrokers' partições
        if (currentPartitionId > 0 && currentPartitionId % brokerArray.length == 0) {
            nextReplicaShift += 1
        }
        // Determina o índice do broker para o líder da partição atual
        val firstReplicaIndex = (currentPartitionId + startIndex) % brokerArray.length
        val replicaBuffer = mutable.ArrayBuffer(brokerArray(firstReplicaIndex))

        // Adiciona os índices dos seguidores
        for (j <- 0 until replicationFactor - 1) {
            replicaBuffer += brokerArray(replicaIndex(firstReplicaIndex, nextReplicaShift, j, brokerArray.length))
        }
        ret.put(currentPartitionId, replicaBuffer)
        currentPartitionId += 1
    }
    ret
}
    

Resumidamente, a estratégia de alocação de líderes é:

1. Um broker é selecionado aleatoriamente para ser o líder da Partição 0.
2. As partições subsequentes são distribuídas sequencialmente pelos brokers na lista.
3. Ao atingir o final da lista de brokers, o ciclo recomeça do início.

Por exemplo, com 5 brokers (1000 a 1004) e Partição 0 liderada pelo broker 1000:

Se a Partição 0 for liderada pelo broker 1002:

O objetivo é distribuir os líderes de partição entre os brokers para evitar sobrecarga (hotspots). No entanto, essa estratégia não considera a alocação de líderes de outros tópicos, o que pode levar a uma concentração de líderes no mesmo broker se muitos tópicos de partição única forem criados.

Alocação de Seguidores de Partição

A alocação dos seguidores deve atender a dois requisitso:

• Alta Disponibilidade: Distribuir seguidores aleatoriamente entre brokers diferentes para garantir que, em caso de falha do líder, um seguidor possa assumir.
• Balanceamento: Evitar que um único broker hospede um número desproporcional de réplicas.

A função replicaIndex é crucial para a alocação de seguidores:

```

private def replicaIndex( firstReplicaIndex: Int, secondReplicaShift: Int, replicaIndex: Int, nBrokers: Int ): Int = { // Calcula um deslocamento que garante que o seguidor não seja o líder e distribui as réplicas val shift = 1 + (secondReplicaShift + replicaIndex) % (nBrokers - 1) // Calcula o índice do broker para o seguidor (firstReplicaIndex + shift) % nBrokers }

  

A lógica de alocação de seguidores é:

1. Um ponto de partida aleatório (`secondReplicaShift`) é escolhido para o primeiro seguidor.
2. Seguidores subsequentes são posicionados sequencialmente a partir desse ponto.
3. O cálculo garante que um seguidor não seja alocado no mesmo broker do líder. Caso contrário, ele é deslocado para o próximo broker disponível.
4. A cada `nBrokers` partições, o `secondReplicaShift` é incrementado, alterando o ponto de partida para os seguidores das partições seguintes, promovendo maior dispersão.
 
Considerando o exemplo anterior (líderes distribuídos como na primeira tabela) e `replicationFactor = 2`:

 | Broker | 1000 | 1001 | 1002 | 1003 | 1004 |
|---|---|---|---|---|---|
| **Líder** | 0 | 1 | 2 | 3 | 4 |
|  | 5 | 6 | 7 | 8 | 9 |
| **Seguidor 1** | 1 | 2 | 3 | 4 | 0 |
|  | 6 | 7 | 8 | 9 | 5 |
| **Seguidor 2** | 4 | 0 | 1 | 2 | 3 |
|  | 9 | 5 | 6 | 7 | 8 |

- **Partição 1:** Líder em 1001. Seguidores são alocados a partir de um ponto inicial aleatório. Se o cálculo inicial apontar para 1000, e 1001 já for o líder, o próximo disponível é 1002.
- **Partição 3:** Líder em 1003. Seguidorse iniciam a partir de 1002, resultando em 1002 e 1004.
- **Partição 7:** O deslocamento `secondReplicaShift` é incrementado, alterando a base para os seguidores, resultando em 1003 e 1004.
 
Essa complexidade visa evitar que, após a falha de um líder, todas as suas réplicas sejam automaticamente promovidas para um único broker, o que poderia criar um novo ponto de contenção de tráfego.

### Alocação Manual de Estratégia

É possível definir manualmente a alocação de réplicas:

 
    ```

bash kafka-topics.sh \
--bootstrap-server localhost:9092 \
--create --topic meuTopico \
--replica-assignment 1000,1001,1002,1003,1004
    

Este comando criaria o tópico meuTopico com 5 partições, cada uma com uma réplica em um broker diferente (1000 a 1004).

Kafka também suporta estratégias de alocação com prioridade para racks, visando distribuir réplicas entre diferentes data centers para maior resiliência.

Máquinas de Estado

Após a definição da estratégia de alocação, o Controller persiste essa configuração no ZooKeeper (em /topics/nomeDoTopico) e retorna o sucesso ao usuário. A criação real é concluída assincronamente através de duas máquinas de estado:

• PartitionStateMachine: Gerencia o ciclo de vida das partições.
• ReplicaStateMachine: Gerencia o ciclo de vida das réplicas.

O ponto de entrada para o processamento de mudanças no ZooKeeper é kafka.controller.KafkaController#processTopicChange.

PartitionStateMachine

Define os estados de uma partição:

• NewPartition: Estado inicial, transitório para OnlinePartition.
• OnlinePartition: Partição ativa e pronta para serviço.
• OfflinePartition: Partição indisponível (ex: durante exclusão).
• NonExistentPartition: Estado inicial antes da criação.

No contexto de criação de tópico, as transições são: NonExistentPartition -> NewPartition -> OnlinePartition.

ReplicaStateMachine

Define os estados de uma réplica. Para a criação de tópicos, os estados relevantes são NewReplica, OnlineReplica e NonExistentReplica. O foco aqui é nas transições para NewReplica e OnlineReplica.

Fluxo de Transição de Estado

Todas as operações do Controller são executadas serialmente em um único thread.

Antes da transição principal, dois passos são executados:

1. Geração de Topic ID: Um identificador único é gerado para o tópico usando org.apache.kafka.common.Uuid#randomUuid.
2. Leitura da Estratégia de Alocação: A estratégia definida no ZooKeeper (/brokers/topics/nomeDoTopico) é carregada para o cache do Controller (kafka.controller.ControllerContext#partitionAssignments).

O método principal kafka.controller.KafkaController#onNewPartitionCreation executa as seguintes etapas:

1. partitionStateMachine: Define o estado para NewPartition.

2. replicaStateMachine: Define o estado para NewReplica.

3. partitionStateMachine: Define o estado para OnlinePartition.

4. replicaStateMachine: Define o estado para OnlineReplica.

// kafka.controller.KafkaController#onNewPartitionCreation private def onNewPartitionCreation(newPartitions: Set[TopicPartition]): Unit = { info(s"New partition creation callback for ${newPartitions.mkString(",")}") partitionStateMachine.handleStateChanges(newPartitions.toSeq, NewPartition) replicaStateMachine.handleStateChanges(controllerContext.replicasForPartition(newPartitions).toSeq, NewReplica) partitionStateMachine.handleStateChanges(newPartitions.toSeq, OnlinePartition, Some(OfflinePartitionLeaderElectionStrategy(false))) replicaStateMachine.handleStateChanges(controllerContext.replicasForPartition(newPartitions).toSeq, OnlineReplica) }

  

#### Transição para `NewPartition` e `NewReplica`

Essas etapas envolvem a atualização dos caches internos do Controller (`partitionStates` e `replicaStates`) para refletir os novos estados iniciais.

#### Transição para `OnlinePartition`

Esta é a fase central da criação:

1. **Persistência no ZooKeeper:** O Controller cria caminhos no ZooKeeper para cada partição (ex: `/brokers/topics/nomeDoTopico/partitions/0`) e persiste as informações de líder e ISR (`In-Sync Replicas`) em `/brokers/topics/nomeDoTopico/partitions/0/state`.
2. **Cache de Liderança:** As informações de líder e ISR são carregadas no cache do Controller (`kafka.controller.ControllerContext#partitionLeadershipInfo`).
3. **Preparação para Sincronização:** As alterações de metadados são registradas em `kafka.controller.AbstractControllerBrokerRequestBatch#leaderAndIsrRequestMap` para serem enviadas aos brokers.
4. **Atualização de Estado:** O estado da partição no cache do Controller (`kafka.controller.ControllerContext#partitionStates`) é atualizado para `OnlinePartition`.
5. **Notificação aos Brokers:** Uma requisição `ApiKeys.LEADER_AND_ISR` é enviada aos brokers relevantes. Ao receberem essa requisição, os brokers executam as operações para se tornarem líderes ou seguidores.
 
#### Transição para `OnlineReplica`

Similar à transição de partição, o estado das réplicas no cache do Controller (`kafka.controller.ControllerContext#replicaStates`) é atualizado para `OnlineReplica`.

Com a conclusão dessas etapas, o tópico é considerado oficialmente criado e pronto para uso no Kafka.