[Bug] set local-merge-buffer-size cause NullPointerException
- Dominant language
- Java
- Stars
- 3.4k
- Forks
- 1.4k
- Avg merge
- 1d 11h
- Merged PRs (30d)
- 396
Description
### Search before asking
- [X] I searched in the [issues](https://github.com/apache/incubator-paimon/issues) and found nothing similar.
### Paimon version
0.5.0-incubating
### Compute Engine
flink 1.17.0
### Minimal reproduce step
create a paimon table with _id as primary key, set "local-merge-buffer-size":"256m" for this table, start mongodb cdc as source and write to this paimon table, a few minutes later, you will got exception:
```
[Writer : mongocdc_t_lint_defect_v2_prod_100205 (20/32)#0] WARN org.apache.flink.runtime.taskmanager.Task - Writer : mongocdc_t_lint_defect_v2_prod_100205 (20/32)#0 (f55b061c8b8bbfbcd0e97c0e4e6a1ed0_306d8342cb5b2ad8b53f1be57f65bee8_19_0) switched from RUNNING to FAILED with failure cause:
java.io.IOException: java.lang.NullPointerException
at org.apache.paimon.flink.sink.RowDataStoreWriteOperator.processElement(RowDataStoreWriteOperator.java:125)
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:550)
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:231)
at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:839)
at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:788)
at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:952)
at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:931)
at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:745)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:562)
at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.NullPointerException: null
at org.apache.paimon.data.BinaryRow.isNullAt(BinaryRow.java:159)
at org.apache.paimon.utils.OffsetRow.isNullAt(OffsetRow.java:65)
at org.apache.paimon.flink.FlinkRowData.isNullAt(FlinkRowData.java:70)
at org.apache.paimon.flink.FlinkRowWrapper.isNullAt(FlinkRowWrapper.java:63)
at Projection$229.apply(Unknown Source)
at org.apache.paimon.table.sink.RowPartitionKeyExtractor.trimmedPrimaryKey(RowPartitionKeyExtractor.java:49)
at org.apache.paimon.table.sink.RowKeyExtractor.trimmedPrimaryKey(RowKeyExtractor.java:65)
at org.apache.paimon.table.sink.FixedBucketRowKeyExtractor.bucketKey(FixedBucketRowKeyExtractor.java:59)
at org.apache.paimon.table.sink.FixedBucketRowKeyExtractor.bucket(FixedBucketRowKeyExtractor.java:70)
at org.apache.paimon.table.sink.TableWriteImpl.toSinkRecord(TableWriteImpl.java:130)
at org.apache.paimon.table.sink.TableWriteImpl.writeAndReturn(TableWriteImpl.java:113)
at org.apache.paimon.flink.sink.StoreSinkWriteImpl.write(StoreSinkWriteImpl.java:159)
at org.apache.paimon.flink.sink.RowDataStoreWriteOperator.processElement(RowDataStoreWriteOperator.java:123)
... 13 common frames omitted
```
if you remove "local-merge-buffer-size" property from this paimon table's properties, then start the application again, it works fine.
table schema:
```
{
"id" : 1,
"fields" : [ {
"id" : 0,
"name" : "_id",
"type" : "STRING NOT NULL"
}, {
"id" : 1,
"name" : "_class",
"type" : "STRING"
}, {
"id" : 2,
"name" : "id",
"type" : "STRING"
}, {
"id" : 3,
"name" : "task_id",
"type" : "BIGINT"
}, {
"id" : 4,
"name" : "tool_name",
"type" : "STRING"
}, {
"id" : 5,
"name" : "file_name",
"type" : "STRING"
}, {
"id" : 6,
"name" : "line_num",
"type" : "BIGINT"
}, {
"id" : 7,
"name" : "author",
"type" : "STRING"
}, {
"id" : 8,
"name" : "checker",
"type" : "STRING"
}, {
"id" : 9,
"name" : "severity",
"type" : "BIGINT"
}, {
"id" : 10,
"name" : "message",
"type" : "STRING"
}, {
"id" : 11,
"name" : "defect_type",
"type" : "BIGINT"
}, {
"id" : 12,
"name" : "status",
"type" : "BIGINT"
}, {
"id" : 13,
"name" : "line_update_time",
"type" : "BIGINT"
}, {
"id" : 14,
"name" : "rel_path",
"type" : "STRING"
}, {
"id" : 15,
"name" : "file_path",
"type" : "STRING"
}, {
"id" : 16,
"name" : "pinpoint_hash",
"type" : "STRING"
}, {
"id" : 17,
"name" : "pinpoint_hash_group",
"type" : "STRING"
}, {
"id" : 18,
"name" : "create_time",
"type" : "BIGINT"
}, {
"id" : 19,
"name" : "fixed_time",
"type" : "BIGINT"
}, {
"id" : 20,
"name" : "exclude_time",
"type" : "BIGINT"
}, {
"id" : 21,
"name" : "ignore_time",
"type" : "BIGINT"
}, {
"id" : 22,
"name" : "mark_time",
"type" : "BIGINT"
}, {
"id" : 23,
"name" : "lang_value",
"type" : "BIGINT"
}, {
"id" : 24,
"name" : "create_date",
"type" : "BIGINT"
}, {
"id" : 25,
"name" : "updated_date",
"type" : "BIGINT"
}, {
"id" : 26,
"name" : "ignore_reason_type",
"type" : "BIGINT"
}, {
"id" : 27,
"name" : "ignore_reason",
"type" : "STRING"
}, {
"id" : 28,
"name" : "ignore_author",
"type" : "STRING"
}, {
"id" : 29,
"name" : "mark_but_no_fixed",
"type" : "BIGINT"
}, {
"id" : 30,
"name" : "ignore_comment_defect",
"type" : "BIGINT"
}, {
"id" : 31,
"name" : "mark",
"type" : "STRING"
}, {
"id" : 32,
"name" : "create_build_number",
"type" : "STRING"
}, {
"id" : 33,
"name" : "fixed_build_number",
"type" : "STRING"
}, {
"id" : 34,
"name" : "url",
"type" : "STRING"
}, {
"id" : 35,
"name" : "repo_id",
"type" : "STRING"
}, {
"id" : 36,
"name" : "revision",
"type" : "STRING"
}, {
"id" : 37,
"name" : "branch",
"type" : "STRING"
}, {
"id" : 38,
"name" : "sub_module",
"type" : "STRING"
}, {
"id" : 39,
"name" : "file_update_time",
"type" : "BIGINT"
}, {
"id" : 40,
"name" : "file_md5",
"type" : "STRING"
}, {
"id" : 41,
"name" : "language",
"type" : "STRING"
}, {
"id" : 42,
"name" : "mask_path",
"type" : "STRING"
}, {
"id" : 43,
"name" : "ignore_build_id",
"type" : "STRING"
}, {
"id" : 44,
"name" : "display_type",
"type" : "STRING"
}, {
"id" : 45,
"name" : "display_category",
"type" : "STRING"
}, {
"id" : 46,
"name" : "ext_bug_id",
"type" : "STRING"
}, {
"id" : 47,
"name" : "file_version",
"type" : "STRING"
}, {
"id" : 48,
"name" : "platform_build_id",
"type" : "STRING"
}, {
"id" : 49,
"name" : "platform_project_id",
"type" : "STRING"
}, {
"id" : 50,
"name" : "stream_name",
"type" : "STRING"
}, {
"id" : 51,
"name" : "created_by",
"type" : "STRING"
}, {
"id" : 52,
"name" : "updated_by",
"type" : "STRING"
}, {
"id" : 53,
"name" : "defect_instances",
"type" : "STRING"
}, {
"id" : 54,
"name" : "____et",
"type" : "TIMESTAMP(3) WITH LOCAL TIME ZONE",
"description" : ""
}, {
"id" : 55,
"name" : "dteventtime",
"type" : "STRING",
"description" : ""
}, {
"id" : 56,
"name" : "dteventtimestamp",
"type" : "BIGINT",
"description" : ""
}, {
"id" : 57,
"name" : "localtime",
"type" : "STRING",
"description" : ""
}, {
"id" : 58,
"name" : "thedate",
"type" : "INT",
"description" : ""
} ],
"highestFieldId" : 58,
"partitionKeys" : [ ],
"primaryKeys" : [ "_id" ],
"options" : {
"bucket" : "64",
"num-sorted-run.stop-trigger" : "2147483647",
"target-file-size" : "64m",
"write-only" : "false",
"num-sorted-run.compaction-trigger" : "10",
"file.format" : "parquet",
"write-buffer-size" : "128m",
"sort-spill-threshold" : "8",
"write-buffer-spillable" : "true"
},
"timeMillis" : 1694693389123
}
```
```
https://github.com/apache/incubator-paimon/blob/release-0.5/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSinkBuilder.java#L73
DataStream input = this.input;
if (table.coreOptions().localMergeEnabled() && table.schema().primaryKeys().size() > 0) {
input =
input.forward()
.transform(
"local merge",
input.getType(),
new LocalMergeOperator(table.schema()))
.setParallelism(input.getParallelism());
}
```
flink topology:
### What doesn't meet your expectations?
task manager works fine without throw exception.
### Anything else?
_No response_
### Are you willing to submit a PR?
- [ ] I'm willing to submit a PR!
Contributor guide
No contributing guide indexed for this repository
Assessment
This issue has not been assessed yet.