[Bug] [KafkaTableStream] Kafka Source

[immustard]

Search before asking

• I had searched in the issues and found no similar issues.

What happened

source is KafkaTableStream and the schema uses the example. but the flink job error : com.fasterxml.jackson.databind.node.ArrayNode cannot be cast to com.fasterxml.jackson.databind.node.ObjectNode

SeaTunnel Version

v2.1.2

SeaTunnel Config

env {
  execution.parallelism=1
}

source {
  KafkaTableStream {
      consumer.bootstrap.servers = "192.168.10.102:9092"
      consumer.group.id = "seatunnel-learn"
      topics = test_csv
      result_table_name = test
      format.type = csv
      schema = "[{\"field\":\"name\",\"type\":\"string\"},{\"field\":\"age\",\"type\":\"int\"}]"
      format.field-delimiter = ";"
      format.allow-comments = "true"
      format.ignore-parse-errors = "true"
  }
}

transform {
  sql {
    sql = "select name, age from test where age > 18"
  }
}

sink {
  Kafka {
    topics = "test_sink"
    producer.bootstrap.server = "192.168.10.102:9092"
  }
}

Running Command

./bin/start-seatunnel-flink.sh --config ./config/kafkaETL.conf

Error Exception

Exception in thread "main" java.lang.RuntimeException: Execute Flink task error
	at org.apache.seatunnel.core.flink.command.FlinkTaskExecuteCommand.execute(FlinkTaskExecuteCommand.java:84)
	at org.apache.seatunnel.core.base.Seatunnel.run(Seatunnel.java:39)
	at org.apache.seatunnel.example.flink.LocalFlinkExample.main(LocalFlinkExample.java:40)
Caused by: java.lang.RuntimeException: String json deserialization exception.
	at org.apache.seatunnel.common.utils.JsonUtils.parseObject(JsonUtils.java:242)
	at org.apache.seatunnel.flink.kafka.source.KafkaTableStream.prepare(KafkaTableStream.java:114)
	at org.apache.seatunnel.flink.kafka.source.KafkaTableStream.prepare(KafkaTableStream.java:51)
	at org.apache.seatunnel.core.base.command.BaseTaskExecuteCommand.lambda$prepare$0(BaseTaskExecuteCommand.java:67)
	at java.util.ArrayList.forEach(ArrayList.java:1259)
	at org.apache.seatunnel.core.base.command.BaseTaskExecuteCommand.prepare(BaseTaskExecuteCommand.java:67)
	at org.apache.seatunnel.core.flink.command.FlinkTaskExecuteCommand.execute(FlinkTaskExecuteCommand.java:80)
	... 2 more
Caused by: java.lang.ClassCastException: com.fasterxml.jackson.databind.node.ArrayNode cannot be cast to com.fasterxml.jackson.databind.node.ObjectNode
	at org.apache.seatunnel.common.utils.JsonUtils.parseObject(JsonUtils.java:239)
	... 8 more

Flink or Spark Version

Flink 1.13.6

Java or Scala Version

Java 1.8
scala 2.12

Screenshots

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[san346596324]

+1,2.11 no problem

[immustard]

+1,2.11 no problem

yep, it's no problem in v2.1.1

[Hisoka-X]

I got, the reason is we replace fastjson by jackson, so if your schema is list not object, the convert will report exception, I will fix it. Thanks for report.

[Hisoka-X]

Did you try use list schema in 2.1.1, is it work fine?

[immustard]

Did you try use list schema in 2.1.1, is it work fine?

yes, it's work fine in 2.1.1

[VictorZeng]

I got, the reason is we replace fastjson by jackson, so if your schema is list not object, the convert will report exception, I will fix it. Thanks for report.

Hi, Why use Jackson to replace fastjson?

[Hisoka-X]

I got, the reason is we replace fastjson by jackson, so if your schema is list not object, the convert will report exception, I will fix it. Thanks for report.

Hi, Why use Jackson to replace fastjson?

Different json frameworks have been used in the project before. For unification, we chose jackson.

[Hisoka-X]

Close by #2168

[immustard]

您好，您的邮件已经收到。