code: @Override
      @SneakyThrows(IOException.class)
      public void initialize(long epoch) {
          Exceptions.checkNotClosed(this.closed.get(), this);
          Preconditions.checkState(this.fileSystem == null, "HDFSStorage has already been initialized.");
          Preconditions.checkArgument(epoch > 0, "epoch must be a positive number. Given %s.", epoch);
          Configuration conf = new Configuration();
          conf.set("fs.default.name", this.config.getHdfsHostURL());
          conf.set("fs.default.fs", this.config.getHdfsHostURL());
          conf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");
  
          // FileSystem has a bad habit of caching Clients/Instances based on target URI. We do not like this, since we
          // want to own our implementation so that when we close it, we don't interfere with others.
          conf.set("fs.hdfs.impl.disable.cache", "true");
          if (!this.config.isReplaceDataNodesOnFailure()) {
              // Default is DEFAULT, so we only set this if we want it disabled.
              conf.set("dfs.client.block.write.replace-datanode-on-failure.policy", "NEVER");
          }
  
          this.epoch = epoch;
          this.fileSystem = openFileSystem(conf);
          log.info("Initialized (HDFSHost = '{}', Epoch = {}).", this.config.getHdfsHostURL(), epoch);
      }
docs: region Storage Implementation