@Override
    public void process(Tuple2 tuple2, RuntimeContext ctx, RequestIndexer indexer) {
        try{
            Tuple2<Boolean, Row> tupleTrans = tuple2;
            Boolean retract = tupleTrans.getField(0);
            Row element = tupleTrans.getField(1);
            if(!retract){
                return;
            }


            indexer.add(createIndexRequest(element));
            outRecords.inc();
        }catch (Throwable e){
            logger.error("", e);
        }
    }
Reason:
Date: 2017/7/19
Company: www.dtstack.com
@author xuchao
elasticsearch5/elasticsearch5-sink/src/main/java/com/dtstack/flink/sql/sink/elasticsearch/CustomerSinkFunc.java