| 51 | |
| 52 | public: |
| 53 | HdfsFileInputStream(std::string filename, ReaderMetrics* metrics) : metrics_(metrics) { |
| 54 | filename_ = filename; |
| 55 | |
| 56 | // Building a URI object from the given uri_path |
| 57 | hdfs::URI uri; |
| 58 | try { |
| 59 | uri = hdfs::URI::parse_from_string(filename_); |
| 60 | } catch (const hdfs::uri_parse_error&) { |
| 61 | throw ParseError("Malformed URI: " + filename_); |
| 62 | } |
| 63 | |
| 64 | // This sets conf path to default "$HADOOP_CONF_DIR" or "/etc/hadoop/conf" |
| 65 | // and loads configs core-site.xml and hdfs-site.xml from the conf path |
| 66 | hdfs::ConfigParser parser; |
| 67 | if (!parser.LoadDefaultResources()) { |
| 68 | throw ParseError("Could not load default resources. "); |
| 69 | } |
| 70 | auto stats = parser.ValidateResources(); |
| 71 | // validating core-site.xml |
| 72 | if (!stats[0].second.ok()) { |
| 73 | throw ParseError(stats[0].first + " is invalid: " + stats[0].second.ToString()); |
| 74 | } |
| 75 | // validating hdfs-site.xml |
| 76 | if (!stats[1].second.ok()) { |
| 77 | throw ParseError(stats[1].first + " is invalid: " + stats[1].second.ToString()); |
| 78 | } |
| 79 | hdfs::Options options; |
| 80 | if (!parser.get_options(options)) { |
| 81 | throw ParseError("Could not load Options object. "); |
| 82 | } |
| 83 | hdfs::IoService* io_service = hdfs::IoService::New(); |
| 84 | // Wrapping file_system into a unique pointer to guarantee deletion |
| 85 | fileSystem_ = |
| 86 | std::unique_ptr<hdfs::FileSystem>(hdfs::FileSystem::New(io_service, "", options)); |
| 87 | if (fileSystem_.get() == nullptr) { |
| 88 | throw ParseError("Can't create FileSystem object. "); |
| 89 | } |
| 90 | hdfs::Status status; |
| 91 | // Checking if the user supplied the host |
| 92 | if (!uri.get_host().empty()) { |
| 93 | // Using port if supplied, otherwise using "" to look up port in configs |
| 94 | std::string port = uri.has_port() ? std::to_string(uri.get_port()) : ""; |
| 95 | status = fileSystem_->Connect(uri.get_host(), port); |
| 96 | if (!status.ok()) { |
| 97 | throw ParseError("Can't connect to " + uri.get_host() + ":" + port + ". " + |
| 98 | status.ToString()); |
| 99 | } |
| 100 | } else { |
| 101 | status = fileSystem_->ConnectToDefaultFs(); |
| 102 | if (!status.ok()) { |
| 103 | if (!options.defaultFS.get_host().empty()) { |
| 104 | throw ParseError("Error connecting to " + options.defaultFS.str() + ". " + |
| 105 | status.ToString()); |
| 106 | } else { |
| 107 | throw ParseError("Error connecting to the cluster: defaultFS is empty. " + |
| 108 | status.ToString()); |
| 109 | } |
| 110 | } |
nothing calls this directly
no test coverage detected