File Coverage

blib/lib/Net/Hadoop/WebHDFS.pm

Criterion	Covered	Total	%
statement	41	355	11.5
branch	8	174	4.6
condition	12	82	14.6
subroutine	13	68	19.1
pod	25	43	58.1
total	99	722	13.7

package Net::Hadoop::WebHDFS;

use strict;
use warnings;
use Carp;

use JSON::XS qw//;

use Furl;
use File::Spec;
use URI;
use Try::Tiny;

use constant GENERIC_FS_ACTION_WITH_NO_PATH => '';

our $VERSION = "0.8";

our %OPT_TABLE = ();

sub new {
    my ($this, %opts) = @_;
    my $self = +{
        host => $opts{host} || 'localhost',
        port => $opts{port} || 50070,
        standby_host => $opts{standby_host},
        standby_port => ($opts{standby_port} || $opts{port} || 50070),
        httpfs_mode => $opts{httpfs_mode} || 0,
        username => $opts{username},
        doas => $opts{doas},
        useragent => $opts{useragent} || 'Furl Net::Hadoop::WebHDFS (perl)',
        timeout => $opts{timeout} || 10,
        suppress_errors => $opts{suppress_errors} || 0,
        last_error => undef,
        under_failover => 0,
    };
    $self->{furl} = Furl::HTTP->new(agent => $self->{useragent}, timeout => $self->{timeout}, max_redirects => 0);
    return bless $self, $this;
}

# curl -i -X PUT "http://:/webhdfs/v1/?op=CREATE
#                 [&overwrite=][&blocksize=][&replication=]
#                 [&permission=][&buffersize=]"
sub create {
    my ($self, $path, $body, %options) = @_;
    if ($self->{httpfs_mode}) {
        %options = (%options, data => 'true');
    }
    my $err = $self->check_options('CREATE', %options);
    croak $err if $err;

    my $res = $self->operate_requests('PUT', $path, 'CREATE', \%options, $body);
    $res->{code} == 201;
}
$OPT_TABLE{CREATE} = ['overwrite', 'blocksize', 'replication', 'permission', 'buffersize', 'data'];

# curl -i -X POST "http://:/webhdfs/v1/?op=APPEND
#                       [&buffersize=]"
sub append {
    my ($self, $path, $body, %options) = @_;
    if ($self->{httpfs_mode}) {
        %options = (%options, data => 'true');
    }
    my $err = $self->check_options('APPEND', %options);
    croak $err if $err;

    my $res = $self->operate_requests('POST', $path, 'APPEND', \%options, $body);
    $res->{code} == 200;
}
$OPT_TABLE{APPEND} = ['buffersize', 'data'];

# curl -i -L "http://:/webhdfs/v1/?op=OPEN
#                [&offset=][&length=][&buffersize=]"
sub read {
    my ($self, $path, %options) = @_;
    my $err = $self->check_options('OPEN', %options);
    croak $err if $err;

    my $res = $self->operate_requests('GET', $path, 'OPEN', \%options);
    $res->{body};
}
$OPT_TABLE{OPEN} = ['offset', 'length', 'buffersize'];
sub open { (shift)->read(@_); }

# curl -i -X PUT "http://:/?op=MKDIRS
#                   [&permission=]"
sub mkdir {
    my ($self, $path, %options) = @_;
    my $err = $self->check_options('MKDIRS', %options);
    croak $err if $err;

    my $res = $self->operate_requests('PUT', $path, 'MKDIRS', \%options);
    $self->check_success_json($res, 'boolean');
}
$OPT_TABLE{MKDIRS} = ['permission'];
sub mkdirs { (shift)->mkdir(@_); }

# curl -i -X PUT "http://:/webhdfs/v1/?op=RENAME
#                   &destination="
sub rename {
    my ($self, $path, $dest, %options) = @_;
    my $err = $self->check_options('RENAME', %options);
    croak $err if $err;

    unless ($dest =~ m!^/!) {
        $dest = '/' . $dest;
    }
    my $res = $self->operate_requests('PUT', $path, 'RENAME', {%options, destination => $dest});
    $self->check_success_json($res, 'boolean');
}

