[
https://issues.apache.org/jira/browse/FLINK-40232?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18100134#comment-18100134
]
Iman Rastkhadiv commented on FLINK-40232:
-----------------------------------------
Hi, I'd like to pick this up.
Checked against master and the analysis holds: both branches dereference the
result of getTimestamp() without a null check, the reported line 74
(consumeRowtimeMetadata) and line 78 (rowtimeIndex != -1), so either path NPEs
when the rowtime attribute is null.
One detail that I think matters beyond the null checks in the suggested fix:
outRecord is a single StreamRecord allocated once in open() and reused for
every element, and the one-argument replace(X) does not touch the timestamp
flag. So skipping setTimestamp() when the rowtime is null leaves the previous
record's timestamp in place with hasTimestamp still true, which converts the
NPE into a silently wrong timestamp on the affected record rather than fixing
it. The null branch needs an explicit outRecord.eraseTimestamp().
Unless a null rowtime is meant to be an error here, my plan would be to
null-check both sites and erase the timestamp when the attribute is null, with
a test that sends a record carrying a rowtime followed by one with a null
rowtime, so the stale-timestamp case is covered and not just the NPE.
Could a committer assign this to me?
> NullPointerException in OutputConversionOperator.processElement when rowtime
> attribute is null
> ----------------------------------------------------------------------------------------------
>
> Key: FLINK-40232
> URL: https://issues.apache.org/jira/browse/FLINK-40232
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / API, Table SQL / Planner
> Affects Versions: 1.15.4, 1.16.3, 1.17.2, 1.18.1, 1.19.3, 2.0.2, 1.20.5,
> 2.1.3
> Environment: Flink 1.15+ (verified on release-1.15, release-2.1,
> master)
> Reporter: Fan Deng
> Priority: Major
>
> h2. Problem
> {{OutputConversionOperator.processElement}} (line 74) throws
> {{NullPointerException}} when a rowtime attribute is {{null}} and
> {{consumeRowtimeMetadata=true}}. The operator calls
> {{rowData.getTimestamp(...).getMillisecond()}} without null check.
> h2. Root Cause
> File:
> {{flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/sink/OutputConversionOperator.java}}
> {code:java}
> @Override
> public void processElement(StreamRecord<RowData> element) throws Exception {
> final RowData rowData = element.getValue();
> if (consumeRowtimeMetadata) {
> // timestamp is TIMESTAMP_LTZ
> final long rowtime = rowData.getTimestamp(rowData.getArity() - 1,
> 3).getMillisecond(); // ← line 74 NPE
> outRecord.setTimestamp(rowtime);
> } else if (rowtimeIndex != -1) {
> // timestamp might be TIMESTAMP or TIMESTAMP_LTZ
> final long rowtime = rowData.getTimestamp(rowtimeIndex,
> 3).getMillisecond(); // ← line 78 same issue
> outRecord.setTimestamp(rowtime);
> }
> ...
> }
> {code}
> The code calls {{rowData.getTimestamp(...)}} and immediately invokes
> {{.getMillisecond()}} on the result without null check. When the rowtime
> field is {{null}}, {{getTimestamp()}} returns Java {{null}} for certain
> {{RowData}} implementations, causing {{NullPointerException}}.
> h2. Why NPE Occurs with BoxedWrapperRowData
> Calc codegen produces {{BoxedWrapperRowData}} as output (see
> {{CalcCodeGenerator.scala:61}}):
> {code:scala}
> // CalcCodeGenerator.scala:61
> classOf[BoxedWrapperRowData],
> {code}
> {{BoxedWrapperRowData.getTimestamp}} returns Java {{null}} for null fields
> (unlike {{BinaryRowData}} which returns {{TimestampData.fromEpochMillis(0)}}):
> {code:java}
> // BoxedWrapperRowData.java:115-116
> public TimestampData getTimestamp(int pos, int precision) {
> return (TimestampData) this.fields[pos]; // fields[pos]==null →
> returns Java null
> }
> {code}
> ||RowData implementation||{{getTimestamp(pos, precision)}} for null
> field||{{.getMillisecond()}}||
> |{{BinaryRowData}}|{{TimestampData.fromEpochMillis(0)}} (non-null)|Returns
> 0, no NPE|
> |{{BoxedWrapperRowData}}|Java {{null}}|NPE|
> |{{GenericRowData}}|Java {{null}}|NPE|
> h2. Trigger Conditions
> The NPE requires all of the following:
> {{consumeRowtimeMetadata=true}}: The sink schema declares a rowtime
> metadata column (e.g., via {{Schema.columnByMetadata("rowtime", ...)}})
> Null rowtime field: A record with null rowtime reaches
> {{OutputConversionOperator}}
> {{BoxedWrapperRowData}} passed directly: Calc codegen output
> ({{BoxedWrapperRowData}}) is passed directly to {{OutputConversionOperator}}
> without serialization (operator chaining enabled, object reuse enabled)
> No upstream {{WatermarkAssignerOperator}}: Source performs watermark
> pushdown (e.g., Kafka source implements {{SupportsWatermarkPushDown}}), so
> the planner does not insert an independent
> {{WatermarkAssignerOperator}} that would otherwise reject null rowtime
> earlier
> h2. Reproduction
> h3. End-to-end MiniCluster test
> Verified on release-1.15:
> {code:java}
> // Source with watermark pushdown (no independent WatermarkAssignerOperator)
> tenv.executeSql(
> "CREATE TABLE src (\n"
> + " f0 STRING,\n"
> + " rtime TIMESTAMP(3),\n"
> + " WATERMARK FOR rtime AS rtime - INTERVAL '10' SECOND\n"
> + ") WITH (\n"
> + " 'connector' = 'pushdown-watermark-values',\n"
> + " 'data-id' = '...' // emits Row with null rtime\n"
> + ")");
> Table table = tenv.from("src");
> // Sink schema with rowtime metadata column → consumeRowtimeMetadata=true
> Schema sinkSchema = Schema.newBuilder()
> .column("f0", DataTypes.STRING())
> .columnByMetadata("rtime", DataTypes.TIMESTAMP_LTZ(3).notNull(),
> "rowtime")
> .build();
> // toChangelogStream triggers
> OutputConversionOperator(consumeRowtimeMetadata=true)
> // Calc codegen produces BoxedWrapperRowData → getTimestamp returns null →
> NPE
> tenv.toChangelogStream(table, sinkSchema,
> ChangelogMode.insertOnly()).print();
> env.execute("repro"); // → NPE at
> OutputConversionOperator.processElement:74
> {code}
> h3. Stack trace
> {code}
> java.lang.NullPointerException
> at
> org.apache.flink.table.runtime.operators.sink.OutputConversionOperator.processElement(OutputConversionOperator.java:74)
> at
> org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:82)
> ...
> Caused by: java.lang.NullPointerException
> at
> org.apache.flink.table.runtime.operators.sink.OutputConversionOperator.processElement(OutputConversionOperator.java:74)
> {code}
> h2. Why disable-operator-chaining works as workaround
> When operator chaining is disabled, the chain boundary forces
> {{RowDataSerializer}} serialization/deserialization. The deserialized row
> becomes {{BinaryRowData}}, whose {{getTimestamp}} is null-tolerant (returns
> {{TimestampData.fromEpochMillis(0)}} for null fields). This is not a fix —
> it's a workaround that masks the underlying bug.
> h2. Impact
> - Affects: release-1.15, release-2.1, master (verified byte-identical via
> {{git diff}})
> - Introduced in: FLINK-19980 (Timo Walther, 2021-04-13)
> - Triggered in production: Oceanus (Tencent's Flink platform) — Kafka
> source with source-watermark pushdown + {{toChangelogStream}} with rowtime
> metadata column
> - No existing JIRA: This bug has not been reported to the community for 5+
> years
> h2. Suggested Fix
> Add null check in {{OutputConversionOperator.processElement}}:
> {code:java}
> @Override
> public void processElement(StreamRecord<RowData> element) throws Exception {
> final RowData rowData = element.getValue();
> if (consumeRowtimeMetadata) {
> // timestamp is TIMESTAMP_LTZ
> TimestampData ts = rowData.getTimestamp(rowData.getArity() - 1, 3);
> if (ts != null) {
> final long rowtime = ts.getMillisecond();
> outRecord.setTimestamp(rowtime);
> }
> } else if (rowtimeIndex != -1) {
> // timestamp might be TIMESTAMP or TIMESTAMP_LTZ
> TimestampData ts = rowData.getTimestamp(rowtimeIndex, 3);
> if (ts != null) {
> final long rowtime = ts.getMillisecond();
> outRecord.setTimestamp(rowtime);
> }
> }
> ...
> }
> {code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
