Starrocks Kafka Connector gives java.lang.NullPointerException: Cannot invoke JsonConverterConfig.decimalFormat()

68 Views Asked by At

I'm trying to run Kafka Connector on strimzi kafka connect cluster with these values:

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnector
metadata:
  name: starrocks-kafka-connector
  labels:
    strimzi.io/cluster: my-debezium-connect-cluster
spec:
  class: com.starrocks.connector.kafka.StarRocksSinkConnector
  tasksMax: 1
  config:
    name: starrocks-kafka-connector
    tasks.max: 1
    topics: ...
    starrocks.http.url: 10....:8030
    starrocks.username: root
    starrocks.password: ....
    starrocks.database.name: kafkatest

    key.converter: io.confluent.connect.json.JsonSchemaConverter
    key.converter.schema.registry.url: http://debezium-schema-registry:8081
    key.converter.decimal.format: NUMERIC
    value.converter: io.confluent.connect.json.JsonSchemaConverter
    value.converter.schema.registry.url: http://debezium-schema-registry:8081    
    key.converter.decimal.format: NUMERIC

    transforms: unwrap
    transforms.unwrap.type: io.debezium.transforms.ExtractNewRecordState
    transforms.unwrap.add.fields: op,source.ts_ms
    transforms.unwrap.delete.handling.mode: rewrite
    transforms.unwrap.drop.tombstones: false    

But I get this error message:

2024-01-14 22:11:46,129 ERROR [starrocks-kafka-connector|task-0] WorkerSinkTask{id=starrocks-kafka-connector-0} Task threw an uncaught and unrecoverable exception. Task is being killed and will not recover until manually restarted (org.apache.kafka.connect.runtime.WorkerTask) [task-thread-starrocks-kafka-connector-0]
Caused by: java.lang.NullPointerException: Cannot invoke "com.starrocks.connector.kafka.json.JsonConverterConfig.decimalFormat()" because "config" is null
        at com.starrocks.connector.kafka.json.JsonConverter$1.toJson(JsonConverter.java:141)
        at com.starrocks.connector.kafka.json.JsonConverter.convertToJson(JsonConverter.java:569)
        at com.starrocks.connector.kafka.json.JsonConverter.convertToJson(JsonConverter.java:656)
        at com.starrocks.connector.kafka.StarRocksSinkTask.getRecordFromSinkRecord(StarRocksSinkTask.java:220)
        at com.starrocks.connector.kafka.StarRocksSinkTask.put(StarRocksSinkTask.java:258)

(StarRocks version 3.1.7-7e897e2)

I tried to add or remove this, but nothing changed:

    key.converter.decimal.format: NUMERIC
0

There are 0 best solutions below