[GitHub] [inlong] wangpeix opened a new issue, #6639: [Bug][Sort] DynamicPulsarDeserializationSchema threw NPE when sourceMetricData is not initialized

[GitBox <gi...@apache.org>]

wangpeix opened a new issue, #6639:
URL: https://github.com/apache/inlong/issues/6639

   ### What happened
   
   When sourceMetricData is not initialized, DynamicPulsarDeserializationSchema throws a null pointer exception.
   
   ### What you expected to happen
   
   ```java
   java.lang.RuntimeException: java.lang.NullPointerException
   	at org.apache.inlong.sort.pulsar.withoutadmin.CallbackCollector.collect(CallbackCollector.java:39)
   	at org.apache.inlong.sort.formats.inlongmsg.InLongMsgDeserializationSchema.emitRow(InLongMsgDeserializationSchema.java:157)
   	at org.apache.inlong.sort.formats.inlongmsg.InLongMsgDeserializationSchema.lambda$deserialize$0(InLongMsgDeserializationSchema.java:113)
   	at java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1380)
   	at java.util.stream.ReferencePipeline$Head.forEach(ReferencePipeline.java:580)
   	at org.apache.inlong.sort.formats.inlongmsg.InLongMsgDeserializationSchema.deserialize(InLongMsgDeserializationSchema.java:113)
   	at org.apache.flink.streaming.util.serialization.ThreadSafeDeserializationSchema.deserialize(ThreadSafeDeserializationSchema.java:53)
   	at org.apache.inlong.sort.pulsar.table.DynamicPulsarDeserializationSchema.deserialize(DynamicPulsarDeserializationSchema.java:128)
   	at org.apache.inlong.sort.pulsar.table.DynamicPulsarDeserializationSchema.deserialize(DynamicPulsarDeserializationSchema.java:119)
   	at org.apache.inlong.sort.pulsar.table.DynamicPulsarDeserializationSchema.deserialize(DynamicPulsarDeserializationSchema.java:47)
   	at org.apache.flink.streaming.connectors.pulsar.internal.ReaderThread.emitRecord(ReaderThread.java:171)
   	at org.apache.flink.streaming.connectors.pulsar.internal.ReaderThread.run(ReaderThread.java:110)
   Caused by: java.lang.NullPointerException
   	at org.apache.inlong.sort.pulsar.table.DynamicPulsarDeserializationSchema.lambda$deserialize$0(DynamicPulsarDeserializationSchema.java:129)
   	at org.apache.inlong.sort.pulsar.withoutadmin.CallbackCollector.collect(CallbackCollector.java:37)
   	... 11 more
   ```
   
   ### How to reproduce
   
   Inlong manager generates the following sql. use this sql can reproduce the problem.
   
   ```sql
   CREATE TABLE `table_stream_01`(
       `uid` STRING,
       `event_time` STRING)
       WITH (
       'inlong.metric.labels' = 'groupId=group_002&streamId=stream_01&nodeId=stream_01',
       'connector' = 'pulsar-inlong',
       'inlong-msg.csv.disable-quote-character' = 'true',
       'inlong-msg.inner.format' = 'csv',
       'inlong-msg.csv.allow-comments' = 'false',
       'format' = 'inlong-msg',
       'inlong-msg.csv.field-delimiter' = ',',
       'inlong-msg.csv.ignore-parse-errors' = 'true',
       'inlong-msg.ignore-parse-errors' = 'false',
       'admin-url' = 'http://x.x.x.x:8080',
       'generic' = 'true',
       'service-url' = 'pulsar://x.x.x.x:6650',
       'topic' = 'public/group_002/stream_01',
       'scan.startup.mode' = 'earliest'
   );
   
   CREATE TABLE `table_sink_01`(
       PRIMARY KEY (`event_time`) NOT ENFORCED,
       `uid` STRING,
       `event_time` STRING)
       WITH (
       'inlong.metric.labels' = 'groupId=group_002&streamId=stream_01&nodeId=sink_01',
       'connector' = 'jdbc-inlong',
       'url' = 'jdbc:mysql://x.x.x.x:3306/test_data',
       'username' = 'xxx',
       'password' = 'xxx',
       'table-name' = 'test_11251558'
   );
   
   INSERT INTO `table_sink_01`
       SELECT
       `uid` AS `uid`,
       `event_time` AS `event_time`
       FROM `table_stream_01`;
   ``` 
   
   ### Environment
   
   _No response_
   
   ### InLong version
   
   master
   
   ### InLong Component
   
   InLong Sort
   
   ### Are you willing to submit PR?
   
   - [X] Yes, I am willing to submit a PR!
   
   ### Code of Conduct
   
   - [X] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct)
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@inlong.apache.org.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [inlong] healchow closed issue #6639: [Bug][Sort] DynamicPulsarDeserializationSchema threw NPE when sourceMetricData is not initialized

[GitBox <gi...@apache.org>]

healchow closed issue #6639: [Bug][Sort] DynamicPulsarDeserializationSchema threw NPE when sourceMetricData is not initialized
URL: https://github.com/apache/inlong/issues/6639


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@inlong.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org