Flink+Redisson??????????????????????????RedisClient????????????????????????????????

Tue, 14 Sep 2021 00:50:38 -0700 

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, and 
the number of Redis-Clients will remain stable in the local mode. The test code 
is below. I wonder if you can provide specific reasons and solutions for this 
situation, thank you.


flink version:1.13.0
redisson version:3.16.1


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);
&nbsp; &nbsp; &nbsp; &nbsp; DataStream<String&gt; mock = createDataStream(env);
&nbsp; &nbsp; &nbsp; &nbsp; mock.keyBy(x -&gt; new Random().nextInt(20))
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .process(new 
ExceptionTestFunction())
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; 
.uid("batch-query-key-process")
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .filter(x-&gt;x!=null)
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .print();
&nbsp; &nbsp; &nbsp; &nbsp; env.execute("Exception-Test");
&nbsp; &nbsp; }


&nbsp; &nbsp; private static DataStream<String&gt; 
createDataStream(StreamExecutionEnvironment env) {
&nbsp; &nbsp; &nbsp; &nbsp; String topic = "test_topic_xhb03";
&nbsp; &nbsp; &nbsp; &nbsp; Properties test = new Properties();
&nbsp; &nbsp; &nbsp; &nbsp; 
test.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "broker");
&nbsp; &nbsp; &nbsp; &nbsp; test.setProperty(ConsumerConfig.GROUP_ID_CONFIG, 
"group");


&nbsp; &nbsp; &nbsp; &nbsp; FlinkKafkaConsumer<String&gt; consumer = new 
FlinkKafkaConsumer<String&gt;(topic, new SimpleStringSchema(), test);
&nbsp; &nbsp; &nbsp; &nbsp; consumer.setStartFromLatest();


&nbsp; &nbsp; &nbsp; &nbsp; DataStream<String&gt; source = 
env.addSource(consumer);
&nbsp; &nbsp; &nbsp; &nbsp; return source;
&nbsp; &nbsp; }
}




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&gt; {
&nbsp; &nbsp; private RedissonRxClient redisson;


&nbsp; &nbsp; @Override
&nbsp; &nbsp; public void close() {
&nbsp; &nbsp; &nbsp; &nbsp; this.redisson.shutdown();
&nbsp; &nbsp; &nbsp; &nbsp; log.info(String.format("Shut down redisson instance 
in close method, RedissonRxClient shutdown is %s", redisson.isShutdown()));


&nbsp; &nbsp; }
&nbsp; &nbsp;&nbsp;
&nbsp; &nbsp; @Override
&nbsp; &nbsp; public void open(Configuration parameters) {
&nbsp; &nbsp; &nbsp; &nbsp; String prefix = "redis://";
&nbsp; &nbsp; &nbsp; &nbsp; Config config = new Config();
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; 
&nbsp; config.useSingleServer()
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; 
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .setClientName("xhb-redisson-main")
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; 
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .setTimeout(5000)
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; 
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .setConnectTimeout(10000)
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; 
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .setConnectionPoolSize(4)
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; 
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .setConnectionMinimumIdleSize(2)
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; 
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .setIdleConnectionTimeout(10000)
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; 
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .setAddress("127.0.0.1:6379")
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; 
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .setDatabase(0)
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; 
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .setPassword(null);
&nbsp; &nbsp; &nbsp; &nbsp; this.redisson = Redisson.create(config).rxJava();
&nbsp; &nbsp; }


&nbsp; &nbsp; @Override
&nbsp; &nbsp; public void processElement(String value, Context ctx, 
Collector<String&gt; out) throws Exception {
&nbsp; &nbsp; &nbsp; &nbsp; throw new NullPointerException("Null Pointer in 
ProcessElement");
&nbsp; &nbsp; }
}

回复