# curl -i -X DELETE "http://:/webhdfs/v1/?op=DELETE
#                          [&recursive=]"
sub delete {
    my ($self, $path, %options) = @_;
    my $err = $self->check_options('DELETE', %options);
    croak $err if $err;

    my $res = $self->operate_requests('DELETE', $path, 'DELETE', \%options);
    $self->check_success_json($res, 'boolean');
}
$OPT_TABLE{DELETE} = ['recursive'];

# curl -i  "http://:/webhdfs/v1/?op=GETFILESTATUS"
sub stat {
    my ($self, $path, %options) = @_;
    my $err = $self->check_options('GETFILESTATUS', %options);
    croak $err if $err;

    my $res = $self->operate_requests('GET', $path, 'GETFILESTATUS', \%options);
    $self->check_success_json($res, 'FileStatus');
}
sub getfilestatus { (shift)->stat(@_); }

# curl -i  "http://:/webhdfs/v1/?op=LISTSTATUS"
sub list {
    my ($self, $path, %options) = @_;
    my $err = $self->check_options('LISTSTATUS', %options);
    croak $err if $err;

    my $res = $self->operate_requests('GET', $path, 'LISTSTATUS', \%options);
    $self->check_success_json($res, 'FileStatuses')->{FileStatus};
}
sub liststatus { (shift)->list(@_); }

# curl -i "http://:/webhdfs/v1/?op=GETCONTENTSUMMARY"
sub content_summary {
    my ($self, $path, %options) = @_;
    my $err = $self->check_options('GETCONTENTSUMMARY', %options);
    croak $err if $err;

    my $res = $self->operate_requests('GET', $path, 'GETCONTENTSUMMARY', \%options);
    $self->check_success_json($res, 'ContentSummary');
}
sub getcontentsummary { (shift)->content_summary(@_); }

# curl -i "http://:/webhdfs/v1/?op=GETFILECHECKSUM"
sub checksum {
    my ($self, $path, %options) = @_;
    my $err = $self->check_options('GETFILECHECKSUM', %options);
    croak $err if $err;

    my $res = $self->operate_requests('GET', $path, 'GETFILECHECKSUM', \%options);
    $self->check_success_json($res, 'FileChecksum');
}
sub getfilechecksum { (shift)->checksum(@_); }

# curl -i "http://:/webhdfs/v1/?op=GETHOMEDIRECTORY"
sub homedir {
    my ($self, %options) = @_;
    my $err = $self->check_options('GETHOMEDIRECTORY', %options);
    croak $err if $err;

    my $res = $self->operate_requests('GET', '/', 'GETHOMEDIRECTORY', \%options);
    $self->check_success_json($res, 'Path');
}
sub gethomedirectory { (shift)->homedir(@_); }

# curl -i -X PUT "http://:/webhdfs/v1/?op=SETPERMISSION
#                 [&permission=]"
sub chmod {
    my ($self, $path, $mode, %options) = @_;
    my $err = $self->check_options('SETPERMISSION', %options);
    croak $err if $err;

    my $res = $self->operate_requests('PUT', $path, 'SETPERMISSION', {%options, permission => $mode});
    $res->{code} == 200;
}
sub setpermission { (shift)->chmod(@_); }

# curl -i -X PUT "http://:/webhdfs/v1/?op=SETOWNER
#                          [&owner=][&group=]"
sub chown {
    my ($self, $path, %options) = @_;
    my $err = $self->check_options('SETOWNER', %options);
    croak $err if $err;

    unless (defined($options{owner}) or defined($options{group})) {
        croak "'chown' needs at least one of owner or group";
    }

    my $res = $self->operate_requests('PUT', $path, 'SETOWNER', \%options);
    $res->{code} == 200;
}
$OPT_TABLE{SETOWNER} = ['owner', 'group'];
sub setowner { (shift)->chown(@_); }

