cluster-common/src/main/java/com/streamsets/pipeline/cluster/Consumer.java
public OffsetAndResult<Map.Entry> take() {
    if (producerError != null) {
      throw new RuntimeException(Utils.format("Producer encountered error: {}", producerError), producerError);
    }
    if (consumerError != null) {
      throw new RuntimeException(Utils.format("Consumer encountered error: {}", consumerError), consumerError);
    }
    try {
      Utils.checkState(batchCommitted, "Cannot take messages when last batch is uncommitted");
      while (running) {
        for (ControlChannel.Message controlMessage : controlChannel.getConsumerMessages()) {
          switch (controlMessage.getType()) {
            case PRODUCER_COMPLETE:
              // producer is complete, empty channel and afterwards return null
              running = false;
              break;
            case PRODUCER_ERROR:
              running = false;
              Throwable throwable = (Throwable) controlMessage.getPayload();
              producerError = throwable;
              throw new ProducerRuntimeException(Utils.format("Producer encountered error: {}", throwable), throwable);
            default:
              String msg = Utils.format("Illegal control message type: '{}'", controlMessage.getType());
              throw new IllegalStateException(msg);
          }
        }
        OffsetAndResult<Map.Entry> batch = dataChannel.take(10, TimeUnit.MILLISECONDS);
        LOG.trace("Received batch: {}", batch);
        if (batch != null) {
          batchCommitted = false; // got a new batch
          return batch;
        }
      }
      LOG.trace("Returning null");
      return null;
    } catch (Throwable throwable) {
      if (!(throwable instanceof ProducerRuntimeException)) {
        String msg = "Error caught in consumer: " + throwable;
        LOG.error(msg, throwable);
        error(throwable);
      }
      throw Throwables.propagate(throwable);
    }
  }
public OffsetAndResult<Map.Entry> take() {
    if (producerError != null) {
      throw new RuntimeException(Utils.format("Producer encountered error: {}", producerError), producerError);
    }
    if (consumerError != null) {
      throw new RuntimeException(Utils.format("Consumer encountered error: {}", consumerError), consumerError);
    }
    try {
      Utils.checkState(batchCommitted, "Cannot take messages when last batch is uncommitted");
      while (running) {
        for (ControlChannel.Message controlMessage : controlChannel.getConsumerMessages()) {
          switch (controlMessage.getType()) {
            case PRODUCER_COMPLETE:
              // producer is complete, empty channel and afterwards return null
              running = false;
              break;
            case PRODUCER_ERROR:
              running = false;
              Throwable throwable = (Throwable) controlMessage.getPayload();
              producerError = throwable;
              throw new ProducerRuntimeException(Utils.format("Producer encountered error: {}", throwable), throwable);
            default:
              String msg = Utils.format("Illegal control message type: '{}'", controlMessage.getType());
              throw new IllegalStateException(msg);
          }
        }
        OffsetAndResult<Map.Entry> batch = dataChannel.take(10, TimeUnit.MILLISECONDS);
        LOG.trace("Received batch: {}", batch);
        if (batch != null) {
          batchCommitted = false; // got a new batch
          return batch;
        }
      }
      LOG.trace("Returning null");
      return null;
    } catch (Throwable throwable) {
      if (!(throwable instanceof ProducerRuntimeException)) {
        String msg = "Error caught in consumer: " + throwable;
        LOG.error(msg, throwable);
        error(throwable);
      }
      throw Throwables.propagate(throwable);
    }
  }
Consumes messages off the queue. Returns null when the producer
has indicated it is complete and throws an exception
when the consumer producer has indicated it is in error.