code: @Override
    public void close() {
      log.info("Updating the current state high water mark with " + this.highWatermark);
      this.workUnitState.setActualHighWatermark(new LongWatermark(this.highWatermark));
      try {
        this.closeConnection();
      } catch (Exception e) {
        log.error("Failed to close the extractor", e);
      }
    }
docs: close extractor read stream
  update high watermark