目前，我设置了2个运行JDBC Sink Connector的独立连接器来摄取从生产者生成的主题以读取到数据库中。有时，我在日志中看到错误，这导致生成的消息无法存储到数据库中。我经常看到的错误是

Caused by: org.apache.kafka.common.errors.SerializationException: Error retrieving Avro schema for id:11
Caused by: io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException: Subject 'topic-io..models.avro.Topic' not found; error code404

这是真的，因为TopicRecordName不应该指向这个主题，而是我指向的另一个主题，它只应该指向model. avro.Topic

我想知道如果这种情况经常发生，是否有一种方法可以在消息产生后将这些产生的记录/消息重新摄取到数据库中。例如，如果消息在12am-1am期间产生，并且日志中出现了某种错误并且在该时间范围内未能使用这些消息，则配置或偏移可以通过将其重新摄取到数据库来恢复它。错误是由于模式注册表链接无法读取/链接到正确的模式链接。它失败了，因为它读取了不正确的工作文件，因为我的一个工作文件有一个value.converter.value.subject.name. policy=io.confluent.kafka.seralizers.sub.TopicRecordNameStrategy而另一个连接器没有读取该subjectName。

目前，我设置consumer.auto. offset.reset=最早开始读取消息。有没有办法像文件一样取回这些数据，我可以恢复这些数据，因为我正在部署到生产环境中，并且必须始终将数据消耗到数据库中而不会出现任何错误。