Kafka 流 :有没有办法在写入另一个主题时忽略主题分区中的特定偏移量

Question

背景：我在生成 to prod 主题时使用错误的 avro 模式注册表，结果 kafka 连接由于具有错误模式 id.So 的消息而关闭，作为恢复计划，我们希望将 prod 主题中的消息复制到测试主题，然后将良好的消息写入 HDFS。但是，在从 prod 主题读取时，我们面临着某些偏移量具有错误架构 ID 的问题。有没有办法在写入另一个主题时忽略此类偏移量。

Exception in thread "StreamThread-1" 
org.apache.kafka.streams.errors.StreamsException: Failed to deserialize value 
for record. topic=xxxx, partition=9, offset=1259032
Caused by: org.apache.kafka.common.errors.SerializationException: Error 
retrieving Avro schema for id 600
Caused by: 
io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException: 
Schema not found io.confluent.rest.exceptions.RestNotFoundException: Schema not found
io.confluent.rest.exceptions.RestNotFoundException: Schema not found

{代码}

Answer 1

您可以更改反序列化异常处理程序以跳过这些记录，如文档中所述：https://docs.confluent.io/current/streams/faq.html#handling-corrupted-records-and-deserialization-errors-poison-pill-records

即，您通过参数default.deserialization.exception.handler在配置中设置LogAndContinueExceptionHandler。