flink-doris-connector-1.17-1.4.0日期处理问题

👨🏻‍💻 技术问答
1

我在使用flink-doris-connector时，关于日期类型数据发生错误，请专家给与解答一下，多谢！信息如下：

1: DorisRowConverter 源码中 L262 代码如下：
return (index, val) → Date.valueOf(LocalDate.ofEpochDay(val.getInt(index)));

val.getInt(index) 是返回 int 类型，而LocalDate.ofEpochDay参数是long类型。
代码类型不匹配。

是不是应该：val.getLong(index) ？

2：对象日期都是long类型数据，这里面是val.getInt(index)，造成错误
java.lang.Long cannot be cast to class java.lang.Integer

2

完整的错误和FlinkSQL贴一下呢。
Date类型，在Flink内部的类型传递应该都是 int，可以参考
https://github.com/apache/flink/blob/master/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/connector/datagen/table/DataGenVisitorBase.java#L50C5-L50C5

3

详细信息如下：
Caused by: java.lang.ClassCastException: class java.lang.Long cannot be cast to class java.lang.Integer (java.lang.Long and java.lang.Integer are in module java.base of loader ‘bootstrap’)
at org.apache.flink.table.data.GenericRowData.getInt(GenericRowData.java:149)
at org.apache.doris.flink.deserialization.converter.DorisRowConverter.lambda$createExternalConverter$81fa9aea$11(DorisRowConverter.java:262)
at org.apache.doris.flink.deserialization.converter.DorisRowConverter.lambda$wrapIntoNullableExternalConverter$97af1164$1(DorisRowConverter.java:150)
at org.apache.doris.flink.deserialization.converter.DorisRowConverter.convertExternal(DorisRowConverter.java:118)
at org.apache.doris.flink.sink.writer.RowDataSerializer.buildJsonString(RowDataSerializer.java:79)
at org.apache.doris.flink.sink.writer.RowDataSerializer.serialize(RowDataSerializer.java:66)
at org.apache.doris.flink.sink.writer.RowDataSerializer.serialize(RowDataSerializer.java:42)
at org.apache.doris.flink.sink.writer.DorisWriter.write(DorisWriter.java:133)
at org.apache.flink.streaming.api.transformations.SinkV1Adapter$SinkWriterV1Adapter.write(SinkV1Adapter.java:136)
at org.apache.flink.streaming.runtime.operators.sink.SinkWriterOperator.processElement(SinkWriterOperator.java:161)
at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:75)
at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:50)
at org.apache.flink.streaming.runtime.tasks.ChainingOutput.collectAndCheckIfChained(ChainingOutput.java:89)
at org.apache.flink.streaming.runtime.tasks.ChainingOutput.collectAndCheckIfChained(ChainingOutput.java:39)
at org.apache.flink.streaming.runtime.tasks.BroadcastingOutputCollector.collect(BroadcastingOutputCollector.java:82)
at org.apache.flink.streaming.runtime.tasks.BroadcastingOutputCollector.collect(BroadcastingOutputCollector.java:33)
at org.apache.flink.streaming.api.operators.StreamMap.processElement(StreamMap.java:38)
at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:75)
at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:50)
at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:29)
at org.apache.flink.streaming.api.operators.TimestampedCollector.collect(TimestampedCollector.java:51)e[33m2023-11-27 17:44:26.436e[0;39m [e[35mForkJoinPool.commonPool-worker-15e[0;39m] e[34mINFO e[0;39m e[36morg.apache.flink.runtime.dispatcher.DispatcherRestEndpointe[0;39m - [revokeLeadership,1126] - http://localhost:60268 lost leadership

at com.spiov.flink.functions.CementProcessFunction.processElement(CementProcessFunction.java:36)
at com.spiov.flink.functions.CementProcessFunction.processElement(CementProcessFunction.java:1)
at org.apache.flink.streaming.api.operators.ProcessOperator.processElement(ProcessOperator.java:66)
at org.apache.flink.streaming.runtime.tasks.OneInputStreamTask$StreamTaskNetworkOutput.emitRecord(OneInputStreamTask.java:237)
at org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.processElement(AbstractStreamTaskNetworkInput.java:146)
at org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.emitNext(AbstractStreamTaskNetworkInput.java:110)
at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65)
at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:562)
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:231)
at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:858)
at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:807)
at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:953)
at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:932)
at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:746)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:562)
at java.base/java.lang.Thread.run(Thread.java:834)