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