我正在使用kafka流媒体库进行流媒体应用。我想设置kafka消费者组id。然后,我将kafka流配置如下所示。
streamsCopnfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, "JoinTestApp");
streamsCopnfiguration.put(StreamsConfig.CLIENT_ID_CONFIG, "JonTestClientId1");
streamsCopnfiguration.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 10 * 1000);
streamsCopnfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, bootstreapServer);
streamsCopnfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
streamsCopnfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.Bytes().getClass().getName());
streamsCopnfiguration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
streamsCopnfiguration.put(StreamsConfig.consumerPrefix("group.id"), "groupId1");
// streamsCopnfiguration.put(ConsumerConfig.GROUP_ID_CONFIG, "groupId1");
但是,我的控制台日志kafka consumer configuration未设置group.id。只是流应用程序id。。
2019-03-19 17:17:03,206 [main] INFO org.apache.kafka.clients.consumer.ConsumerConfig - ConsumerConfig values:
auto.commit.interval.ms = 5000
auto.offset.reset = earliest
bootstrap.servers = [....]
check.crcs = true
client.dns.lookup = default
client.id = JonTestClientId111-StreamThread-1-consumer
connections.max.idle.ms = 540000
default.api.timeout.ms = 60000
enable.auto.commit = false
exclude.internal.topics = true
fetch.max.bytes = 52428800
fetch.max.wait.ms = 500
fetch.min.bytes = 1
group.id = JoinTestApp
heartbeat.interval.ms = 3000
interceptor.classes = []
internal.leave.group.on.close = false
我可以设置kafka stream consumer group.id吗??
1条答案
按热度按时间ia2d9nvy1#
不,你不能。为此,在Kafka流你需要使用
application.id
.Kafka河
application.id
用于将应用程序使用的资源与其他应用程序隔离开来。application.id
被用作Kafka消费者group.id
为了协调。所以你不能group.id
明确地。Kafka的官方文件,
application.id
也用于以下地方: