import org.embulk.config.ConfigException; //导入依赖的package包/类
@Override
public void transaction(ConfigSource config, Schema schema,
        FormatterPlugin.Control control)
{
    PluginTask task = config.loadConfig(PluginTask.class);

    // validate avsc option
    try {
        File avsc = task.getAvsc().getFile();
        new org.apache.avro.Schema.Parser().parse(avsc);
    } catch (IOException e) {
        throw new ConfigException("avsc file is not found");
    }


    // validate column_options
    for (String columnName : task.getColumnOptions().keySet()) {
        schema.lookupColumn(columnName);  // throws SchemaConfigException
    }

    control.run(task.dump());
}
 

import org.embulk.config.ConfigException; //导入依赖的package包/类
@Override
public void transaction(ConfigSource config, ParserPlugin.Control control)
{
    PluginTask task = config.loadConfig(PluginTask.class);

    File avsc = task.getAvsc().getFile();
    org.apache.avro.Schema avroSchema;
    try {
        avroSchema = new org.apache.avro.Schema.Parser().parse(avsc);
    } catch (IOException e) {
        throw new ConfigException("avsc file is not found");
    }

    Schema schema = buildSchema(task.getColumns(), avroSchema);

    control.run(task.dump(), schema);
}
 

import org.embulk.config.ConfigException; //导入依赖的package包/类
private void configure(PluginTask task, Schema inputSchema)
{
    List<ColumnConfig> columns = task.getColumns();

    if (columns.size() < 2) {
        throw new ConfigException("\"columns\" should be specified 2~ columns.");
    }

    // throw if column type is not supported
    for (ColumnConfig columnConfig : columns) {
        String name = columnConfig.getName();
        Type type = inputSchema.lookupColumn(name).getType();

        if (type instanceof JsonType) {
            throw new ConfigException(String.format("casting to json is not available: \"%s\"", name));
        }
        if (type instanceof TimestampType) {
            throw new ConfigException(String.format("casting to timestamp is not available: \"%s\"", name));
        }
    }
}
 

import org.embulk.config.ConfigException; //导入依赖的package包/类
public static void assertJsonPathFormat(String path)
{
    Path compiledPath;
    try {
        compiledPath = PathCompiler.compile(path);
    }
    catch (InvalidPathException e) {
        throw new ConfigException(String.format("jsonpath %s, %s", path, e.getMessage()));
    }
    PathToken pathToken = compiledPath.getRoot();
    while (true) {
        assertSupportedPathToken(pathToken, path);
        if (pathToken.isLeaf()) {
            break;
        }
        pathToken = pathToken.next();
    }
}
 

import org.embulk.config.ConfigException; //导入依赖的package包/类
protected static void assertSupportedPathToken(PathToken pathToken, String path)
{
    if (pathToken instanceof ArrayPathToken) {
        ArrayIndexOperation arrayIndexOperation = ((ArrayPathToken) pathToken).getArrayIndexOperation();
        assertSupportedArrayPathToken(arrayIndexOperation, path);
    }
    else if (pathToken instanceof ScanPathToken) {
        throw new ConfigException(String.format("scan path token is not supported \"%s\"", path));
    }
    else if (pathToken instanceof FunctionPathToken) {
        throw new ConfigException(String.format("function path token is not supported \"%s\"", path));
    }
    else if (pathToken instanceof PredicatePathToken) {
        throw new ConfigException(String.format("predicate path token is not supported \"%s\"", path));
    }
}
 

import org.embulk.config.ConfigException; //导入依赖的package包/类
@SuppressWarnings("unchecked")
@Override
protected RestClientInputPluginDelegate dispatchPerTask(PluginTask task)
{
    Target target = task.getTarget();
    switch (target) {
        case LEAD:
        case ACTIVITY:
            if (!task.getWrappedFromDate().isPresent()) {
                throw new ConfigException("From date is required for target LEAD or ACTIVITY");
            }
            Date date = task.getWrappedFromDate().get();
            task.setFromDate(date);
            break;
    }
    return target.getRestClientInputPluginDelegate();
}
 

