Flink es sink的一次历险

不谈需求的铺代码==耍流氓。。。

需求（概略）：从kafka读数据=》业务处理=》写入es=》写入kafka（通过kafka通知其他业务：某条数据的id,以及所在es的索引）

第一版代码（摘要）：

//写入es        
mapDs.addSink(new ElasticsearchSink<JSONObject>(esProps, transportAddresses, (ElasticsearchSinkFunction<JSONObject>) (jsonObject, runtimeContext, requestIndexer) -> {
            long createTime = jsonObject.getLong("create_time");
            String indexName = esPrefix.concat(Utils.longDate2String(createTime, "yyyy.MM.dd"));
            requestIndexer.add(Requests.indexRequest().index(indexName)
                    .type(esType)
                    .source(jsonObject));
        }, new RetryRejectedExecutionFailureHandler())).name("sop-controller-es-sink");

//写入kafka
mapDs.filter(jsonObject -> "初始化".equals(jsonObject.getString("