[jira] [Commented] (FLINK-23132) flink upgrade issue(1.11.3->1.13.0)

["Martijn Visser (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/FLINK-23132?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17368698#comment-17368698 ] 

Martijn Visser commented on FLINK-23132:
----------------------------------------

Hi, thanks for opening a ticket. It would be great if you could first reach out to the [User ML|https://flink.apache.org/community.html] and see if it could get resolved there. I'm closing this as "Abandoned". We can always re-open the ticket if it turns out that's needed. 

> flink upgrade issue(1.11.3->1.13.0)
> -----------------------------------
>
>                 Key: FLINK-23132
>                 URL: https://issues.apache.org/jira/browse/FLINK-23132
>             Project: Flink
>          Issue Type: Bug
>            Reporter: Jeff Hu
>            Priority: Major
>
>  
> In order to improve the performance of data process, we store events to a map and do not process them untill event count reaches 100. in the meantime, start a timer in open method, so data is processed every 60 seconds
> this works when flink version is *1.11.3*,
> after upgrading flink version to *1.13.0*
> I found sometimes events were consumed from Kafka continuously, but were not processed in RichFlatMapFunction, it means data was missing. after restarting service, it works well, but several hours later the same thing happened again.
> any known issue for this flink version? any suggestions are appreciated.
>  {{public class MyJob \{
>     public static void main(String[] args) throws Exception {
>         ...
>         DataStream<String> rawEventSource = env.addSource(flinkKafkaConsumer);
>         ...
>     }}} {{public class MyMapFunction extends RichFlatMapFunction<String, String> implements Serializable \{
>     @Override
>     public void open(Configuration parameters) {
>         ...
>         long periodTimeout = 60;
>         pool.scheduleAtFixedRate(() -> {
>             // processing data
>         }, periodTimeout, periodTimeout, TimeUnit.SECONDS);
>     }
>     
>     @Override
>     public void flatMap(String message, Collector<String> out) \{
>         // store event to map 
>         // count event, 
>         // when count = 100, start data processing
>     }
> }}}



--
This message was sent by Atlassian Jira
(v8.3.4#803005)