[Feature] Kafka CDC type conversion

[kwontaeheon]

Search before asking

• [X] I searched in the issues and found nothing similar.

Motivation

Using Maxwell CDC action, complex types can not changed to string. Can I define the type of kafka source from commandline or, not to cast to string? the samples are below.

• paimon table (created from spark)

  create table if not exists `local`.`default`.`id_mapping`
  ( id STRING, val_list ARRAY<STRING>, primary key (id) NOT ENFORCED));

• kafka message json

  {"type": "insert", "primary_key": ["1111"], "primary_key_columns": ["id"], "data": {"id": "1111", "val_list": ["2", "3"]}}

• error messages

  java.lang.UnsupportedOperationException: Cannot convert field val_list from type ARRAY<STRING> to STRING of Paimon table default.id_mapping.
  at org.apache.paimon.flink.sink.cdc.UpdatedDataFieldsProcessFunctionBase.applySchemaChange(UpdatedDataFieldsProcessFunctionBase.java:124)
  at org.apache.paimon.flink.sink.cdc.UpdatedDataFieldsProcessFunction.processElement(UpdatedDataFieldsProcessFunction.java:58)
  at org.apache.paimon.flink.sink.cdc.UpdatedDataFieldsProcessFunction.processElement(UpdatedDataFieldsProcessFunction.java:39)
  at org.apache.flink.streaming.api.operators.ProcessOperator.processElement(ProcessOperator.java:66)
  at org.apache.flink.streaming.runtime.tasks.OneInputStreamTask$StreamTaskNetworkOutput.emitRecord(OneInputStreamTask.java:238)
  at org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.processElement(AbstractStreamTaskNetworkInput.java:157)
  at org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.emitNext(AbstractStreamTaskNetworkInput.java:114)
  at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65)
  at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:579)
  at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:231)
  at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:909)
  at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:858)
  at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:958)
  at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:937)
  at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:751)
  at org.apache.flink.runtime.taskmanager.Task.run(Task.java:566)
  at java.base/java.lang.Thread.run(Thread.java:829)

Solution

No response

Anything else?

No response

Are you willing to submit a PR?

• [ ] I'm willing to submit a PR!

[kwontaeheon]

Without paimon flink action, I found I can easily ingest cdc by flink table api. Thanks.