public void consumerCommit ( String offset ) { Object offsetValue = offset ; if ( offsetValue == null ) { offsetValue = new NullOffset ( ) ; } LOG . trace ( "Commit Offset: '{}'" , offsetValue ) ; try { producerQueue . put ( new Message ( MessageType . CONSUMER_COMMIT , offsetValue ) ) ; } catch ( InterruptedException e ) { LOG . info ( "Interrupted while queuing '{}'" , MessageType . CONSUMER_COMMIT . name ( ) , offsetValue ) ; Thread . currentThread ( ) . interrupt ( ) ; } }
If a null value is passed to this method it s replaced with a dummy due to the fact the payload for each message is wrapped in an Optional .