Ошибка сериализации Kafka Connect в Elastic Sink - PullRequest
Новая
       6

Ошибка сериализации Kafka Connect в Elastic Sink

0 голосов
/ 05 октября 2018

Я использую соединитель приемника kafkaasticsearch для передачи входящего сообщения в ES, но сталкиваюсь со следующей проблемой 

[2018-10-05 13:01:21,388] ERROR WorkerSinkTask{id=elasticsearch.sink.direct-
10} Task threw an uncaught and unrecoverable exception (org.apache.kafka.connect.runtime.WorkerTask:172)
org.apache.kafka.connect.errors.DataException: Converting byte[] to Kafka Connect data failed due to serialization error:

Caused by: org.apache.kafka.common.errors.SerializationException: com.fasterxml.jackson.core.JsonParseException: Illegal character ((CTRL-CHAR, code 0)): only regular white space (\r, \n, \t) is allowed between tokens
 at [Source: (byte[])"  "; line: 1, column: 2]
Caused by: com.fasterxml.jackson.core.JsonParseException: Illegal character ((CTRL-CHAR, code 0)): only regular white space (\r, \n, \t) is allowed between tokens
 at [Source: (byte[])"  "; line: 1, column: 2]

Входящее сообщение о значении ключа выглядит следующим образом при запуске консоли пользователя сДля свойства print.key установлено значение true 

{
"schema": {
    "type": "struct",
    "fields": [{
            "type": "int32",
            "optional": false,
            "field": "MY_SETTING_ID"
        }
    ],
    "optional": false
},
"payload": {
    "MY_SETTING_ID": 9
}
}


{
    "schema": {
        "type": "struct",
        "fields": [{
                "type": "int32",
                "optional": false,
                "field": "MY_SETTING_ID"
            }, {
                "type": "string",
                "optional": true,
                "field": "MY_SETTING_NAME"
            }
        ],
        "optional": false
    },
    "payload": {
        "MY_SETTING_ID": 9,
        "MY_SETTING_NAME": "setting_name"
    }
}

Здесь ключ MY_SETTING_ID действует как ключ.

У меня есть следующий автономный файл свойств 

bootstrap.servers=dev-insight-kafka01:9092,dev-insight-kafka02:9092,dev-

insight-kafka03:9092
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter

key.converter.schemas.enable=false
value.converter.schemas.enable=false

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

offset.storage.file.filename=/apps/dev/logs/offsets/elasticsearch-direct.offsets
offset.flush.interval.ms=120000
rest.port=8099

plugin.path=/usr/share/java
producer.max.request.size = 10485760
consumer.auto.offset.reset=latest
consumer.session.timeout.ms=300000
consumer.request.timeout.ms=310000
flush.timeout.ms=160000
heartbeat.interval.ms= 60000
session.timeout.ms= 200000

и файл поддержки приемника: 

name=elasticsearch.sink.direct
connector.class=io.confluent.connect.elasticsearch.ElasticsearchSinkConnector
tasks.max=16
topics=stream.app_setting

connection.url=http://dev-elastic-search01:9200
type.name=logs
topic.index.map=stream.app_setting:direct_app_setting_index
batch.size=2048
max.buffered.records=32768
flush.timeout.ms=60000
max.retries=10
retry.backoff.ms=1000
schema.ignore=true

Буду признателен, если кто-то может просмотреть мой файл свойстви скажи мне, где я могу пойти не так

1 Ответ

0 голосов
/ 06 октября 2018

Поскольку у вас есть схема как часть вашего JSON, вы должны установить 

key.converter.schemas.enable=true
value.converter.schemas.enable=true
...