code: @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);
              }
          }
      }
docs: Publish will publish to the cluster and wait for an ACK.