| 69 | } |
| 70 | |
| 71 | Status HdfsFsCache::GetConnection(const string& path, hdfsFS* fs, HdfsFsMap* local_cache, |
| 72 | const HdfsConnOptions* options) { |
| 73 | string err; |
| 74 | const string& namenode = GetNameNodeFromPath(path, &err); |
| 75 | if (!err.empty()) return Status(err); |
| 76 | DCHECK(!namenode.empty()); |
| 77 | |
| 78 | // First, check the local cache to avoid taking the global lock. |
| 79 | if (local_cache != NULL) { |
| 80 | HdfsFsMap::iterator local_iter = local_cache->find(namenode); |
| 81 | if (local_iter != local_cache->end()) { |
| 82 | *fs = local_iter->second; |
| 83 | return Status::OK(); |
| 84 | } |
| 85 | } |
| 86 | // Otherwise, check the global cache. |
| 87 | { |
| 88 | lock_guard<mutex> l(lock_); |
| 89 | HdfsFsMap::iterator i = fs_map_.find(namenode); |
| 90 | if (i == fs_map_.end()) { |
| 91 | hdfsBuilder* hdfs_builder = hdfsNewBuilder(); |
| 92 | hdfsBuilderSetNameNode(hdfs_builder, namenode.c_str()); |
| 93 | if (!s3a_access_key_.empty() || (options != nullptr && !options->empty())) { |
| 94 | // Use a new instance of the filesystem object to be sure that it picks up the |
| 95 | // configuration changes we're going to make. Without this call, a cached |
| 96 | // filesystem object is used which is unaffected by calls to |
| 97 | // hdfsBuilderConfSetStr(). This is unexpected behavior in the HDFS API, but is |
| 98 | // unlikely to change. |
| 99 | hdfsBuilderSetForceNewInstance(hdfs_builder); |
| 100 | if (!s3a_access_key_.empty()) { |
| 101 | hdfsBuilderConfSetStr( |
| 102 | hdfs_builder, "fs.s3a.access.key", s3a_access_key_.c_str()); |
| 103 | hdfsBuilderConfSetStr( |
| 104 | hdfs_builder, "fs.s3a.secret.key", s3a_secret_key_.c_str()); |
| 105 | } |
| 106 | if (options != nullptr) { |
| 107 | for (const auto& option : *options) { |
| 108 | hdfsBuilderConfSetStr( |
| 109 | hdfs_builder, option.first.c_str(), option.second.c_str()); |
| 110 | } |
| 111 | } |
| 112 | } |
| 113 | *fs = hdfsBuilderConnect(hdfs_builder); |
| 114 | if (*fs == NULL) { |
| 115 | return Status(GetHdfsErrorMsg("Failed to connect to FS: ", namenode)); |
| 116 | } |
| 117 | fs_map_.insert(make_pair(namenode, *fs)); |
| 118 | } else { |
| 119 | *fs = i->second; |
| 120 | } |
| 121 | } |
| 122 | DCHECK(*fs != NULL); |
| 123 | // Populate the local cache for the next lookup. |
| 124 | if (local_cache != NULL) { |
| 125 | local_cache->insert(make_pair(namenode, *fs)); |
| 126 | } |
| 127 | return Status::OK(); |
| 128 | } |