import org.embulk.config.ConfigException; //导入依赖的package包/类
@Override
public void validateInputTask(T task)
{
    super.validateInputTask(task);
    if (task.getFromDate() == null) {
        throw new ConfigException("From date is required for Bulk Extract");
    }
    if (task.getFromDate().getTime() >= task.getJobStartTime().getMillis()) {
        throw new ConfigException("From date can't not be in future");
    }
    if (task.getIncremental()
            && task.getIncrementalColumn().isPresent()
            && task.getIncrementalColumn().get().equals("updatedAt")) {
        throw new ConfigException("Column 'updatedAt' cannot be incremental imported");
    }
    //Calculate to date
    DateTime toDate = getToDate(task);
    task.setToDate(Optional.of(toDate.toDate()));
}
 

import org.embulk.config.ConfigException; //导入依赖的package包/类
public CsvTokenizer(String delimiter, char quote, char escape, String newline, boolean trimIfNotQuoted, long maxQuotedSizeLimit, String commentLineMarker, LineDecoder input, String nullStringOrNull)
{
    if (delimiter.length() == 0) {
        throw new ConfigException("Empty delimiter is not allowed");
    }
    else {
        this.delimiterChar = delimiter.charAt(0);
        if (delimiter.length() > 1) {
            delimiterFollowingString = delimiter.substring(1);
        }
        else {
            delimiterFollowingString = null;
        }
    }
    this.quote = quote;
    this.escape = escape;
    this.newline = newline;
    this.trimIfNotQuoted = trimIfNotQuoted;
    this.maxQuotedSizeLimit = maxQuotedSizeLimit;
    this.commentLineMarker = commentLineMarker;
    this.input = input;
    this.nullStringOrNull = nullStringOrNull;
}
 

import org.embulk.config.ConfigException; //导入依赖的package包/类
@Test()
public void validateInputTaskFromDateMoreThanJobStartTime()
{
    Date fromDate = new Date(1507619744000L);
    DateTime jobStartTime = new DateTime(1506842144000L);
    MarketoBaseBulkExtractInputPlugin.PluginTask pluginTask = Mockito.mock(MarketoBaseBulkExtractInputPlugin.PluginTask.class);
    Mockito.when(pluginTask.getFromDate()).thenReturn(fromDate);
    Mockito.when(pluginTask.getJobStartTime()).thenReturn(jobStartTime);

    try {
        baseBulkExtractInputPlugin.validateInputTask(pluginTask);
    }
    catch (ConfigException ex) {
        return;
    }
    fail();
}
 

import org.embulk.config.ConfigException; //导入依赖的package包/类
private StandardFileSystemManager initializeStandardFileSystemManager()
{
    if (!logger.isDebugEnabled()) {
        // TODO: change logging format: org.apache.commons.logging.Log
        System.setProperty("org.apache.commons.logging.Log", "org.apache.commons.logging.impl.NoOpLog");
    }
    StandardFileSystemManager manager = new StandardFileSystemManager();
    manager.setClassLoader(SftpUtils.class.getClassLoader());
    try {
        manager.init();
    }
    catch (FileSystemException e) {
        logger.error(e.getMessage());
        throw new ConfigException(e);
    }

    return manager;
}
 

import org.embulk.config.ConfigException; //导入依赖的package包/类
public static void assertSupportedPathToken(PathToken pathToken, String path)
{
    if (pathToken instanceof ArrayPathToken) {
        ArrayIndexOperation arrayIndexOperation = ((ArrayPathToken) pathToken).getArrayIndexOperation();
        assertSupportedArrayPathToken(arrayIndexOperation, path);
    }
    else if (pathToken instanceof ScanPathToken) {
        throw new ConfigException(String.format("scan path token is not supported \"%s\"", path));
    }
    else if (pathToken instanceof FunctionPathToken) {
        throw new ConfigException(String.format("function path token is not supported \"%s\"", path));
    }
    else if (pathToken instanceof PredicatePathToken) {
        throw new ConfigException(String.format("predicate path token is not supported \"%s\"", path));
    }
}
 

