You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Danny Cranmer (Jira)" <ji...@apache.org> on 2023/10/04 16:01:00 UTC
[jira] [Commented] (FLINK-33176) Kinesis source throws NullPointerException in Table API on ignored parsing errors
[ https://issues.apache.org/jira/browse/FLINK-33176?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17771905#comment-17771905 ]
Danny Cranmer commented on FLINK-33176:
---------------------------------------
merged commit [{{5105d48}}|https://github.com/apache/flink-connector-aws/commit/5105d48bf7334503b314d188462c3542f438f1d6] into apache:main
> 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: 1.15.4, aws-connector-4.1.0
> Reporter: Aleksandr Pilipenko
> Assignee: Aleksandr Pilipenko
> Priority: Major
> Labels: pull-request-available
> Fix For: aws-connector-4.2.0
>
>
> 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)