StarRocks / starrocks-connector-for-kafka

Apache License 2.0
6 stars 10 forks source link

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

Closed armamut closed 8 months ago

armamut commented 8 months ago

Steps to reproduce the behavior (Required)

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)

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

    key.converter.decimal.format: NUMERIC

I have also tried:

    key.converter: org.apache.kafka.connect.json.JsonConverter
    key.converter.schemas.enable: true
    value.converter: org.apache.kafka.connect.json.JsonConverter
    value.converter.schemas.enable: true

But the same result.

StarRocks version (Required)

StarRocks version 3.1.7-7e897e2

armamut commented 8 months ago

Hello, I've upgraded to latest connector, and my problem has solved.