import org.embulk.config.ConfigException; //导入依赖的package包/类
public static Long getTailIndex(String path)
{
    Path compiledPath = PathCompiler.compile(path);
    PathToken tail = ((RootPathToken) compiledPath.getRoot()).getTail();
    if (tail instanceof ArrayPathToken) {
        ArrayIndexOperation arrayIndexOperation = ((ArrayPathToken) tail).getArrayIndexOperation();
        if (arrayIndexOperation == null) {
            throw new ConfigException(String.format("Array Slice Operation is not supported \"%s\"", path));
        }
        if (arrayIndexOperation.isSingleIndexOperation()) {
            return arrayIndexOperation.indexes().get(0).longValue();
        }
        else {
            throw new ConfigException(String.format("Multi Array Indexes is not supported \"%s\"", path));
        }
    }
    else {
        return null;
    }
}
 

import org.embulk.config.ConfigException; //导入依赖的package包/类
private SchemaConfig getSchemaConfig(PluginTask task)
{
    if (task.getOldSchemaConfig().isPresent()) {
        log.warn("Please use 'columns' option instead of 'schema' because the 'schema' option is deprecated. The next version will stop 'schema' option support.");
    }

    if (task.getSchemaConfig().isPresent()) {
        return task.getSchemaConfig().get();
    }
    else if (task.getOldSchemaConfig().isPresent()) {
        return task.getOldSchemaConfig().get();
    }
    else {
        throw new ConfigException("Attribute 'columns' is required but not set");
    }
}
 

import org.embulk.config.ConfigException; //导入依赖的package包/类
private String getLocalMd5hash(String filePath) throws IOException
{
    try {
        MessageDigest md = MessageDigest.getInstance("MD5");
        try (BufferedInputStream input = new BufferedInputStream(new FileInputStream(new File(filePath)))) {
            byte[] buffer = new byte[256];
            int len;
            while ((len = input.read(buffer, 0, buffer.length)) >= 0) {
                md.update(buffer, 0, len);
            }
            return new String(Base64.encodeBase64(md.digest()));
        }
    }
    catch (NoSuchAlgorithmException ex) {
        throw new ConfigException("MD5 algorism not found");
    }
}
 

import org.embulk.config.ConfigException; //导入依赖的package包/类
@Test(expected = ConfigException.class)
public void checkDefaultValuesP12keyNull()
{
    ConfigSource config = Exec.newConfigSource()
            .set("in", inputConfig())
            .set("parser", parserConfig(schemaConfig()))
            .set("type", "gcs")
            .set("bucket", GCP_BUCKET)
            .set("path_prefix", "my-prefix")
            .set("file_ext", ".csv")
            .set("auth_method", "private_key")
            .set("service_account_email", GCP_EMAIL)
            .set("p12_keyfile", null)
            .set("formatter", formatterConfig());

    Schema schema = config.getNested("parser").loadConfig(CsvParserPlugin.PluginTask.class).getSchemaConfig().toSchema();

    runner.transaction(config, schema, 0, new Control());
}
 

import org.embulk.config.ConfigException; //导入依赖的package包/类
@Test(expected = ConfigException.class)
public void checkDefaultValuesConflictSetting()
{
    ConfigSource config = Exec.newConfigSource()
            .set("in", inputConfig())
            .set("parser", parserConfig(schemaConfig()))
            .set("type", "gcs")
            .set("bucket", GCP_BUCKET)
            .set("path_prefix", "my-prefix")
            .set("file_ext", ".csv")
            .set("auth_method", "private_key")
            .set("service_account_email", GCP_EMAIL)
            .set("p12_keyfile", GCP_P12_KEYFILE)
            .set("p12_keyfile_path", GCP_P12_KEYFILE)
            .set("formatter", formatterConfig());

    Schema schema = config.getNested("parser").loadConfig(CsvParserPlugin.PluginTask.class).getSchemaConfig().toSchema();

    runner.transaction(config, schema, 0, new Control());
}
 

