[jira] [Commented] (FLINK-24369) Resource Leak may happen if flink cluster runs abnormally and repeatedly retry

["Yao Zhang (Jira)" <ji...@apache.org>]

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

Yao Zhang commented on FLINK-24369:
-----------------------------------

Hi [~turtlebin],

I am not quite sure that what has been changed on Flink 1.13.0. But I suggest we might need a shutdown hook that ensure each operator is closed before JVM terminates. Does anyone else have better solutions?

> Resource Leak may happen if flink cluster runs abnormally and repeatedly retry
> ------------------------------------------------------------------------------
>
>                 Key: FLINK-24369
>                 URL: https://issues.apache.org/jira/browse/FLINK-24369
>             Project: Flink
>          Issue Type: Bug
>          Components: API / DataStream
>    Affects Versions: 1.13.0
>         Environment: flink version:1.13.0
> redisson version:3.16.1
>            Reporter: turtlebin
>            Priority: Critical
>   Original Estimate: 24h
>  Remaining Estimate: 24h
>
> I encountered a problem in the process of integrating Flink and Redisson. When the task encounters abnormalities and keeps retries, it will cause the number of Redis Clients to increase volatility (sometimes the number increases, sometimes the number decreases, but the overall trend is growth). Even if I shutdown the Redisson Instance by overwriting the close function , the number of Redis-Clients cannot be prevented from continuing to grow, and eventually the number of Clients will reach the upper limit and an error will be thrown. Moreover, this situation only occurs in the Flink cluster operation mode with version 1.13.0 or above, and the number of Redis-Clients will remain stable in the local mode or in the cluster mode with version under1.12. The test code is below. I wonder if you can provide specific reasons and solutions for this situation, thank you.
> {code:java}
> import org.apache.flink.api.common.serialization.SimpleStringSchema;
> import org.apache.flink.configuration.Configuration;
> import org.apache.flink.streaming.api.TimeCharacteristic;
> import org.apache.flink.streaming.api.datastream.DataStream;
> import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
> import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
> import org.apache.kafka.clients.consumer.ConsumerConfig;
> import java.util.Properties;
> import java.util.Random;
> public class ExceptionTest {
>     public static void main(String[] args) throws Exception{
>         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(new Configuration());
>         env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);
>         env.enableCheckpointing(1000 * 60);
>         DataStream<String> mock = createDataStream(env);
>         mock.keyBy(x -> new Random().nextInt(20))
>                 .process(new ExceptionTestFunction())
>                 .uid("batch-query-key-process")
>                 .filter(x->x!=null)
>                 .print();
>         env.execute("Exception-Test");
>     }
>     private static DataStream<String> createDataStream(StreamExecutionEnvironment env) {
>         String topic = "test_topic_xhb03";
>         Properties test = new Properties();
>         test.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "broker");
>         test.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "group");
>         FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<String>(topic, new SimpleStringSchema(), test);
>         consumer.setStartFromLatest();
>         DataStream<String> source = env.addSource(consumer);
>         return source;
>     }
> }
> import lombok.extern.slf4j.Slf4j;
> import org.apache.flink.configuration.Configuration;
> import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
> import org.apache.flink.util.Collector;
> import org.redisson.Redisson;
> import org.redisson.api.RedissonRxClient;
> import org.redisson.config.Config;
> @Slf4j
> public class ExceptionTestFunction extends KeyedProcessFunction<Integer, String, String> {
>     private RedissonRxClient redisson;
>     @Override
>     public void close() {
>         this.redisson.shutdown();
>         log.info(String.format("Shut down redisson instance in close method, RedissonRxClient shutdown is %s", redisson.isShutdown()));
>     }
>    
>     @Override
>     public void open(Configuration parameters) {
>         String prefix = "redis://";
>         Config config = new Config();
>                         config.useSingleServer()
>                                 .setClientName("xhb-redisson-main")
>                                 .setTimeout(5000)
>                                 .setConnectTimeout(10000)
>                                 .setConnectionPoolSize(4)
>                                 .setConnectionMinimumIdleSize(2)
>                                 .setIdleConnectionTimeout(10000)
>                                 .setAddress("127.0.0.1:6379")
>                                 .setDatabase(0)
>                                 .setPassword(null);
>         this.redisson = Redisson.create(config).rxJava();
>     }
>     @Override
>     public void processElement(String value, Context ctx, Collector<String> out) throws Exception {
>         throw new NullPointerException("Null Pointer in ProcessElement");
>     }
> }
> {code}
>  
>  



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