java - Datastax Kafka 连接器无法解析 Json 主题

Question

我有一个生产者类使用 Github 中的自定义 JsonSerializer 发送到主题

public class JsonSerializer<T> implements Serializer<T> {
    ...
    @Override
    public byte[] serialize(String topic, T data) {
        try {
            return this.objectMapper.writeValueAsBytes(data);
        } catch (JsonProcessingException e) {
            throw new SerializationException(e);
        }
    }
    ...
}

我正在使用这些配置运行 Datastax Kafka Connector:

value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false

当连接器尝试使用主题时，我收到这些错误:

[2020-01-12 13:57:53,324] WARN Error inserting/updating row for Kafka record SinkRecord{kafkaOffset=416, timestampType=CreateTime} ConnectRecord{topic='test-3', kafkaPartition=17, key=null, keySchema=Schema{STRING}, value={}, valueSchema=null, timestamp=1578811437723, headers=ConnectHeaders(headers=)}: Primary key column(s) mmsi, ts cannot be left unmapped. Check that your mapping setting matches your dataset contents. (com.datastax.kafkaconnector.DseSinkTask:286)

从该错误中，我认为连接器无法检索 Json 数据。我做错了什么？

更新

我尝试了 Kafka JsonSerializer。

我尝试了 StringSerializer，因为连接器说它也受支持。

我发现一些数据实际上写入了数据库，但与kafka主题发送的总数据相比，它总是相对较小的数量。大约5到10个数据。

我尝试让连接器保持运行，发现写入失败后，就不会再写入了。

Answer 1

实际上是配置相关的问题。正如我在更新中提到的，如果出现错误，它不再写入数据。

这是因为 Datastax 有配置 ignoreErrors，这些配置具有默认值 false。这意味着如果连接器在消息中发现错误，它将无限期地重试。我将其设置为 true，问题就解决了。