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 .