Flink RedisRichSinkFunction

public class RedisRichSinkFunction<IN> extends RichSinkFunction<IN> {
   private JedisPool pool = null;

   @Override
   public void open(Configuration parameters) throws Exception {
      ParameterTool parameterTool = (ParameterTool) getRuntimeContext().getExecutionConfig().getGlobalJobParameters();
      String hostname = parameterTool.get("redis_hostname");
      int port = Integer.parseInt(parameterTool.get("redis_port"));
      pool = RedisPoolFactory.getPool(hostname, port);
   }

   public JedisPool getPool() {
      return pool;
   }

}