我会提出什么样的错误，所以我的消息最终出现在DLQ中

Question

我有一个可能令人尴尬的问题，但我有点迷路了。

我已经阅读了这篇文章，我认为Kafkaconnect如何处理DLQ非常全面。 https://www.confluent.io/blog/kafka-connect-deep-deep-dive-error handling-dead-letter-queues/

我将其设置在连接器上：

errors.tolerance: all
errors.log.enable: true
errors.log.include.messages: false
errors.deadletterqueue.topic.name: com.kafkaconnect.jdbc.dlq
errors.deadletterqueue.topic.replication.factor: 1
errors.deadletterqueue.context.headers.enable: true

但是我体验了2件事： 首先，我的DLQ总是空的。 其次，如果我在其中一个主题中有错误，我正在处理所有内容，一切都停止，并且连接器进入错误模式。

我首先想问我的假设： 我的假设是我可以发送到DLQ的消息是失败的消息，无论是ValueConverter还是KeyConverter处理。这将是我的期望。

其次，可以调用ValueConverter期间的错误，以便任务/连接器继续处理其他消息，但错误消息将进入DLQ。

目前，我在ValueConverter中发生故障的代码看起来像这样，这基本上使连接器爆炸而不是将消息发送到DLQ。

  public SchemaAndValue toConnectData(String topic, Headers headers, byte[] value) {
    try {
      Object deserialized = this.deserializer.deserialize(topic, value);

      if (deserialized == null) {
        return SchemaAndValue.NULL;
      } else if (deserialized instanceof Message) {
        Message message = (Message) deserialized;
        return this.protobufData.toConnectData(message.getDescriptorForType(), message, topic);
      }
      throw new DataException(String.format(
          "Unsupported type returned during deserialization of topic %s ",
          topic
      ));
    } catch (SerializationException e) {
      throw new DataException(String.format(
          "Failed to deserialize data for topic %s to Protobuf: ",
          topic
      ), e);
    }
  }

我知道这不是问题的一个小场景，但是老实说，我不知道如何轻松创建一个。我更想知道有使用Kafkaconnectors经验的人通常如何实施它，以便在DLQ中出现故障的消息最终会出现。

感谢您的任何反馈。