Transazioni in Apache Kafka per Hub eventi di Azure

Questo articolo fornisce informazioni dettagliate su come usare l'API transazionale Apache Kafka con Hub eventi di Azure.

Panoramica

Hub eventi fornisce un endpoint Kafka che può essere usato dalle applicazioni client Kafka esistenti come alternativa all'esecuzione del proprio cluster Kafka. Hub eventi funziona con numerose applicazioni Kafka esistenti. Per altre informazioni, vedere Hub eventi per Apache Kafka.

Questo documento è incentrato su come usare l'API transazionale di Kafka con Hub eventi di Azure senza problemi.

Transazioni in Apache Kafka

Negli ambienti nativi del cloud, le applicazioni devono essere rese resilienti alle interruzioni di rete e ai riavvii e agli aggiornamenti dello spazio dei nomi. Le applicazioni che richiedono garanzie di elaborazione rigorose devono usare un framework transazionale o un'API per garantire che tutte le operazioni vengano eseguite o che nessuno sia in modo che l'applicazione e lo stato dei dati siano gestiti in modo affidabile. Se il set di operazioni ha esito negativo, è possibile riprovare in modo affidabile in modo atomico per garantire le garanzie di elaborazione corrette.

Apache Kafka fornisce un'API transazionale per garantire questo livello di garanzia di elaborazione nello stesso set di argomenti o partizioni diverso.

Le transazioni si applicano ai casi seguenti:

• Producer transazionali.
• Esattamente una volta la semantica di elaborazione.

Producer transazionali

I producer transazionali assicurano che i dati vengano scritti in modo atomico in più partizioni in diversi argomenti. I producer possono avviare una transazione, scrivere in più partizioni nello stesso argomento o in argomenti diversi e quindi eseguire il commit o interrompere la transazione.

Per assicurarsi che un producer sia transazionale, enable.idempotence deve essere impostato su true per assicurarsi che i dati vengano scritti esattamente una volta, evitando così duplicati sul lato di invio . Inoltre, transaction.id deve essere impostato per identificare in modo univoco il producer.

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

Dopo l'inizializzazione del producer, la chiamata seguente garantisce che il producer si registri con il broker come producer transazionale:

    producer.initTransactions();

Il producer deve quindi avviare una transazione in modo esplicito, eseguire operazioni di invio tra diversi argomenti e partizioni come di consueto e quindi eseguire il commit della transazione con la chiamata seguente:

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

Se la transazione deve essere interrotta a causa di un errore o di un timeout, il producer può chiamare il abortTransaction() metodo .

	producer.abortTransaction();

Semantica esattamente una volta

Esattamente una volta la semantica si basa sui producer transazionali aggiungendo consumer nell'ambito transazionale dei producer, in modo che ogni record sia garantito di essere letto, elaborato e scritto esattamente una volta.

Viene creata un'istanza del producer transazionale :

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

    producer.initTransactions();

Il consumer deve quindi essere configurato per leggere solo messaggi non transazionali o messaggi transazionali di cui è stato eseguito il commit impostando la proprietà seguente:

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

Dopo aver creato un'istanza del consumer, può sottoscrivere l'argomento da cui devono essere letti i record:

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

Dopo che il consumer esegue il polling dei record dall'argomento di input, il producer inizia l'ambito transazionale all'interno del quale viene elaborato e scritto nell'argomento di output. Dopo aver scritto i record, viene creata la mappa aggiornata degli offset per tutte le partizioni. Il producer invia quindi questo mapping di offset aggiornato alla transazione prima di eseguire il commit della transazione.

In qualsiasi eccezione, la transazione viene interrotta e il producer ritenta nuovamente l'elaborazione in modo atomico.

	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();
        }
	}

Guida alla migrazione

Se si dispone di applicazioni Kafka esistenti che si vuole usare con Hub eventi di Azure, vedere la guida alla migrazione di Kafka per Hub eventi di Azure per raggiungere rapidamente il livello di esecuzione.

Passaggi successivi

Per altre informazioni su Hub eventi e Hub eventi per Kafka, vedere gli articoli seguenti:

• Guida alla risoluzione dei problemi di Apache Kafka per Hub eventi
• Domande frequenti - Hub eventi per Apache Kafka
• Guida per sviluppatori Apache Kafka per Hub eventi di Azure
• Configurazioni consigliate