Transações no Apache Kafka para Hubs de Eventos do Azure

Este artigo fornece detalhes sobre como usar a API transacional do Apache Kafka com os Hubs de Eventos do Azure.

Descrição geral

Os Hubs de Eventos fornecem um ponto de extremidade Kafka que pode ser usado por seus aplicativos cliente Kafka existentes como uma alternativa para executar seu próprio cluster Kafka. Os Hubs de Eventos funcionam com muitos dos seus aplicativos Kafka existentes. Para obter mais informações, consulte Hubs de eventos para Apache Kafka.

Este documento se concentra em como usar a API transacional do Kafka com os Hubs de Eventos do Azure perfeitamente.

Transações no Apache Kafka

Em ambientes nativos da nuvem, os aplicativos devem ser resilientes a interrupções de rede e reinicializações e atualizações de namespace. Os aplicativos que exigem garantias de processamento rigorosas devem utilizar uma estrutura transacional ou API para garantir que todas as operações sejam executadas ou nenhuma seja para que o aplicativo e o estado dos dados sejam gerenciados de forma confiável. Se o conjunto de operações falhar, elas podem ser tentadas novamente de forma confiável para garantir as garantias de processamento corretas.

O Apache Kafka fornece uma API transacional para garantir esse nível de garantias de processamento no mesmo ou em diferentes conjuntos de tópicos/partições.

As transações aplicam-se aos seguintes casos:

• Produtores transacionais.
• Exatamente uma vez processando semântica.

Produtores Transacionais

Os produtores transacionais garantem que os dados sejam gravados atomicamente em várias partições em diferentes tópicos. Os produtores podem iniciar uma transação, gravar em várias partições no mesmo tópico ou em tópicos diferentes e, em seguida, confirmar ou abortar a transação.

Para garantir que um produtor é transacional, enable.idempotence deve ser definido como true para garantir que os dados sejam gravados exatamente uma vez, evitando assim duplicações no lado do envio . Além disso, transaction.id deve ser definido para identificar exclusivamente o produtor.

    producerProps.put("enable.idempotence", "true");
    producerProps.put("transactional.id", "transactional-producer-1");
    KafkaProducer<String, String> producer = new KafkaProducer(producerProps);

Uma vez que o produtor é inicializado, a chamada abaixo garante que o produtor se registre com o corretor como um produtor transacional -

    producer.initTransactions();

O produtor deve então iniciar uma transação explicitamente, executar operações de envio em diferentes tópicos e partições como normal e, em seguida, confirmar a transação com a chamada abaixo –

    producer.beginTransaction();
	/*
        Send to multiple topic partitions.
    */
    producer.commitTransaction();

Se a transação precisar ser abortada devido a uma falha ou um tempo limite, o produtor pode chamar o abortTransaction() método.

	producer.abortTransaction();

Semântica exatamente uma vez

Exatamente uma semântica se baseia nos produtores transacionais, adicionando consumidores no escopo transacional dos produtores, de modo que cada registro tenha a garantia de ser lido, processado e escrito exatamente uma vez.

Primeiro, o produtor transacional é instanciado -

    producerProps.put("enable.idempotence", "true");
    producerProps.put("transactional.id", "transactional-producer-1");
    KafkaProducer<K, V> producer = new KafkaProducer(producerProps);

    producer.initTransactions();

Em seguida, o consumidor deve ser configurado para ler apenas mensagens não transacionais ou mensagens transacionais confirmadas, definindo a propriedade abaixo –

	consumerProps.put("isolation.level", "read_committed");
	KafkaConsumer <K,V> consumer = new KafkaConsumer<>(consumerProps);

Uma vez instanciado, o consumidor pode subscrever o tópico a partir do qual os registos devem ser lidos –

    consumer.subscribe(singleton("inputTopic"));

Depois que o consumidor sonda os registros do tópico de entrada, o produtor inicia o escopo transacional dentro do qual o registro é processado e gravado no tópico de saída. Uma vez que os registros são gravados, o mapa atualizado de deslocamentos para todas as partições é criado. Em seguida, o produtor envia esse mapa de deslocamento atualizado para a transação antes de confirmar a transação.

Em qualquer exceção, a transação é abortada e o produtor tenta novamente o processamento atomicamente.

	while (true) {
		ConsumerRecords records = consumer.poll(Long.Max_VALUE);
		producer.beginTransaction();
        try {
    		for (ConsumerRecord record : records) {
    			/*
                    Process record as appropriate
                */
                // Write to output topic
    	        producer.send(producerRecord(“outputTopic”, record));
    		}
    
            /*
                Generate the offset map to be committed.
            */
            Map <TopicPartition, OffsetAndMetadata> offsetsToCommit = new Hashap<>();
            for (TopicPartition partition : records.partitions()) {
                // Calculate the offset to commit and populate the map.
                offsetsToCommit.put(partition, new OffsetAndMetadata(calculated_offset))
            }
            
            // send offsets to transaction and then commit the transaction.
    		producer.sendOffsetsToTransaction(offsetsToCommit, group);
    		producer.commitTransaction();
        } catch (Exception e)
        {
            producer.abortTransaction();
        }
	}

Guia de Migração

Se você tiver aplicativos Kafka existentes que gostaria de usar com os Hubs de Eventos do Azure, consulte o guia de migração do Kafka para Hubs de Eventos do Azure para começar a funcionar rapidamente.

Próximos passos

Para saber mais sobre Hubs de Eventos e Hubs de Eventos para Kafka, consulte os seguintes artigos:

• Guia de solução de problemas do Apache Kafka para Hubs de Eventos
• Perguntas frequentes - Hubs de Eventos para Apache Kafka
• Guia do desenvolvedor do Apache Kafka para Hubs de Eventos do Azure
• Configurações recomendadas