cluster-common/src/main/java/com/streamsets/pipeline/cluster/ControlChannel.java
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();
    }
  }
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();
    }
  }
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.
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 .