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 .