Кафка потребительская Java с несколькими темами

Question

у нас есть одна группа потребителей и три темы, все три темы имеют разные схемы.создал одного потребителя с циклом for, передавая каждую тему за раз и опрашивая ее, обрабатывая и фиксируя вручную.Используемый метод является общим для потребителя, и в течение цикла я подписываюсь по одной теме за раз и обрабатываю данные.Я вижу случайное отставание потребителя, хотя у темы есть данные, которые мой потребитель не получает из темы, а иногда извлекает.Когда я работаю с одной темой, а не перебираю три темы, она работает, но не может воспроизвести.нужна помощь, чтобы отладить проблему и воспроизвести то же самое,

Answer 1

Вместо того, чтобы зацикливать три темы в одном методе, вы можете создать скелетный поток, подобный такому, который будет использовать любую тему. См. Примеры здесь

Я не могу сказать, если это "решит" проблему, но попытка использовать темы с разными схемами в одном приложении обычно не является масштабируемым шаблоном, но этоне очень понятно, что ты пытаешься сделать.

class ConsumerThread extends Thread {

    KafkaConsumer consumer;
    AtomicBoolean stopped = new AtomicBoolean();

    ConsumerThread(Properties props, String subscribePattern) {
        this.consumer = new KafkaConsumer...
        this.consumer.subscribe(subscribePattern);
    } 

    @Override
    public void run() {
        while (!this.stopped.get()) {
            ... records = this.consumer.poll(100);
            for ( ... each record ... ) {
               // Process record
            } 
        }
    }

    public void stop() {
        this.stopped.set(true);
    }
}

Не предназначено для производственного класса

Затем независимо от трех потребителей.

new ConsumerThread("t1").start();
new ConsumerThread("t2").start();
new ConsumerThread("t3").start();

Примечание : KafkaConsumer не является поточно-ориентированным.