called when new data arrives to the sink , and forwards it to kafka . concode_field_sep boolean initDone concode_elem_sep String topicId concode_elem_sep boolean closeWithoutSend concode_elem_sep String brokerAddr concode_elem_sep kafka.javaapi.producer.Producer<Integer,OUT> producer concode_elem_sep long serialVersionUID concode_elem_sep boolean sendAndClose concode_elem_sep Properties props concode_field_sep OUT serialize concode_elem_sep void closeWithoutSend concode_elem_sep void sendAndClose concode_elem_sep void initialize
void function ( IN arg0 ) { if ( ! initDone ) { initialize ( ) ; } OUT loc0 = serialize ( arg0 ) ; KeyedMessage < Integer , OUT > loc1 = new KeyedMessage < Integer , OUT > ( topicId , loc0 ) ; if ( ! closeWithoutSend ) { producer . send ( loc1 ) ; } if ( sendAndClose ) { producer . close ( ) ; } }