title: Exception in flatmap function when deploying to cluster
body: <p>I have a flink-ignite application. I receive messages from kafka and process the mesages and then cache to ignite. when i run program in ide(intellij) and stand-alone jar there is no problem but when i deploy to cluster i got this exception (I have created the table earlier in the code.). Thanks in advance.
  Note that connection variables are static in my main class.</p>
  
  <pre><code>   Caused by: java.lang.NullPointerException
          at altosis.flinkcompute.compute.Main$2.flatMap(Main.java:95)
          at altosis.flinkcompute.compute.Main$2.flatMap(Main.java:85)
          at org.apache.flink.streaming.api.operators.StreamFlatMap.processElement(StreamFlatMap.java:50)
          at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.pushToOperator(OperatorChain.java:579)
          ... 22 more
  </code></pre>
  
  <pre><code>            StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
              environment.getConfig();
              environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
              environment.setParallelism(1);
              Properties props = new Properties();
              props.setProperty("bootstrap.servers", "localhost:9092");
              props.setProperty("group.id","event-group");
  
              FlinkKafkaConsumer&lt;EventSalesQuantity&gt; consumer = new FlinkKafkaConsumer&lt;EventSalesQuantity&gt;("EventTopic",new EventSerializationSchema(),props);
              DataStream&lt;EventSalesQuantity&gt; eventDataStream = environment.addSource(consumer);
  
              KeyedStream&lt;EventSalesQuantity, String&gt; keyedEventStream = eventDataStream.assignTimestampsAndWatermarks(
                      new AssignerWithPeriodicWatermarksImpl()
              ).
                      keyBy(new KeySelector&lt;EventSalesQuantity, String&gt;() {
                          @Override
                          public String getKey(EventSalesQuantity eventSalesQuantity) throws Exception {
                              return  eventSalesQuantity.getDealer();
                          }
                      });
  
              DataStream&lt;Tuple2&lt;EventSalesQuantity,Integer&gt;&gt; eventSinkStream = keyedEventStream.window(TumblingEventTimeWindows.of(Time.of(1, TimeUnit.DAYS),Time.hours(21))).aggregate(new AggregateImpl());
              ignite = Ignition.start();
              ClientConfiguration cfg = new ClientConfiguration().setAddresses("127.0.0.1:10800");
              igniteClient = Ignition.startClient(cfg);
  
              System.out.println("&gt;&gt;&gt; Thin client put-get example started.");
              igniteClient.query(
                      new SqlFieldsQuery(String.format(
                              "CREATE TABLE IF NOT EXISTS Eventcache (eventtime VARCHAR PRIMARY KEY, bayi VARCHAR, sales INT ) WITH \"VALUE_TYPE=%s\"",
                              EventSalesQuantity.class.getName()
                      )).setSchema("PUBLIC")
              ).getAll();
  
              eventSinkStream.addSink(new FlinkKafkaProducer&lt;Tuple2&lt;EventSalesQuantity, Integer&gt;&gt;("localhost:9092","SinkEventTopic",new EventSinkSerializationSchema()));
              Class.forName("org.apache.ignite.IgniteJdbcThinDriver");
  
              conn = DriverManager.getConnection("jdbc:ignite:thin://127.0.0.1/");
              eventSinkStream.flatMap(new FlatMapFunction&lt;Tuple2&lt;EventSalesQuantity, Integer&gt;, Object&gt;() {
                  @Override
                  public void flatMap(Tuple2&lt;EventSalesQuantity, Integer&gt; eventSalesQuantityIntegerTuple2, Collector&lt;Object&gt; collector) throws Exception {
                      Ignsql= conn.prepareStatement(
                              "INSERT INTO Eventcache (eventtime, bayi, sales) VALUES (?, ?, ?)");
  
                      Ignsql.setString(1, eventSalesQuantityIntegerTuple2.f0.getTransactionDate());
                      Ignsql.setString(2, eventSalesQuantityIntegerTuple2.f0.getDealer());
                      Ignsql.setInt(3, eventSalesQuantityIntegerTuple2.f1);
                      Ignsql.execute();
                      Ignsql.close();
                  }
              });
  
             // eventSinkStream.print();
              environment.execute();```
  </code></pre>