Configurando propriedades do consumidor

Existem muitas definições de configuração para o consumidor que controlam aspectos de seu comportamento. As configurações a seguir são algumas das mais importantes.

Muitas outras definições de configuração estão disponíveis, mas leia a Apache Kafka antes de começar a experimentar com elas

Grupos de consumidores

Um grupo de consumidores é aquele que coopera para consumir mensagens de um ou mais tópicos. Todos os consumidores em um grupo usam o mesmo valor para a configuração de group.id. Caso você precise de mais de um consumidor para lidar com sua carga, é possível executar múltiplos consumidores no mesmo grupo de consumidores. Mesmo que você precise apenas de um consumidor, é comum também especificar um valor para group.id.

Cada grupo de consumidores possui um servidor no cluster que é chamado de coordenador responsável por atribuir partições aos consumidores no grupo. Essa responsabilidade é propagada entre os servidores no cluster para balancear a carga. A designação de partições para os consumidores pode mudar a cada rebalanceamento de grupo.

Quando um consumidor se associa a um grupo de consumidores, ele descobre o coordenador do grupo. Em seguida, o consumidor informar ao coordenador que deseja se associar ao grupo e o coordenador inicia um rebalanceamento das partições em todo o grupo para incluir o novo membro.

As mensagens de uma única partição são processadas por somente um consumidor em cada grupo. Esse mecanismo assegura que as mensagens em cada partição sejam processadas em ordem. Consulte o diagrama a seguir para obter um exemplo em que um tópico contém três partições e um grupo de consumidores, que está consumindo esse tópico, contém dois consumidores Um consumidor no grupo recebe duas partições e o outro consumidor recebe uma partição.

Quando uma das mudanças a seguir ocorre em um grupo de consumidores, o grupo rebalanceia deslocando a atribuição de partições aos membros do grupo para acomodar a mudança:

• Um consumidor se associa ao grupo.
• Um consumidor sai do grupo.
• Um consumidor não mais é considerado em tempo real pelo coordenador.
• Novas partições são incluídas em um tópico existente.

Se você tiver um grupo de consumidores que é reequilibrado, saiba que qualquer consumidor que tenha saído do grupo terá seus commits rejeitados até que ele volte ao grupo. Nesse caso, o consumidor precisa unir-se novamente ao grupo, em que ele pode ser designado a uma partição diferente daquela que ele estava consumindo anteriormente.

Atividade do consumidor

O Kafka detecta automaticamente os consumidores com falha para que ele possa designar novamente partições para consumidores ativos. Ele usa dois mecanismos: pesquisa e pulsação.

Se o lote de mensagens retornadas por Consumer.poll(...) for grande ou o processamento for demorado, o atraso antes de poll() ser chamado novamente pode ser significativo ou imprevisível. Em alguns casos, é necessário configurar um intervalo de pesquisa longo máximo para que os consumidores não sejam removidos de seus grupos só porque o processamento de mensagens está demorando um pouco. Se esse mecanismo for o único disponível, o tempo que leva para detectar um consumidor fracassado também é longo.

Para tornar a atividade do consumidor mais fácil de lidar, uma pulsação em segundo plano foi incluída no Kafka 0.10.1. O coordenador do grupo espera que os membros do grupo enviem pulsações regulares para indicar que eles permanecem ativos. Um encadeamento de pulsação de plano de fundo é executado no consumidor e envia pulsações regulares para o coordenador. Se o coordenador não receber uma pulsação de um membro do grupo dentro do tempo limite de sessão, o coordenador removerá o membro do grupo e iniciará um rebalanceamento do grupo. O tempo limite da sessão pode ser muito menor do que o intervalo máximo de sondagem, de modo que o tempo necessário para detectar um consumidor com falha pode ser curto, mesmo que o processamento da mensagem demore muito.

É possível configurar o intervalo máximo de pesquisa usando a propriedade max.poll.interval.ms e o tempo limite de sessão usando a propriedade session.timeout.ms. Não é necessário usar essas configurações a menos que demore mais de 5 minutos para processar um lote de mensagens.

Gerenciando deslocamentos

Para cada grupo de consumidores, o Kafka mantém o deslocamento comprometido para cada partição que é consumida. Quando um consumidor processa uma mensagem, ele não a remove da partição. Em vez disso, ele apenas atualiza seu deslocamento atual usando um processo que é chamado confirmando o deslocamento.

O Event Streams retém as informações de deslocamento confirmadas por 7 dias.

Controlando a velocidade de consumo de mensagem

