cluster-common/src/main/java/com/streamsets/pipeline/cluster/Producer.java
public Object put(OffsetAndResult<Map.Entry> batch) {
    if (consumerError != null) {
      throw new RuntimeException(Utils.format("Consumer encountered error: {}", consumerError), consumerError);
    }
    if (producerError != null) {
      throw new RuntimeException(Utils.format("Producer encountered error: {}", producerError), producerError);
    }
    try {
      Object expectedOffset = "EMPTY_BATCH";
      if (!batch.getResult().isEmpty()) {
        expectedOffset = batch.getResult().get(batch.getResult().size() - 1).getKey(); // get the last one
      }
      while (!dataChannel.offer(batch, 10, TimeUnit.MILLISECONDS)) {
        for (ControlChannel.Message controlMessage : controlChannel.getProducerMessages()) {
          switch (controlMessage.getType()) {
            case CONSUMER_ERROR:
              Throwable throwable = (Throwable) controlMessage.getPayload();
              consumerError = throwable;
              throw new ConsumerRuntimeException(Utils.format("Consumer encountered error: {}", throwable), throwable);
            default:
              String msg = Utils.format("Illegal control message type: '{}'", controlMessage.getType());
              throw new IllegalStateException(msg);
          }
        }
      }
      return expectedOffset;
    } catch (Throwable throwable) {
      controlChannel.producerComplete();
      if (!(throwable instanceof ConsumerRuntimeException)) {
        String msg = "Error caught in producer: " + throwable;
        LOG.error(msg, throwable);
        controlChannel.producerError(throwable);
        if (producerError == null) {
          producerError = throwable;
        }
      }
      throw Throwables.propagate(throwable);
    }
  }
public Object put(OffsetAndResult<Map.Entry> batch) {
    if (consumerError != null) {
      throw new RuntimeException(Utils.format("Consumer encountered error: {}", consumerError), consumerError);
    }
    if (producerError != null) {
      throw new RuntimeException(Utils.format("Producer encountered error: {}", producerError), producerError);
    }
    try {
      Object expectedOffset = "EMPTY_BATCH";
      if (!batch.getResult().isEmpty()) {
        expectedOffset = batch.getResult().get(batch.getResult().size() - 1).getKey(); // get the last one
      }
      while (!dataChannel.offer(batch, 10, TimeUnit.MILLISECONDS)) {
        for (ControlChannel.Message controlMessage : controlChannel.getProducerMessages()) {
          switch (controlMessage.getType()) {
            case CONSUMER_ERROR:
              Throwable throwable = (Throwable) controlMessage.getPayload();
              consumerError = throwable;
              throw new ConsumerRuntimeException(Utils.format("Consumer encountered error: {}", throwable), throwable);
            default:
              String msg = Utils.format("Illegal control message type: '{}'", controlMessage.getType());
              throw new IllegalStateException(msg);
          }
        }
      }
      return expectedOffset;
    } catch (Throwable throwable) {
      controlChannel.producerComplete();
      if (!(throwable instanceof ConsumerRuntimeException)) {
        String msg = "Error caught in producer: " + throwable;
        LOG.error(msg, throwable);
        controlChannel.producerError(throwable);
        if (producerError == null) {
          producerError = throwable;
        }
      }
      throw Throwables.propagate(throwable);
    }
  }
Queues the batch for the consumer and waits until the consumer
successfully commits the batch. While waiting processes any
control messages from the consumer. Throws an exception
when the consumer has indicated it encountered an error.