Fan Deng created FLINK-40232:
--------------------------------
Summary: 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: 2.1.3, 1.20.5, 2.0.2, 1.19.3, 1.18.1, 1.17.2, 1.16.3,
1.15.4
Environment: Flink 1.15+ (verified on release-1.15, release-2.1,
master)
Reporter: Fan Deng
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)
