title: Flink: java.io.NotSerializableException: redis.clients.jedis.JedisCluster
body: <p>When I submit new flink job, it throws</p>
  
  <pre><code>Caused by: java.io.NotSerializableException: redis.clients.jedis.JedisCluster
      at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1184)
      at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
      at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
      at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
      at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
      at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:348)
      at org.apache.flink.util.InstantiationUtil.serializeObject(InstantiationUtil.java:512)
      at org.apache.flink.api.java.ClosureCleaner.clean(ClosureCleaner.java:81)
      ... 24 more
  </code></pre>
  
  <p>this is my code:</p>
  
  <pre><code>    JedisCluster jedisCluster = JedisClusterBuilder.getInstance(JedisClusterEnum.THIRD);
  
  
      DataStream&lt;MobileClickEvent&gt; clickEventDataStream = environment.addSource(clickConsumer);
  
  
      clickEventDataStream
              .filter(Objects::nonNull)
              .keyBy(new KeySelector&lt;MobileClickEvent, String&gt;() {
                  @Override
                  public String getKey(MobileClickEvent value) throws Exception {
                      return value.getItemId() + "_" + value.getItemType();
                  }
              })
              .process(new KeyedProcessFunction&lt;String, MobileClickEvent, Object&gt;() {
                  @Override
                  public void processElement(MobileClickEvent value, Context ctx, Collector&lt;Object&gt; out) throws Exception {
                      String key = ctx.getCurrentKey();
                      jedisCluster.hincrBy("{item_feature}" + key, "click", 1);
                      jedisCluster.expire("{item_feature}" + key, 60 * 10);
                  }
              });
  </code></pre>