Apache Flink KafkaSource 没有设置 group.id答案

【问题标题】：Apache Flink KafkaSource doesnt set group.idApache Flink KafkaSource 没有设置 group.id
【发布时间】：2022-01-31 20:26:40
【问题描述】：

我有一个简单的流执行配置为：

val config: Configuration = new Configuration()
config.setString("taskmanager.memory.managed.size", "4g")
config.setString("parallelism.default", "4")
val env: StreamExecutionEnvironment = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(config)

env
  .fromSource(KafkaSource.builder[String]
    .setBootstrapServers("node1:9093,node2:9093,node3:9093")
    .setTopics("example-topic")
    //.setProperties(kafkaProps) // didn't work
    .setProperty("security.protocol", "SASL_SSL")
    .setProperty("sasl.mechanism", "GSSAPI")
    .setProperty("sasl.kerberos.service.name", "kafka")
    .setProperty("group.id","groupid-test")
    //.setGroupId("groupid-test") // didn't work
    .setStartingOffsets(OffsetsInitializer.earliest)
    .setProperty("partition.discovery.interval.ms", "60000") // discover part
    .setDeserializer(KafkaRecordDeserializationSchema.valueOnly(classOf[StringDeserializer]))
    .build,
    WatermarkStrategy.noWatermarks[String],
    "example-input-topic"
  )
  .print

env.execute("asdasd")

我的flink版本是：1.14.2

我的 kafka 在 cloudera 上运行。卡夫卡版本：2.2.1-cdh6.3.2

能够使用来自 Kafka 的记录。但它没有为主题设置 groupid。有人有什么想法吗？

【问题讨论】：

标签： apache-kafka streaming apache-flink flink-streaming

【解决方案1】：

从 Flink 1.14.0 开始，group.id 是一个可选值。见https://issues.apache.org/jira/browse/FLINK-24051。如果您想指定一个，您可以设置自己的值。您可以从随附的 PR 中看到这是如何在之前设置为 https://github.com/apache/flink/pull/17052/files#diff-34b4ff8d43271eeac91ba17f29b13322f6e0ff3d15f71003a839aeb780fe30fbL56

【讨论】：

非常感谢，我不知道。