@Override
    protected void execute(HadoopSettings hadoopSettings) throws MojoExecutionException, MojoFailureException {
        if (!copyFromLocalSource.exists()) {
            throw new MojoExecutionException("copyFromLocalSource " + copyFromLocalSource + " doesn't exist");
        }
        String canonicalLocalSource;
        try {
            canonicalLocalSource = copyFromLocalSource.getCanonicalPath();
        } catch (IOException e) {
            throw new MojoExecutionException("Unable to resolve copyFromLocalSource " + copyFromLocalSource, e);
        }
        if (copyFromLocalTarget == null || copyFromLocalTarget.trim().isEmpty()) {
            copyFromLocalTarget = canonicalLocalSource;
        }
        if (autoCleanCopyFromLocalTarget) {
            String dfsTargetURI = hadoopSettings.getDFSURI() + copyFromLocalTarget;
            if (getLog().isInfoEnabled()) {
                getLog().info("Cleaning dfs:" + copyFromLocalTarget);
            }
            try {
                executeCommand(hadoopSettings, "bin/hadoop fs -rmr " + dfsTargetURI);
            } catch (IOException e) {
                getLog().warn("Unable to clean dfs:" + dfsTargetURI);
            }
        }
        try {
            if (getLog().isInfoEnabled()) {
                getLog().info("Copying data from " + canonicalLocalSource + " to dfs:" + copyFromLocalTarget);
            }
            executeCommand(hadoopSettings, "bin/hadoop fs -copyFromLocal " + canonicalLocalSource + " " + copyFromLocalTarget);
        } catch (IOException e) {
            throw new MojoFailureException("Unable to perform Hadoop copyFromLocal: " + e.getMessage());
        }
    }