我正在使用简单的Kafka客户端API。据我所知,消费消息有两种方式,订阅主题和将分区分配给消费者。
但是,第一种方法不起作用。消费者poll()将永远挂起。它只适用于assign。
// common config for consumer
Map<String, Object> config = new HashMap<>();
config.put("bootstrap.servers", bootstrap);
config.put("group.id", KafkaTestConstants.KAFKA_GROUP);
config.put("enable.auto.commit", "true");
config.put("auto.offset.reset", "earliest");
config.put("key.deserializer", StringDeserializer.class.getName());
config.put("value.deserializer", StringDeserializer.class.getName());
StringDeserializer deserializer = new StringDeserializer();
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(config, deserializer, deserializer);
// subscribe does not work, poll() hangs
consumer.subscribe(Arrays.asList(KafkaTestConstants.KAFKA_TOPIC));下面是通过分配分区来工作的代码。
// assign works
TopicPartition tp = new TopicPartition(KafkaTestConstants.KAFKA_TOPIC, 0);
List<TopicPartition> tps = Arrays.asList(tp);
consumer.assign(tps);因为我想要利用自动提交特性,根据这个post,这个特性应该只与消费者组管理一起使用。为什么subscribe()不起作用?
发布于 2020-09-08 13:55:23
我也面临着同样的问题。我使用的是kafka_2.12 jar版本,当我将其降级为kafka_2.11时,它可以工作。
https://stackoverflow.com/questions/42707131
复制相似问题