apache-kafka - 更改 Kafka 中的订阅主题

Question

我想在 Kafka 中测试更改订阅主题，所以我这样编码。

for (int count = 0; count < 30000; count++) {
        consumer.subscribe(Collections.singletonList("test1"));
        consumer.unsubscribe();

        consumer.subscribe(Collections.singletonList("test2"));
        ConsumerRecords<String, Integer> records = consumer.poll(100);
        for (ConsumerRecord<String, Integer> record : records) {
            String msgString = String.format("key:%s, value:%d, partition:%d offset:%d",
                    record.key(), record.value(), record.partition(), record.offset());
            System.out.println(msgString);
        }
        try {
            Thread.sleep(10);
        } catch (InterruptedException ex) {
            ex.printStackTrace();
        }
    }

但是数据要么不被轮询，要么只被部分轮询。我该如何解决？

标签： apache-kafkakafka-consumer-api