@ Override public void publish ( String subject , byte [ ] data ) throws IOException , InterruptedException , TimeoutException { final BlockingQueue < String > ch = createErrorChannel ( ) ; publish ( subject , data , null , ch ) ; String err ; if ( ! ch . isEmpty ( ) ) { err = ch . take ( ) ; if ( ! err . isEmpty ( ) ) { throw new IOException ( err ) ; } } }
Publish will publish to the cluster and wait for an ACK .