# curl -i -X PUT "http://:/webhdfs/v1/?op=SETREPLICATION
#                           [&replication=]"
sub replication {
    my ($self, $path, $replnum, %options) = @_;
    my $err = $self->check_options('SETREPLICATION', %options);
    croak $err if $err;

    my $res = $self->operate_requests('PUT', $path, 'SETREPLICATION', {%options, replication => $replnum});
    $self->check_success_json($res, 'boolean');
}
sub setreplication { (shift)->replication(@_); }

# curl -i -X PUT "http://:/webhdfs/v1/?op=SETTIMES
#                           [&modificationtime=][&accesstime=]"
# modificationtime: radix-10 long integer
# accesstime: radix-10 long integer
$OPT_TABLE{SETTIMES} = [ qw( modificationtime accesstime ) ];
sub touch {
    my ($self, $path, %options) = @_;
    my $err = $self->check_options('SETTIMES', %options);
    croak $err if $err;

    unless (defined($options{modificationtime}) or defined($options{accesstime})) {
        croak "'touch' needs at least one of modificationtime or accesstime";
    }

    my $res = $self->operate_requests('PUT', $path, 'SETTIMES', \%options);
    $res->{code} == 200;
}

#---------------------------- EXTENDED ATTRIBUTES START -----------------------#

sub xattr {
    my($self, $path, $action, @args) = @_;
    croak "No action defined for xattr" if ! $action;
    my $target  = sprintf '_%s_xattr', $action;
    my $target2 = sprintf '_%s_xattrs', $action;
    my $method  = $self->can( $target )
                    || $self->can( $target2 )
                    || croak "invalid action `$action`";
    $self->$method( $path, @args );
}

# curl -i "http://:/webhdfs/v1/?op=GETXATTRS
#                           &xattr.name=&encoding="
#
# curl -i "http://:/webhdfs/v1/?op=GETXATTRS
#                           &xattr.name=&xattr.name=&encoding="
$OPT_TABLE{GETXATTRS} = [qw( names encoding flatten )];
sub _get_xattrs {
    my($self, $path, %options) = @_;
    my $err = $self->check_options('GETXATTRS', %options);
    croak $err if $err;

    my $flatten = delete $options{flatten};

    # limit to a subset? will return all of the attributes otherwise
    if ( my $name = delete $options{names} ) {
        croak "getxattrs: name needs to be an arrayref" if ref $name ne 'ARRAY';
        $options{'xattr.name'} = $name;
    }

    my $res = $self->operate_requests('GET', $path, 'GETXATTRS', \%options);
    if ( my $rv = $self->check_success_json($res, 'XAttrs') ) {
        croak "Unexpected return value from listxattrs: $rv"
            if ref $rv ne 'ARRAY';
        return $rv if ! $flatten;
        map { @{ $_ }{qw/ name value /} } @{ $rv };
    }
}

# curl -i "http://:/webhdfs/v1/?op=LISTXATTRS"
sub _list_xattrs  {
    my($self, $path) = @_;

    my $res = $self->operate_requests('GET', $path, 'LISTXATTRS');
    if ( my $rv = $self->check_success_json($res, 'XAttrNames') ) {
        my $attr = JSON::XS::decode_json $rv;
        croak "Unexpected return value from listxattrs: $attr"
            if ref $attr ne 'ARRAY';
        return @{ $attr };
    }
}

# curl -i -X PUT "http://:/webhdfs/v1/?op=SETXATTR
#                           &xattr.name=&xattr.value=&flag="
# https://blog.cloudera.com/blog/2014/06/why-extended-attributes-are-coming-to-hdfs/
# flag: [CREATE,REPLACE]
$OPT_TABLE{SETXATTR} = [qw( name value flag )];
sub _set_xattr {
    my($self, $path, %options) = @_;
    my $err = $self->check_options('SETXATTR', %options);
    croak $err if $err;

    croak "value of xattr not set" if ! exists $options{value};

    $options{ 'xattr.name' }  = delete $options{name}  || croak "name of xattr not set";
    $options{ 'xattr.value' } = delete $options{value};

    croak 'flag was not specified.' if ! $options{flag};

    my $res = $self->operate_requests( PUT => $path, 'SETXATTR', \%options);
    $res->{code} == 200;
}

