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);
    }
  }
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.
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 .