Aleksandr Pilipenko created FLINK-33176:
-------------------------------------------
Summary: Kinesis source throws NullPointerException in Table API
on ignored parsing errors
Key: FLINK-33176
URL: https://issues.apache.org/jira/browse/FLINK-33176
Project: Flink
Issue Type: Bug
Components: Connectors / Kinesis
Affects Versions: aws-connector-4.1.0, 1.15.4
Reporter: Aleksandr Pilipenko
Using following example table:
{code:java}
CREATE TABLE source (
text STRING,
`arrival_time` TIMESTAMP(3) METADATA FROM 'timestamp' VIRTUAL
) WITH (
'connector' = 'kinesis',
'stream' = 'test',
'aws.region' = 'us-east-1',
'json.ignore-parse-errors' = 'true',
'format' = 'json'
) {code}
Connector throws NullPointerException when source consumes malformed json
message:
{code:java}
java.lang.NullPointerException
at
org.apache.flink.streaming.connectors.kinesis.table.RowDataKinesisDeserializationSchema.deserialize(RowDataKinesisDeserializationSchema.java:137)
at
org.apache.flink.streaming.connectors.kinesis.table.RowDataKinesisDeserializationSchema.deserialize(RowDataKinesisDeserializationSchema.java:44)
at
org.apache.flink.streaming.connectors.kinesis.internals.ShardConsumer.deserializeRecordForCollectionAndUpdateState(ShardConsumer.java:202)
at
org.apache.flink.streaming.connectors.kinesis.internals.ShardConsumer.lambda$run$0(ShardConsumer.java:126)
at
org.apache.flink.streaming.connectors.kinesis.internals.publisher.polling.PollingRecordPublisher.run(PollingRecordPublisher.java:118)
at
org.apache.flink.streaming.connectors.kinesis.internals.publisher.polling.PollingRecordPublisher.run(PollingRecordPublisher.java:102)
at
org.apache.flink.streaming.connectors.kinesis.internals.ShardConsumer.run(ShardConsumer.java:114)
at
java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
at
java.base/java.util.concurrent.FutureTask.run$$$capture(FutureTask.java:264)
at java.base/java.util.concurrent.FutureTask.run(FutureTask.java)
at
java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
at
java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
at java.base/java.lang.Thread.run(Thread.java:829) {code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