sub _create_xattr {
    my($self, $path, $name, $value) = @_;
    $self->_set_xattr(
        $path,
        name  => $name,
        value => $value,
        flag  => 'CREATE',
    );
}

sub _replace_xattr {
    my($self, $path, $name, $value) = @_;
    $self->_set_xattr(
        $path,
        name  => $name,
        value => $value,
        flag  => 'REPLACE',
    );
}

# curl -i -X PUT "http://:/webhdfs/v1/?op=REMOVEXATTR
#                           &xattr.name="
sub _remove_xattr {
    my($self, $path, $name) = @_;

    my %options;
    $options{'xattr.name'} =  $name || croak "xattr name was not specified";

    my $res = $self->operate_requests( PUT => $path, 'REMOVEXATTR', \%options);
    $res->{code} == 200;
}

#---------------------------- EXTENDED ATTRIBUTES END -------------------------#

# curl -i "http://:/webhdfs/v1/?op=CHECKACCESS
#                           &fsaction=
# this seems to be broken in some versions. You may get a "No enum constant ..."
# error if this is the case.
# Also see https://issues.apache.org/jira/browse/HDFS-9695
#
sub checkaccess  {
    my($self, $path, $fsaction, %options) = @_;
    croak "checkaccess: fsaction parameter was not specified" if ! $fsaction;
    my $err = $self->check_options('CHECKACCESS', %options);
    croak $err if $err;

    $options{fsaction} = $fsaction;

    my $res = $self->operate_requests('GET', $path, 'CHECKACCESS', \%options);
    $res->{code} == 200;
}

# curl -i -X POST "http://:/webhdfs/v1/?op=CONCAT
#                           &sources="
sub concat  {
    my($self, $path, @sources) = @_;

    croak "At least one source path needs to be specified" if ! @sources;

    my $paths = join q{,}, @sources;

    my $res = $self->operate_requests(
                    POST => $path,
                    'CONCAT',
                    { sources => $paths },
                );
    $res->{code} == 200;
}

# curl -i -X POST "http://:/webhdfs/v1/?op=TRUNCATE
#                           &newlength="
# Available after Hadoop v2.7
# https://issues.apache.org/jira/browse/HDFS-7655
#
sub truncate {
    my($self, $path, $newlength) = @_;
    $newlength = 0 if ! defined $newlength;

    my $res = $self->operate_requests(
                    POST => $path,
                    'TRUNCATE',
                    { newlength => $newlength },
                );

    if ( my $rv = $self->check_success_json($res, 'boolean') ) {
        $rv eq 'true';
    }
}

# curl -i -X PUT "http://:/webhdfs/v1/?op=CREATESYMLINK
#                           &destination=[&createParent=]"
# currently broken/disabled
# https://issues.apache.org/jira/browse/HADOOP-10019

#$OPT_TABLE{CREATESYMLINK} = [qw( destination createParent )];
#sub createsymlink {
#    # Not available yet
#    # https://issues.apache.org/jira/browse/HADOOP-10019
#    my($self, $path, $destination, $createParent) = @_;
#
#    croak "createsymlink: destination not specified" if ! $destination;
#
#    my %options = (
#        destination  => $destination,
#        ($createParent ? (
#        createParent => $createParent ? 'true' : 'false',
#        ) : ())
#    );
#
#    my $res = $self->operate_requests( PUT => $path, 'CREATESYMLINK', \%options);
#    $res->{code} == 200;
#}

#---------------------------- DELEGATION TOKEN START --------------------------#
# Also see
# http://hadoop.apache.org/docs/r2.6.0/hadoop-hdfs-httpfs/httpfs-default.html

# GETDELEGATIONTOKENS: Obsolete and removed after HDFS-10200, HDFS-3667
#

sub delegation_token {
    my($self, $action, @args) = @_;
    croak "No action defined for delegation_token" if ! $action;
    my $target = sprintf '_%s_delegation_token', $action;
    croak "invalid action $action" if ! $self->can( $target );
    $self->$target( @args );
}

