void pushDataMessage(GTSEncoder encoder) throws IOException {
  if (null != encoder) {
    KafkaDataMessage msg=new KafkaDataMessage();
    msg.setType(KafkaDataMessageType.STORE);
    msg.setData(encoder.getBytes());
    msg.setClassId(encoder.getClassId());
    msg.setLabelsId(encoder.getLabelsId());
    sendDataMessage(msg);
  }
 else {
    sendDataMessage(null);
  }
}

void pushDataMessage ( GTSEncoder encoder ) throws IOException { if ( null != encoder ) { KafkaDataMessage msg = new KafkaDataMessage ( ) ; msg . setType ( KafkaDataMessageType . STORE ) ; msg . setData ( encoder . getBytes ( ) ) ; msg . setClassId ( encoder . getClassId ( ) ) ; msg . setLabelsId ( encoder . getLabelsId ( ) ) ; sendDataMessage ( msg ) ; } else { sendDataMessage ( null ) ; } }
push a metadata message onto the buffered list of kafka messages and flush the list to kafka if it has reached a threshold .
