apache-kafka - ElasticsearchSinkConnector 无法将数据反序列化到 Avro
问题描述
我创建了最简单的 kafka sink 连接器配置,我使用的是 confluent 4.1.0:
{
"connector.class":
"io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
"type.name": "test-type",
"tasks.max": "1",
"topics": "dialogs",
"name": "elasticsearch-sink",
"key.ignore": "true",
"connection.url": "http://localhost:9200",
"schema.ignore": "true"
}
在主题中,我将消息保存为JSON
{ "topics": "resd"}
但结果我得到一个错误:
起因:org.apache.kafka.common.errors.SerializationException:反序列化 id -1 的 Avro 消息时出错 起因:org.apache.kafka.common.errors.SerializationException:未知魔术字节!
解决方案
正如 cricket_007 所说,您需要告诉 Connect 使用 Json 反序列化器,如果这是您的数据格式。将其添加到您的连接器配置中:
"value.converter": "org.apache.kafka.connect.json.JsonConverter",
"value.converter.schemas.enable": "false",
"key.converter": "org.apache.kafka.connect.json.JsonConverter",
"key.converter.schemas.enable": "false"
推荐阅读
- html - 正面大小的最大值不影响浏览器字体大小
- java - SOAP XSD 根元素问题
- expression - 如何从自定义 geojson 道具中获取价值,而没有 Mapbox 中的绘画表达式属性
- azure-devops - 是否可以使用 $(RELEASE.TRIGGERINGARTIFACT.ALIAS) 作为触发不同发布阶段的条件?
- python - 在 jenkins 中执行 pytest TestCases 给出错误,因为找不到 .py 模块
- orm - 使用 NOT IN 子句和嵌套 SELECT 对查询进行续集
- c++11 - C++:不是从 obj1 转换为 obj2,而是从 obj1* 转换为 obj2?
- reactjs - 如何在反例中使用上下文 API 做出反应?
- expo - 安装新版 expo-cli@3.26.2 出现大量警告
- javascript - 没有wsdl的nodejs中的SOAP调用