如何在两个消费者组之间进行 Kafka 同步

Question

在我的要求中，我有两个消费者组，一个组（主）只是获取数据并发送到其他服务器，如果发送到其他服务器失败，那么我需要重新加入（启动）其他消费者组（处理失败） .

在这种情况下，主组继续读取和重试，它会继续相同，当消息发送成功时需要通知其他消费者组（处理失败）。现在失败的处理应该从第一次失败到最后一次失败的地方开始发送。

public void StartMainstreamHandler() {
       StreamsBuilder builder = new StreamsBuilder();

        KStream<String, String> userStream = builder.stream("usertopic",Consumed.with(Serdes.String(), Serdes.String()));
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "main-streams-userstream");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "ALL my bootstrap servers);
        props.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 4);
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "500");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
        //consumer_timeout_ms
        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 2000);

        props.put("state.dir","/tmp/kafka/stat));


   userStream.peek((key,value)->System.out.println("key :"+key+" value :"+value));
   /* Send Data to other Server if Failed call other consumer */

KafkaStreams kafkaStreams = new KafkaStreams(builder.build(), props);


kafkaStreams.setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
            @Override
            public void uncaughtException(Thread t, Throwable e) {

                logger.error("Thread Name :" + t.getName() + " Error while processing:", e);
            }
        });


        kafkaStreams.cleanUp();
        kafkaStreams.start();
    }

其他消费者

public void StartFailstreamHandler() {
       StreamsBuilder builder = new StreamsBuilder();

        KStream<String, String> userStream = builder.stream("usertopic",Consumed.with(Serdes.String(), Serdes.String()));
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "failed-streams-userstream");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "ALL my bootstrap servers);
        props.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 4);
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "500");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
        //consumer_timeout_ms
        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 2000);

        props.put("state.dir","/tmp/kafka/stat));

     Wait('till get notfication from other consumer" ){
     userStream.peek((key,value)->System.out.println("key :"+key+" value :"+value));
     /* start sending */
     /* how to break when it is reached last offest */ 
   }

KafkaStreams kafkaStreams = new KafkaStreams(builder.build(), props);


kafkaStreams.setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
            @Override
            public void uncaughtException(Thread t, Throwable e) {

                logger.error("Thread Name :" + t.getName() + " Error while processing:", e);
            }
        });


        kafkaStreams.cleanUp();
        kafkaStreams.start();
    }

现在如何知道和同步第二个消费者的偏移细节，以在最后失败时完全停止（最后失败发生在主要消费者）

Answer 1

仅使用消息在集群之间同步消费者组是不可能的，因为消费者无法找到特定的“复制起点”。

您必须将额外的元数据存储到一边，例如复制开始的时间戳，可能将该信息嵌入到记录标头中（假设您的 Kafka 版本支持它}。否则，您或多或少是在盲目地复制数据以最低限度的交付保证（因此无论如何使用 MirrorMaker 可能会更好）。

MirrorMaker 2 或 Confluent Replicator 是目前仅有的两个可用于“同步”消费者组的选项，以进行主动-主动双集群设置