MyFn

MyFn.java

public class MyFn extends DoFn<String, String> {

    private  static RedissonClient redisClient;
    private final static Object lock = new Object();
    private String redisHost;
    private int redisPort;

  
   MyFn(String redisHost, Integer redisPort) {
        this.redisHost = redisHost;
        this.redisPort = redisPort;
    }
  
   private RedissonClient getRedisClient() {

          if (redisClient == null) {
            synchronized (lock) {
                if (redisClient == null) {
                    Config config = new Config();
                    String password = decryptRedisPassword();
                    config.useSingleServer()
                            .setAddress("redis://" + redisHost + ":" + redisPort);
                    redisClient = Redisson.create(config);
                }
            }
        }
        return redisClient;

    }
  
  @ProcessElement
   public void processElement(ProcessContext context) {
      
    RedissonClient redis = getRedisClient();
    //some logic enrichment
    redis.get() 
    // output 
   }