import org.embulk.config.ConfigException; //导入依赖的package包/类
@Test(expected = ConfigException.class)
public void checkDefaultValuesInvalidPrivateKey()
{
    ConfigSource config = Exec.newConfigSource()
            .set("in", inputConfig())
            .set("parser", parserConfig(schemaConfig()))
            .set("type", "gcs")
            .set("bucket", GCP_BUCKET)
            .set("path_prefix", "my-prefix")
            .set("file_ext", ".csv")
            .set("auth_method", "private_key")
            .set("service_account_email", GCP_EMAIL)
            .set("p12_keyfile", "invalid-key.p12")
            .set("formatter", formatterConfig());

    Schema schema = config.getNested("parser").loadConfig(CsvParserPlugin.PluginTask.class).getSchemaConfig().toSchema();

    runner.transaction(config, schema, 0, new Control());
}
 

import org.embulk.config.ConfigException; //导入依赖的package包/类
@Test(expected = ConfigException.class)
public void checkDefaultValuesJsonKeyfileNull()
{
    ConfigSource config = Exec.newConfigSource()
            .set("in", inputConfig())
            .set("parser", parserConfig(schemaConfig()))
            .set("type", "gcs")
            .set("bucket", GCP_BUCKET)
            .set("path_prefix", "my-prefix")
            .set("file_ext", ".csv")
            .set("auth_method", "json_key")
            .set("service_account_email", GCP_EMAIL)
            .set("json_keyfile", null)
            .set("formatter", formatterConfig());

    Schema schema = config.getNested("parser").loadConfig(CsvParserPlugin.PluginTask.class).getSchemaConfig().toSchema();

    runner.transaction(config, schema, 0, new Control());
}
 

import org.embulk.config.ConfigException; //导入依赖的package包/类
void configure(PluginTask task, Schema inputSchema) throws ConfigException
{
    if (task.getConditions().isPresent()) {
        logger.warn("embulk-filter-row: \"conditions\" is deprecated, use \"where\" instead.");
        for (ConditionConfig conditionConfig : task.getConditions().get()) {
            String columnName = conditionConfig.getColumn();
            inputSchema.lookupColumn(columnName); // throw SchemaConfigException if not found
        }

        String condition = task.getCondition().toLowerCase();
        if (!condition.equals("or") && !condition.equals("and")) {
            throw new ConfigException("condition must be either of \"or\" or \"and\".");
        }
    }
    else if (task.getWhere().isPresent()) {
        String where = task.getWhere().get();
        Parser parser = new Parser(inputSchema);
        // objects must be created in open since plugin instances in transaction and open are different
        parser.parse(where); // throws ConfigException if something wrong
    }
    else {
        throw new ConfigException("Either of `conditions` or `where` must be set.");
    }
}
 

import org.embulk.config.ConfigException; //导入依赖的package包/类
public TimestampOpExp(ParserLiteral left, ParserLiteral right, int operator)
{
    this.left = left.isIdentifier() ? left : new TimestampLiteral(left);
    this.right = right.isIdentifier() ? right : new TimestampLiteral(right);
    this.operator = operator;

    if (! this.left.isTimestamp()) {
        throw new ConfigException(String.format("\"%s\" is not a Timestamp column", ((IdentifierLiteral)this.left).name));
    }
    if (! this.right.isTimestamp()) {
        throw new ConfigException(String.format("\"%s\" is not a Timestamp column", ((IdentifierLiteral)this.right).name));
    }
    if (! isOperatorAllowed(operators, operator)) {
        throw new ConfigException(String.format("\"%s\" is not an allowed operator for TimestampOpExp", Parser.yyname[operator]));
    }
}