public void send(CoordinatorStreamMessage message){
  log.debug("Sending {}",message);
  try {
    String source=message.getSource();
    byte[] key=keySerde.toBytes(Arrays.asList(message.getKeyArray()));
    byte[] value=null;
    if (!message.isDelete()) {
      value=messageSerde.toBytes(message.getMessageMap());
    }
    OutgoingMessageEnvelope envelope=new OutgoingMessageEnvelope(systemStream,Integer.valueOf(0),key,value);
    systemProducer.send(source,envelope);
  }
 catch (  Exception e) {
    throw new SamzaException(e);
  }
}

public void send ( CoordinatorStreamMessage message ) { log . debug ( "Sending {}" , message ) ; try { String source = message . getSource ( ) ; byte [ ] key = keySerde . toBytes ( Arrays . asList ( message . getKeyArray ( ) ) ) ; byte [ ] value = null ; if ( ! message . isDelete ( ) ) { value = messageSerde . toBytes ( message . getMessageMap ( ) ) ; } OutgoingMessageEnvelope envelope = new OutgoingMessageEnvelope ( systemStream , Integer . valueOf ( 0 ) , key , value ) ; systemProducer . send ( source , envelope ) ; } catch ( Exception e ) { throw new SamzaException ( e ) ; } }
serialize and send a coordinator stream message .
