@Test
public void validateConfigSource()
        throws IOException
{

    { // specified column is not included in schema
        ConfigSource config = runtime.getExec().newConfigSource()
                .set("column", "_c0").set("unit", "hour").set("timezone", "UTC");
        Schema schema = Schema.builder().add("not_included", Types.TIMESTAMP).build();
        try {
            tp.configure(config, schema, 0);
            fail();
        } catch (Throwable t) {
            assertTrue(t instanceof ConfigException);
        }
    }
    { // only UTC is supported now
        ConfigSource config = runtime.getExec().newConfigSource()
                .set("column", "_c0").set("unit", "hour").set("timezone", "PDT");
        Schema schema = Schema.builder().add("_c0", Types.TIMESTAMP).build();
        try {
            tp.configure(config, schema, 0);
            fail();
        } catch (Throwable t) {
            assertTrue(t instanceof ConfigException);
        }
    }
    { // the unit is only 'hour' or 'day'
        ConfigSource config = runtime.getExec().newConfigSource()
                .set("column", "_c0").set("unit", "invalid").set("timezone", "UTC");
        Schema schema = Schema.builder().add("_c0", Types.TIMESTAMP).build();
        try {
            tp.configure(config, schema, 0);
            fail();
        } catch (Throwable t) {
            assertTrue(t instanceof ConfigException);
        }
    }
    { // the column type is only timestamp or long
        ConfigSource config = runtime.getExec().newConfigSource()
                .set("column", "_c0").set("unit", "hour").set("timezone", "UTC");
        Schema schema = Schema.builder().add("_c0", Types.STRING).build();
        try {
            tp.configure(config, schema, 0);
            fail();
        } catch (Throwable t) {
            assertTrue(t instanceof ConfigException);
        }
    }
    { // if the column type is long, unix_timestamp_unit is required
        ConfigSource config = runtime.getExec().newConfigSource()
                .set("column", "_c0").set("unit", "hour").set("timezone", "UTC").set("unix_timestamp_unit", "invalid");
        Schema schema = Schema.builder().add("_c0", Types.LONG).build();
        try {
            tp.configure(config, schema, 0);
            fail();
        } catch (Throwable t) {
            assertTrue(t instanceof ConfigException);
        }
    }
}