private void runInternal(
            final ConfigSource originalConfigSource,
            final Path configDiffPath,
            final Path outputPath,  // deprecated
            final Path resumeStatePath) throws IOException {
        try {
            checkFileWritable(outputPath);
        } catch (IOException ex) {
            throw new RuntimeException("Not writable: " + outputPath.toString());
        }
        try {
            checkFileWritable(configDiffPath);
        } catch (IOException ex) {
            throw new RuntimeException("Not writable: " + configDiffPath.toString());
        }
        try {
            checkFileWritable(resumeStatePath);
        } catch (IOException ex) {
            throw new RuntimeException("Not writable: " + resumeStatePath.toString());
        }

        final ConfigSource configSource;
        if (configDiffPath != null && Files.size(configDiffPath) > 0L) {
            configSource = originalConfigSource.merge(
                    readConfig(configDiffPath, Collections.<String, Object>emptyMap(), null));
        } else {
            configSource = originalConfigSource;
        }

        final ConfigSource resumeConfig;
        if (resumeStatePath != null) {
            ConfigSource resumeConfigTemp = null;
            try {
                resumeConfigTemp = readYamlConfigFile(resumeStatePath);
            } catch (Throwable ex) {
                // TODO log?
                resumeConfigTemp = null;
            }
            if (resumeConfigTemp == null || resumeConfigTemp.isEmpty()) {
                resumeConfig = null;
            } else {
                resumeConfig = resumeConfigTemp;
            }
        } else {
            resumeConfig = null;
        }

        final EmbulkEmbed.ResumableResult resumableResult;
        final ExecutionResult executionResultTemp;
        if (resumeConfig != null) {
            resumableResult = this.embed.resumeState(configSource, resumeConfig).resume();
            executionResultTemp = null;
        } else if (resumeStatePath != null) {
            resumableResult = this.embed.runResumable(configSource);
            executionResultTemp = null;
        } else {
            resumableResult = null;
            executionResultTemp = this.embed.run(configSource);
        }

        final ExecutionResult executionResult;
        if (executionResultTemp == null) {
            if (!resumableResult.isSuccessful()) {
                if (resumableResult.getTransactionStage().isBefore(TransactionStage.RUN)) {
                    // retry without resume state file if no tasks started yet
                    // delete resume file
                    if (resumeStatePath != null) {
                        try {
                            Files.deleteIfExists(resumeStatePath);
                        } catch (Throwable ex) {
                            System.err.println("Failed to delete: " + resumeStatePath.toString());
                        }
                    }
                } else {
                    rootLogger.info("Writing resume state to '" + resumeStatePath.toString() + "'");
                    try {
                        writeResumeState(resumeStatePath, resumableResult.getResumeState());
                    } catch (IOException ex) {
                        throw new RuntimeException(ex);
                    }
                    rootLogger.info("Resume state is written. Run the transaction again with -r option to resume or use \"cleanup\" subcommand to delete intermediate data.");
                }
                throw new RuntimeException(resumableResult.getCause());
            }
            executionResult = resumableResult.getSuccessfulResult();
        } else {
            executionResult = executionResultTemp;
        }

        // delete resume file
        if (resumeStatePath != null) {
            try {
                Files.deleteIfExists(resumeStatePath);
            } catch (Throwable ex) {
                System.err.println("Failed to delete: " + resumeStatePath.toString());
            }
        }

        final ConfigDiff configDiff = executionResult.getConfigDiff();
        rootLogger.info("Committed.");
        rootLogger.info("Next config diff: " + configDiff.toString());

        writeConfig(configDiffPath, configDiff);
        writeConfig(outputPath, configSource.merge(configDiff));  // deprecated
    }
Runs the run subcommand.

It receives a ConfigSource and a String as parameters to be called from Ruby (embulk/runner.rb).
embulk-core/src/main/java/org/embulk/EmbulkRunner.java