# curl -i "http://:/webhdfs/v1/?op=GETDELEGATIONTOKEN
#                           &renewer=&service=&kind="
# kind: The kind of the delegation token requested
#        (Server sets the default kind for the service)
#       A string that represents token kind e.g "HDFS_DELEGATION_TOKEN" or "WEBHDFS delegation"
# service: The name of the service where the token is supposed to be used, e.g. ip:port of the namenode
#
$OPT_TABLE{GETDELEGATIONTOKEN} = [qw( renewer service kind )];
sub _get_delegation_token  {
    my($self, $path, %options) = @_;
    my $err = $self->check_options('GETDELEGATIONTOKEN', %options);
    croak $err if $err;

    $options{renewer} ||= $self->{username} if $self->{username};

    my $res = $self->operate_requests( GET => $path, 'GETDELEGATIONTOKEN', \%options);

    if ( my $rv = $self->check_success_json($res, 'Token') ) {
        $rv->{urlString};
    }
}

# curl -i -X PUT "http://:/webhdfs/v1/?op=RENEWDELEGATIONTOKEN
#                           &token="
sub _renew_delegation_token {
    my($self, $token) = @_;

    croak "No token was specified" if ! $token;

    my $res = $self->operate_requests(
                    PUT => GENERIC_FS_ACTION_WITH_NO_PATH,
                    'RENEWDELEGATIONTOKEN',
                    { token => $token },
                );
    if ( my $rv = $self->check_success_json($res, 'long') ) {
        $rv; # new expiration time in miliseconds
    }
}

# curl -i -X PUT "http://:/webhdfs/v1/?op=CANCELDELEGATIONTOKEN
#                           &token="
sub _cancel_delegation_token {
    my($self, $token) = @_;

    croak "No token was specified" if ! $token;

    my $res = $self->operate_requests(
                    PUT => GENERIC_FS_ACTION_WITH_NO_PATH,
                    'CANCELDELEGATIONTOKEN',
                    { token => $token },
                );
    $res->{code} == 200;
}

#---------------------------- DELEGATION TOKEN END ----------------------------#

#---------------------------- SNAPSHOT START ----------------------------------#

# Needs testing, seems to be buggy and can be destructive in earlier versions
# i.e.: https://issues.apache.org/jira/browse/HDFS-9406
#
# Snaphotting is not enabled by default and this needs to be executed as a super user:
# hdfs dfsadmin -allowSnapshot $path
#
sub snapshot {
    my($self, $path, $action, @args) = @_;
    croak "No action defined for delegation_token" if ! $action;
    my $target = sprintf '_%s_snapshot', $action;
    croak sprintf "%s: invalid action $action", (caller 0)[3] if ! $self->can( $target );
    $self->$target( $path => @args );
}

# curl -i -X PUT "http://:/webhdfs/v1/?op=CREATESNAPSHOT
#                           [&snapshotname=]"
sub _create_snapshot {
    my($self, $path, $snapshotname) = @_;

    my %options;
    $options{snapshotname} = $snapshotname if $snapshotname;
    my $res = $self->operate_requests('PUT', $path, 'CREATESNAPSHOT', \%options);
    if ( my $rv = $self->check_success_json($res, 'Path') ) {
        $rv;
    }
}

# curl -i -X PUT "http://:/webhdfs/v1/?op=RENAMESNAPSHOT
#                           &oldsnapshotname=&snapshotname="
sub _rename_snapshot {
    my($self, $path, $oldsnapshotname, $snapshotname) = @_;

    my %options = (
        oldsnapshotname => $oldsnapshotname,
        snapshotname    => $snapshotname,
    );

    my $res = $self->operate_requests('PUT', $path, 'RENAMESNAPSHOT', \%options);
    $res->{code} == 200;
}

# curl -i -X DELETE "http://:/webhdfs/v1/?op=DELETESNAPSHOT
#                           &snapshotname="
sub _delete_snapshot {
    my($self, $path, $snapshotname) = @_;
    croak "No snapshotname specified" if ! $snapshotname;

    my %options = (
        snapshotname => $snapshotname,
    );
    my $res = $self->operate_requests('DELETE', $path, 'DELETESNAPSHOT', \%options);
    $res->{code} == 200;
}