Se você tiver problemas com o tratamento de mensagens causado por inundação de mensagens, poderá definir uma opção de consumidor para controlar a velocidade do consumo de mensagens. Use fetch.max.bytes e max.poll.records para controlar quantos dados uma chamada para poll() pode retornar.

Manipulando o rebalanceamento do consumidor

Quando os consumidores são adicionados ou removidos de um grupo, ocorre um reequilíbrio do grupo, e os consumidores não podem consumir mensagens. Isso faz com que todos os consumidores em um grupo de consumidores fiquem indisponíveis por um curto período.

Se você for notificado com o retorno de chamada "on partitions revoked", use um ConsumerRebalanceListener para confirmar manualmente os deslocamentos (se não estiver usando o auto-commit) e para pausar o processamento adicional até que seja notificado do rebalanceamento bem-sucedido usando o retorno de chamada "on partition assigned".

Fragmentos de código

Esses fragmentos de código estão em um alto nível para ilustrar os conceitos envolvidos. Para obter exemplos completos, consulte as amostras do Event Streams em GitHub.

Para conectar um consumidor ao Event Streams, é necessário criar credenciais de serviço. Para obter informações sobre como obter essas credenciais, consulte Conectando ao Event Streams.

No código do consumidor, primeiro é necessário construir o conjunto de propriedades de configuração Todas as conexões com Event Streams são protegidas pelo uso de TLS e autenticação de senha de usuário, portanto, você precisa de pelo menos essas propriedades. Substitua BOOTSTRAP_ENDPOINTS, USER e PASSWORD por aqueles de suas próprias credenciais de serviço:

Properties props = new Properties();
 props.put("bootstrap.servers", BOOTSTRAP_ENDPOINTS);
 props.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"USER\" password=\"PASSWORD\";");
 props.put("security.protocol", "SASL_SSL");
 props.put("sasl.mechanism", "PLAIN");
 props.put("ssl.protocol", "TLSv1.2");
 props.put("ssl.enabled.protocols", "TLSv1.2");
 props.put("ssl.endpoint.identification.algorithm", "HTTPS");

Para consumir mensagens, você também precisa especificar desserializadores para as chaves e os valores, como no exemplo a seguir.

 props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
 props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

Esses desserializadores devem corresponder aos serializadores usados pelos produtores..

Em seguida, use um KafkaConsumer para consumir mensagens, em que cada mensagem é representada por um ConsumerRecord. A maneira mais comum de consumir mensagens é colocar o consumidor em um grupo de consumidores configurando o ID do grupo e, em seguida, chamando subscribe() para obter uma lista de tópicos. São designadas ao consumidor algumas partições para consumir, no entanto, se houver mais consumidores no grupo do que partições no tópico, é possível que nenhuma partição seja designada ao consumidor. Em seguida, chame poll() em um loop, recebendo um lote de mensagens para processar, em que cada mensagem é representada por um ConsumerRecord.

props.put("group.id", "G1");
Consumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("T1"));  // T1 is the topic name
while (true) {
   ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
   for (ConsumerRecord<String, String> record : records)
     System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
}

Esse loop de consumo é executado para sempre, mas pode ser interrompido por outro thread chamando Consumer.wakeup() para obter um desligamento organizado.

Para confirmar deslocamentos manualmente, primeiramente é necessário definir a configuração do enable.auto.commit como false. Em seguida, use Consumer.commmitSync() ou Consumer.commitAsync() para atualizar o deslocamento confirmado do consumidor periodicamente. Para simplificar, este exemplo processa os registros para cada partição e confirma o último deslocamento separadamente.

props.put("group.id", "G1");
props.put("enable.auto.commit", "false");
Consumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("T1"));
try {
  while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (TopicPartition tp : records.partitions()) {
      List<ConsumerRecord<String, String>> partRecords = records.records(tp);
      long lastOffset = 0;
      for (ConsumerRecord<String, String> record : partRecords) {
        System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
        lastOffset = record.offset();
      }
      // having processed all the records in the above loop, we commit the partition's offset to 1 more than the last offset
      consumer.commitSync(Collections.singletonMap(tp, new OffsetAndMetadata(lastOffset + 1)));
    }
  }
}
finally {
  consumer.close();
}

Manipulação de exceção

Qualquer aplicativo robusto que usa o cliente Kafka precisa manipular exceções para determinadas situações esperadas. Em alguns casos, as exceções não são lançadas diretamente, porque alguns métodos são assíncronos e entregam seus resultados usando um Future ou um retorno de chamada. Verifique o código de exemplo no GitHub que mostra exemplos completos.

Manipule a seguinte lista de exceções em seu código: