我们通过spring云流集成使用kafka流。我通过设置
spring.cloud.stream.kafka.streams.binder.configuration.replication.factor=${REPL_FACTOR}它适用于kafka streams内部使用的大多数重新分区/更改日志主题。但是，此设置似乎对通过手动创建的状态存储更改日志主题没有影响 Materialized#as(StoreSupplier) . 对于这些主题，我仍然可以看到复制因子设置为默认值1。也不可能使用 Materialized#withLoggingEnabled(Map<String, String>) 因为这只接受主题级配置( replication.factor 是流配置）。这是Kafka流中已知的错误吗？我什么也找不到。如果是这样，是否有一个解决方法来增加这些changelog主题的复制因子？
我们在代理端使用kafka v2.3.1，在客户端使用kafka 2.5.0。
1条答案
按热度按时间wz8daaqr1#
从版本2.4开始
AdminClient现在可以在中将复制因子设置为-1NewTopic，意思是default.replication.factor创建主题时应使用-kip-464。然而，Kafka流似乎目前没有使用这一功能；有一个开放的问题，这是Kafka-8531。
可以使用设置内部主题的复制因子
https://kafka.apache.org/documentation/#replication.factor
流处理应用程序创建的更改日志主题和重新分区主题的复制因子。
由于您是通过binder配置来设置的，所以它应该按预期工作。
编辑
您使用的是什么版本的SpringCloudStream？我刚刚用3.0.8进行了测试，结果和预期的一样。
2020-10-15 12:03:55601错误[kafka stre]o.a.k.s.p.i.streamthread:673 - 流线程[kafka-streams-inventory-processor-b8d07a5a-f3c4-476a-a265-119163d2acb7-streamthread-1]在处理过程中遇到以下意外的kafka异常，这通常表示流内部错误：org.apache.kafka.streams.errors.streamsexception:无法创建主题kafka streams inventory processor inventory counts changelog。
原因：org.apache.kafka.common.errors.invalidreplicationfactorexception:复制因子：3大于可用代理：1。