#---------------------------- SNAPSHOT END ------------------------------------#

sub touchz {
    my ($self, $path) = @_;
    return $self->create( $path, '', overwrite => 'true' );
}

sub settimes { (shift)->touch(@_); }

# sub delegation_token {}
# sub renew_delegation_token {}
# sub cancel_delegation_token {}

sub check_options {
    my ($self, $op, %opts) = @_;
    my @ex = ();
    my $opts = $OPT_TABLE{$op} || [];
    foreach my $k (keys %opts) {
        push @ex, $k if scalar(grep {$k eq $_} @$opts) < 1;
    }
    return undef unless @ex;
    'no such option: ' . join(' ', @ex);
}

sub check_success_json {
    my ($self, $res, $attr) = @_;
    $res->{code} == 200 and $res->{content_type} =~ m!^application/json! and
        (not defined($attr) or JSON::XS::decode_json($res->{body})->{$attr});
}

sub api_path {
    my ($self, $path) = @_;
    return '/webhdfs/v1' . $path if $path =~ m!^/!;
    '/webhdfs/v1/' . $path;
}

sub build_path {
    my ($self, $path, $op, %params) = @_;
    my %opts = (('op' => $op),
                ($self->{username} ? ('user.name' => $self->{username}) : ()),
                ($self->{doas} ? ('doas' => $self->{doas}) : ()),
                %params);
    my $u = URI->new('', 'http');
    $u->query_form(%opts);
    $self->api_path($path) . $u->path_query; # path_query() #=> '?foo=1&bar=2'
}

sub connect_to {
    my $self = shift;
    if ($self->{under_failover}) {
        return ($self->{standby_host}, $self->{standby_port});
    }
    return ($self->{host}, $self->{port});
}

our %REDIRECTED_OPERATIONS = (APPEND => 1, CREATE => 1, OPEN => 1, GETFILECHECKSUM => 1);
sub operate_requests {
    my ($self, $method, $path, $op, $params, $payload) = @_;

    my ($host, $port) = $self->connect_to();

    my $headers = []; # or undef ?
    if ($self->{httpfs_mode} or not $REDIRECTED_OPERATIONS{$op}) {
        # empty files are ok
        if ($self->{httpfs_mode} and defined($payload)) {
            $headers = ['Content-Type' => 'application/octet-stream'];
        }
        return $self->request($host, $port, $method, $path, $op, $params, $payload, $headers);
    }

    # pattern for not httpfs and redirected by namenode
    my $res = $self->request($host, $port, $method, $path, $op, $params, undef);
    unless ($res->{code} >= 300 and $res->{code} <= 399 and $res->{location}) {
        my $code = $res->{code};
        my $body = $res->{body};
        croak "NameNode returns non-redirection (or without location header), code:$code, body:$body.";
    }
    my $uri = URI->new($res->{location});
    $headers = ['Content-Type' => 'application/octet-stream'];
    return $self->request($uri->host, $uri->port, $method, $uri->path_query, undef, {}, $payload, $headers);
}

sub request {
    my $self = shift;
    return $self->_request(@_) unless $self->{suppress_errors};

    try {
        $self->_request(@_);
    } catch {
        $self->{last_error} = $_;
        0;
    };
}

