title: The implementation of the provided ElasticsearchSinkFunction is not serializable(flink-connector-elasticsearch6_2.11)
body: <p>"non-serializable" error occurs when I follow <a href="https://ci.apache.org/projects/flink/flink-docs-stable/dev/connectors/elasticsearch.html" rel="nofollow noreferrer">flink document</a> to write data via flink streaming. I use flink1.6,Elastic-Search-6.4 and flink-connector-elasticsearch6.<br>
  My code is like </p>
  
  <p><div class="snippet" data-lang="js" data-hide="false" data-console="true" data-babel="false">
  <div class="snippet-code">
  <pre class="snippet-code-html lang-html prettyprint-override"><code>@Test
    public void testStringInsert() throws Exception {
      StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
      env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
      env.enableCheckpointing(100); //
      DataStreamSource&lt;String&gt; input = env.fromCollection(Collections.singleton("testData"));
      List&lt;HttpHost&gt; httpHosts = new ArrayList&lt;&gt;();
      httpHosts.add(new HttpHost("127.0.0.1", 9200, "http"));
      ElasticsearchSink.Builder&lt;String&gt; esSinkBuilder = new ElasticsearchSink.Builder&lt;&gt;(
          httpHosts,
          new ElasticsearchSinkFunction&lt;String&gt; ()  {
            public IndexRequest createIndexRequest(String element) {
              Map&lt;String, String&gt; json = new HashMap&lt;&gt;();
              json.put("data", element);
              return Requests.indexRequest()
                  .index("my-index")
                  .type("my-type")
                  .source(json);
            }
            @Override
            public void process(String element, RuntimeContext ctx, RequestIndexer indexer) {
              indexer.add(createIndexRequest(element));
            }
          }
      );
      esSinkBuilder.setBulkFlushMaxActions(1);
      input.addSink(esSinkBuilder.build());
      env.execute("test es string insert");
    }</code></pre>
  </div>
  </div>
  </p>
  
  <p>When I run the code above,I got the exception </p>
  
  <p><div class="snippet" data-lang="js" data-hide="false" data-console="true" data-babel="false">
  <div class="snippet-code">
  <pre class="snippet-code-html lang-html prettyprint-override"><code>java.lang.IllegalArgumentException: The implementation of the provided ElasticsearchSinkFunction is not serializable. The object probably contains or references non-serializable fields.
  
   at org.apache.flink.util.Preconditions.checkArgument(Preconditions.java:139)
   at org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkBase.&lt;init&gt;(ElasticsearchSinkBase.java:216)
   at org.apache.flink.streaming.connectors.elasticsearch6.ElasticsearchSink.&lt;init&gt;(ElasticsearchSink.java:71)
   at org.apache.flink.streaming.connectors.elasticsearch6.ElasticsearchSink.&lt;init&gt;(ElasticsearchSink.java:60)
   at org.apache.flink.streaming.connectors.elasticsearch6.ElasticsearchSink$Builder.build(ElasticsearchSink.java:208)
   at com.lianlianpay.erebus.erebusaccess.ElasticsearchSinkTest.testStringInsert(ElasticsearchSinkTest.java:151)
   at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
   at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
   at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
   at java.lang.reflect.Method.invoke(Method.java:498)
   at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)
   at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
   at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)
   at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
   at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:325)
   at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:78)
   at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:57)
   at org.junit.runners.ParentRunner$3.run(ParentRunner.java:290)
   at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71)
   at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288)
   at org.junit.runners.ParentRunner.access$000(ParentRunner.java:58)
   at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268)
   at org.junit.runners.ParentRunner.run(ParentRunner.java:363)
   at org.junit.runner.JUnitCore.run(JUnitCore.java:137)
   at com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:68)
   at com.intellij.rt.execution.junit.IdeaTestRunner$Repeater.startRunnerWithArgs(IdeaTestRunner.java:51)
   at com.intellij.rt.execution.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:237)
   at com.intellij.rt.execution.junit.JUnitStarter.main(JUnitStarter.java:70)</code></pre>
  </div>
  </div>
  </p>
  
  <p>I searched a lot and still puzzled.I did not pass Object to ElasticsearchSink,other than String,and String is serializable without a doubt.I really can not understand , what is wrong with my code or the develop enviroment ? </p>