@Override
  public void close() throws InterruptedException {
    synchronized (this) { //avoid concurrent .close()
      if (state != STATE_CLOSED) {
        housekeeperThread.interrupt();
        housekeeperThread.join();
        state = STATE_CLOSED;
      }
    }
  }
@Override
  public void close()