daguimu commented on code in PR #28071:
URL: https://github.com/apache/flink/pull/28071#discussion_r3192675493
##########
flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/RowDataToAvroConverters.java:
##########
@@ -166,7 +168,11 @@ public Object convert(Schema schema, Object object) {
@Override
public Object convert(Schema schema, Object
object) {
- return ((TimestampData)
object).toInstant().toEpochMilli();
+ final TimestampData timestampData =
(TimestampData) object;
+ if (isMicrosLogicalType(schema)) {
Review Comment:
Thanks for the careful read.
My read is that this is a correctness fix rather than a behavior change
worth preserving. The read path in `AvroToRowDataConverters` already honors
`timestamp-micros` / `local-timestamp-micros` and returns microseconds, while
the write path here ignores the same logical type and always emits
milliseconds. A pipeline that round-trips through Flink with a
`timestamp-micros` schema silently scales timestamps by 1000× — that's a
data-corruption bug, not a stable contract.
The fix is also gated by what the schema declares: users whose schemas are
`timestamp-millis` (the previous behavior for both branches) see no change at
all. Only users who already declared `timestamp-micros` and were silently
getting wrong values are affected, and for them the new output is what their
schema asked for.
That said, the scope question deserves a committer's call. Could one of the
flink-avro maintainers weigh in on whether a config flag is warranted on
master? If so, happy to add one as a follow-up.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
