ElasticsearchSink Serialization Error

[Vijay Srinivasaraghavan]

Hello,
I am seeing below error when I try to use ElasticsearchSink. It complains about serialization and looks like it is leading to "IndexRequestBuilder" implementation. I have tried the suggestion as mentioned in http://stackoverflow.com/questions/33246864/elasticsearch-sink-seralizability (changed from anonymous class to concrete class) but it did not help. However, when I call "ElasticsearchSink<>(config, transports, null)" by passing "null" for "IndexRequestBuilder" then I don't see the serialization error. This suggests the problem could be with the IndexRequestBuilder implementation but I am not able to move further.
Could someone please let me know what's the right way to use ElasticsearchSink() API? 
Build DetailsFlink 1.2.0Elastic Search 5.3.0

Error Message

org.apache.flink.api.common.InvalidProgramException: The implementation of the RichSinkFunction is not serializable. The object probably contains or references non serializable fields.        at org.apache.flink.api.java.ClosureCleaner.clean(ClosureCleaner.java:100)        at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.clean(StreamExecutionEnvironment.java:1539)        at org.apache.flink.streaming.api.datastream.DataStream.clean(DataStream.java:161)        at org.apache.flink.streaming.api.datastream.DataStream.addSink(DataStream.java:1076)
Code Snippet
``` private ElasticsearchSink<Result>  sinkToElasticSearch(AppConfiguration appConfiguration) throws Exception {
 String host = appConfiguration.getPipeline().getElasticSearch().getHost(); int port = appConfiguration.getPipeline().getElasticSearch().getPort(); String cluster = appConfiguration.getPipeline().getElasticSearch().getCluster();
 Map<String, String> config = new HashMap<>(); config.put("bulk.flush.max.actions", "1"); config.put("cluster.name", cluster);
 List<TransportAddress> transports = new ArrayList<>(); transports.add(new InetSocketTransportAddress(host, port));
 return new ElasticsearchSink<>(config, transports, new ResultIndexRequestBuilder(appConfiguration)); }
 public class ResultIndexRequestBuilder implements IndexRequestBuilder<Result>, Serializable {
 private String index; private String type; //private transient Gson gson = new Gson();
 public ResultIndexRequestBuilder() {}
 public ResultIndexRequestBuilder(AppConfiguration appConfiguration) { index = appConfiguration.getPipeline().getElasticSearch().getIndex(); type = appConfiguration.getPipeline().getElasticSearch().getType(); }
 @Override public IndexRequest createIndexRequest(Result result, RuntimeContext ctx) { Gson gson = new Gson(); String resultAsJson = gson.toJson(result); System.out.println(resultAsJson); Map<String, String> jsonMap = new HashMap<>(); jsonMap.put("data", resultAsJson);
 return Requests.indexRequest() .index(index) .type(type) .source(jsonMap); }```
RegardsVijay