flink-cdc-pipeline-connector-starrocks-3.5.0

[liuyq2step]

StarRocks does not have timestamp-with-time-zone data types.

Testing Environment Specifications:

• Apache Flink: Version 1.20.2
• Flink CDC: Version 3.5
• PostgreSQL: Version 9.5
• StarRocks: Version 3.5.6

I tried to use flink-cdc pipeline yaml to sync table from postgresql to starrocks, if table have "timestamptz" column, the StarRocks sink connector throws error:
“
2025-10-27 15:45:37,447 INFO org.apache.flink.cdc.runtime.operators.schema.regular.SchemaCoordinator [] - All sink subtask have flushed for table public.t_flinktest. Start to apply schema change request:
SchemaChangeRequest{tableId=public.t_flinktest, schemaChangeEvent=CreateTableEvent{tableId=public.t_flinktest, schema=columns={id BIGINT NOT NULL,txt VARCHAR(255),dt TIMESTAMP(6) WITH TIME ZONE 'now()'}, primaryKeys=id, options=()}, subTaskId=0}
that extracts to:
CreateTableEvent{tableId=bgwltest.t_flinktest, schema=columns={id BIGINT NOT NULL,txt VARCHAR(255),dt TIMESTAMP(6) WITH TIME ZONE 'now()'}, primaryKeys=id, options=()}
2025-10-27 15:45:37,447 ERROR org.apache.flink.cdc.runtime.operators.schema.common.SchemaRegistry [] - An exception was triggered from Schema change applying task. Job will fail now.
org.apache.flink.util.FlinkRuntimeException: Failed to apply schema change event.
at org.apache.flink.cdc.runtime.operators.schema.regular.SchemaCoordinator.lambda$startSchemaChangesEvolve$1(SchemaCoordinator.java:249) ~[blob_p-8653b0f314a29add02bbe48f41a95c6f1bf2e1f7-d0bcfea2e9a125082a2675cdf4897b5d:3.5.0]
at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539) [?:?]
at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264) [?:?]
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) [?:?]
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) [?:?]
at java.base/java.lang.Thread.run(Thread.java:842) [?:?]
Caused by: java.lang.UnsupportedOperationException: Unsupported CDC data type TIMESTAMP(6) WITH TIME ZONE
at org.apache.flink.cdc.connectors.starrocks.sink.StarRocksUtils$CdcDataTypeTransformer.defaultMethod(StarRocksUtils.java:388) ~[blob_p-4b335e291125dd2ed4c2279fbcfaedf8cbc09e1d-5d045ecbf9ed8850946f607438ab0c30:3.5.0]
at org.apache.flink.cdc.connectors.starrocks.sink.StarRocksUtils$CdcDataTypeTransformer.defaultMethod(StarRocksUtils.java:241) ~[blob_p-4b335e291125dd2ed4c2279fbcfaedf8cbc09e1d-5d045ecbf9ed8850946f607438ab0c30:3.5.0]
at org.apache.flink.cdc.common.types.DataTypeDefaultVisitor.visit(DataTypeDefaultVisitor.java:105) ~[blob_p-8653b0f314a29add02bbe48f41a95c6f1bf2e1f7-d0bcfea2e9a125082a2675cdf4897b5d:3.5.0]
at org.apache.flink.cdc.common.types.ZonedTimestampType.accept(ZonedTimestampType.java:110) ~[blob_p-8653b0f314a29add02bbe48f41a95c6f1bf2e1f7-d0bcfea2e9a125082a2675cdf4897b5d:3.5.0]
at org.apache.flink.cdc.connectors.starrocks.sink.StarRocksUtils.toStarRocksDataType(StarRocksUtils.java:119) ~[blob_p-4b335e291125dd2ed4c2279fbcfaedf8cbc09e1d-5d045ecbf9ed8850946f607438ab0c30:3.5.0]
at org.apache.flink.cdc.connectors.starrocks.sink.StarRocksUtils.toStarRocksTable(StarRocksUtils.java:93) ~[blob_p-4b335e291125dd2ed4c2279fbcfaedf8cbc09e1d-5d045ecbf9ed8850946f607438ab0c30:3.5.0]
at org.apache.flink.cdc.connectors.starrocks.sink.StarRocksMetadataApplier.applyCreateTable(StarRocksMetadataApplier.java:138) ~[blob_p-4b335e291125dd2ed4c2279fbcfaedf8cbc09e1d-5d045ecbf9ed8850946f607438ab0c30:3.5.0]
at org.apache.flink.cdc.connectors.starrocks.sink.StarRocksMetadataApplier.lambda$applySchemaChange$2(StarRocksMetadataApplier.java:115) ~[blob_p-4b335e291125dd2ed4c2279fbcfaedf8cbc09e1d-5d045ecbf9ed8850946f607438ab0c30:3.5.0]
at org.apache.flink.cdc.common.event.visitor.SchemaChangeEventVisitor.visit(SchemaChangeEventVisitor.java:57) ~[blob_p-8653b0f314a29add02bbe48f41a95c6f1bf2e1f7-d0bcfea2e9a125082a2675cdf4897b5d:3.5.0]
at org.apache.flink.cdc.connectors.starrocks.sink.StarRocksMetadataApplier.applySchemaChange(StarRocksMetadataApplier.java:104) ~[blob_p-4b335e291125dd2ed4c2279fbcfaedf8cbc09e1d-5d045ecbf9ed8850946f607438ab0c30:3.5.0]
at org.apache.flink.cdc.runtime.operators.schema.regular.SchemaCoordinator.applyAndUpdateEvolvedSchemaChange(SchemaCoordinator.java:437) ~[blob_p-8653b0f314a29add02bbe48f41a95c6f1bf2e1f7-d0bcfea2e9a125082a2675cdf4897b5d:3.5.0]
at org.apache.flink.cdc.runtime.operators.schema.regular.SchemaCoordinator.applySchemaChange(SchemaCoordinator.java:406) ~[blob_p-8653b0f314a29add02bbe48f41a95c6f1bf2e1f7-d0bcfea2e9a125082a2675cdf4897b5d:3.5.0]
at org.apache.flink.cdc.runtime.operators.schema.regular.SchemaCoordinator.lambda$startSchemaChangesEvolve$1(SchemaCoordinator.java:247) ~[blob_p-8653b0f314a29add02bbe48f41a95c6f1bf2e1f7-d0bcfea2e9a125082a2675cdf4897b5d:3.5.0]
... 5 more
”

is this flink-cdc CreateTableEvent issue ? or starrocks sink connector issue?

PostgreSQL stores timestamptz (timestamp with time zone) values internally as UTC timestamps.
The original timezone information provided during insertion is converted to UTC and not preserved.
When queried, the values are converted to the client's timezone setting.