# IllegalArgumentException      400 Bad Request
# UnsupportedOperationException 400 Bad Request
# SecurityException             401 Unauthorized
# IOException                   403 Forbidden
# FileNotFoundException         404 Not Found
# RumtimeException              500 Internal Server Error
sub _request {
    my ($self, $host, $port, $method, $path, $op, $params, $payload, $header) = @_;

    my $request_path = $op ? $self->build_path($path, $op, %$params) : $path;
    my ($ver, $code, $msg, $headers, $body) = $self->{furl}->request(
        method => $method,
        host => $host,
        port => $port,
        path_query => $request_path,
        headers => $header,
        ($payload ? (content => $payload) : ()),
    );

    my $res = { code => $code, body => $body };

    for (my $i = 0; $i < scalar(@$headers); $i += 2) {
        my $header = $headers->[$i];
        my $value = $headers->[$i + 1];

        if ($header =~ m!^location$!i) { $res->{location} = $value; }
        elsif ($header =~ m!^content-type$!i) { $res->{content_type} = $value; }
    }

    return $res if $code >= 200 and $code <= 299;
    return $res if $code >= 300 and $code <= 399;

    my $errmsg = $res->{body} || 'Response body is empty...';
    $errmsg =~ s/\n//g;

    if ($code == 400) { croak "ClientError: $errmsg"; }
    elsif ($code == 401) { croak "SecurityError: $errmsg"; }
    elsif ($code == 403) {
        if ($errmsg =~ /org\.apache\.hadoop\.ipc\.StandbyException/) {
            if ($self->{httpfs_mode} || not defined($self->{standby_host})) {
                # failover is disabled
            } elsif ($self->{retrying}) {
                # more failover is prohibited
                $self->{retrying} = 0;
            } else {
                $self->{under_failover} = not $self->{under_failover};
                $self->{retrying} = 1;
                my ($next_host, $next_port) = $self->connect_to();
                my $val = $self->request($next_host, $next_port, $method, $path, $op, $params, $payload, $header);
                $self->{retrying} = 0;
                return $val;
            }
        }
        croak "IOError: $errmsg";
    }
    elsif ($code == 404) { croak "FileNotFoundError: $errmsg"; }
    elsif ($code == 500) { croak "ServerError: $errmsg"; }

    croak "RequestFailedError, code:$code, message:$errmsg";
}

sub exists {
    my $self = shift;
    my $path = shift || croak "No HDFS path was specified";
    my $stat;
    eval {
        $stat = $self->stat( $path );
        1;
    } or do {
        my $eval_error = $@ || 'Zombie error';
        return if $eval_error =~ m<
            \QFileNotFoundError: {"RemoteException":{"message":"File does not exist:\E
        >xms;
        # just re-throw
        croak $eval_error;
    };
    return $stat;
}

sub find {
    my $self      = shift;
    my $file_path = shift || croak "No file path specified";
    my $cb        = shift;
    my $opt       = @_ && ref $_[-1] eq 'HASH' ? pop @_ : {};

    if ( ref $cb ne 'CODE' ) {
        die "Call back needs to be a CODE ref";
    }

    my $suppress = $self->{suppress_errors};
    # can be used to quickly skip the java junk like, file names starting with
    # underscores, etc.
    my $re_ignore     = $opt->{re_ignore} ? qr/$opt->{re_ignore}/ : undef;

    #
    # No such thing like symlinks (yet) in HDFS, in case you're wondering:
    # https://issues.apache.org/jira/browse/HADOOP-10019
    # although check that link yourself
    #
    my $looper;
    $looper = sub {
        my $thing = shift;
        if ( ! $self->exists( $thing ) ) {
            # should happen at the start, so this will short-circuit the recursion
            warn "The HDFS directory specified ($thing) does not exist! Please guard your HDFS paths with exists()";
            return;
        }
        my $list = $self->list( $thing );
        foreach my $e ( @{ $list } ) {
            my $path = $e->{pathSuffix};
            my $type = $e->{type};

            next if $re_ignore && $path && $path =~ $re_ignore;

            if ( $type eq 'DIRECTORY' ) {
                $cb->( $thing, $e );
                eval {
                    $looper->( File::Spec->catdir( $thing, $path ) );
                    1;
                } or do {
                    my $eval_error = $@ || 'Zombie error';
                    if ( $suppress ) {
                        warn "[ERROR DOWNGRADED] Failed to check $thing/$path: $eval_error";
                        next;
                    }
                    croak $eval_error;
                }
            }
            elsif ( $type eq 'FILE' ) {
                $cb->( $thing, $e );
            }
            else {
                my $msg = "I don't know what to do with type=$type!";
                if ( $suppress ) {
                    warn "[ERROR DOWNGRADED] $msg";
                    next;
                }
                croak $msg;
            }
        }
        return;
    };

    $looper->( $file_path );

    return;
}

1;

__END__