在 Kafka 压缩主题上写入数据 - 如何设置记录键答案

【问题标题】：Writing data on a Kafka compacted topic - how to set the record key在 Kafka 压缩主题上写入数据 - 如何设置记录键
【发布时间】：2018-08-03 15:22:17
【问题描述】：

使用 spring 云数据流，我正在尝试使用 Kafka 压缩主题从数据库中复制数据。出于这个原因，我需要在将记录生成到主题（称为 foo）时设置记录的键。

kafka-topics --zookeeper localhost:2181 --create --topic foo --replication-factor=1 --partitions 3 --config cleanup.policy=compact 



stream create --name foo --deploy --definition "jdbc --spring.datasource.url=jdbc:postgresql://localhost:5432/chiodonia --spring.datasource.driver-class-name=org.postgresql.Driver --jdbc.max-rows-per-poll=10 --jdbc.query='select id, value from foo where seen is null' --jdbc.split=true --jdbc.update='update foo set seen=current_timestamp where id in (:id)' --trigger.fixed-delay=5 --trigger.time-unit=SECONDS | header-enricher --headers='recordkey=payload.id' > :foo --spring.cloud.stream.kafka.bindings.output.producer.messageKeyExpression=headers['recordkey']"

任何人都可以使用 Spring Cloud 数据流将记录的密钥设置到 kafka 中吗？

【问题讨论】：

标签： apache-kafka key record spring-cloud-dataflow

【解决方案1】：

要覆盖输出通道的绑定配置，您必须将其设置为从header-enricher 处理器向外到foo 主题的通道。

那就是：

stream create --name foo --deploy --definition "jdbc ...... | header-enricher --headers='recordkey=payload.id' --spring.cloud.stream.kafka.bindings. output.producer.messageKeyExpression=headers['recordkey'] > :foo"

【讨论】：