Konsumsilah pesan yang sama lagi jika pemrosesan pesan gagal

Saya menggunakan Confluent.Kafka .NET klien versi 1.3.0. Saya mengikuti dokumen :

var consumerConfig = new ConsumerConfig
{
    BootstrapServers = "server1, server2",
    AutoOffsetReset = AutoOffsetReset.Earliest,
    EnableAutoCommit = true,
    EnableAutoOffsetStore = false,
    GroupId = this.groupId,
    SecurityProtocol = SecurityProtocol.SaslPlaintext,
    SaslMechanism = SaslMechanism.Plain,
    SaslUsername = this.kafkaUsername,
    SaslPassword = this.kafkaPassword,
};

using (var consumer = new ConsumerBuilder<Ignore, string>(consumerConfig).Build())
{
    var cancellationToken = new CancellationTokenSource();
    Console.CancelKeyPress += (_, e) =>
    {
        e.Cancel = true;
        cancellationToken.Cancel();
    };

    consumer.Subscribe("my-topic");
    while (true)
    {
        try
        {
            var consumerResult = consumer.Consume();
            // process message
            consumer.StoreOffset(consumerResult);
        }
        catch (ConsumeException e)
        {
            // log
        }
        catch (KafkaException e)
        {
            // log
        }
        catch (OperationCanceledException e)
        {
            // log
        }
    }
}

Masalahnya adalah bahwa bahkan jika saya berkomentar keluar baris consumer.StoreOffset(consumerResult);, saya terus mendapatkan pesan yang tidak dikonsumsi berikutnya saat saya Mengkonsumsinya berikutnya , yaitu offset terus meningkat yang tampaknya tidak seperti yang diklaim oleh dokumentasi, yaitu setidaknya satu pengiriman .

Bahkan jika saya mengatur EnableAutoCommit = falsedan menghapus 'EnableAutoOffsetStore = false' dari konfigurasi, dan ganti consumer.StoreOffset(consumerResult)dengan consumer.Commit(), saya masih melihat perilaku yang sama, yaitu bahkan jika saya berkomentar Commit, saya masih terus mendapatkan pesan yang tidak digunakan berikutnya.

Saya merasa seperti kehilangan sesuatu yang mendasar di sini, tetapi tidak dapat menemukan apa. Bantuan apa pun dihargai!

c# apache-kafka kafka-consumer-api havij
sumber

Pesan-pesan telah dikembalikan ke aplikasi dari sudut pandang kafka sehingga ketika Anda mengkomitnya disimpan sebagai offset yang dikomit terakhir tetapi mengkonsumsi akan terus mengembalikan pesan berikutnya apakah Anda telah mengkonsumsi atau belum. Apa harapan Anda di sini? Bisakah Anda jelaskan apa yang Anda harapkan terjadi sebelum / sesudah komit dan konsumsi?

Sagar Veeram

Pesan tidak diambil ulang sampai Anda menggunakan berusaha mengimbangi. Ini akan mempengaruhi konsumsi dan pesan akan dikembalikan dari pencarian ofset.

Sagar Veeram

@ user2683814 Dalam posting saya, saya menyebutkan dua skenario tergantung pada apa EnableAutoCommityang diatur. Katakanlah kita punya EnableAutoCommit = false, dan ketika saya Consume, saya mendapatkan kembali pesan dengan offset 11. Saya berharap untuk terus mendapatkan pesan yang sama dengan offset 11 berulang-ulang jika memproses pesan terus melempar dan karenanya tidak ada panggilan untuk Commitdibuat.

havij

Tidak, bukan itu masalahnya. Anda tidak dapat mengontrol apa yang akan disurvei ( Consume) dengan menggunakan Commitsetelah Anda sudah Subscribeke topik .. Kafka (seperti dalam lib klien) di belakang layar mempertahankan semua offset yang telah dikirim ke aplikasi di Consumedan akan mengirimkannya secara linear. Jadi untuk memproses ulang pesan seperti dalam skenario kegagalan, Anda harus melacaknya dalam kode Anda dan berusaha mengimbangi dan mulai memproses pesan dan dan Anda juga harus tahu apa yang harus dilewati jika sudah diproses dalam permintaan sebelumnya. Saya tidak terbiasa dengan .net library tetapi seharusnya tidak terlalu penting karena ini adalah desain kafka.

Sagar Veeram

Saya pikir Anda harus menggunakan kombinasi berlangganan dan menetapkan dan mungkin memerlukan konsumen yang berbeda untuk mendukung kasus penggunaan Anda. Jika terjadi kegagalan, gunakan penetapan / ganti rugi untuk partisi topik dengan satu konsumen untuk memproses ulang pesan dan untuk pemrosesan normal gunakan konsumen lain dengan aliran langganan / konsumsi / komit.

Sagar Veeram

Konsumsilah pesan yang sama lagi jika pemrosesan pesan gagal

Jawaban: