Hi ,Is anyone using the flink filesystem (SQL) in idea? when I use the

source.monitor-interval

in IDEA I get exception

Unsupported options: source.monitor-interval

My Environment : JDK 1.8 、IDEA 2020、Flink 1.14 Here is my code:

// create stream
StreamExecutionEnvironment env = initStreamExecutionEnvironment();
// create table env
StreamTableEnvironment tableEnvironment = initStreamTableEnvironment(env);
// create file source
tableEnvironment.executeSql(
        "create table\n" +
                " MyUserTable(\n" +
                "  column_name1 STRING,\n" +
                "  column_name2 STRING,\n" +
                "  part_name1 STRING,\n" +
                "  part_name2 STRING\n" +
                " ) WITH ('connector' = 'filesystem','path' = 'file:///F:/other/temp-data','format' = 'csv', 'source.monitor-interval' = '10')");
// create es sink
tableEnvironment.executeSql(
        "CREATE TABLE myEsTable (\n" +
                "  id string," +
                "  column_name1 STRING,\n" +
                "  column_name2 STRING,\n" +
                "  part_name1 STRING,\n" +
                "  part_name2 STRING,\n" +
                "  PRIMARY KEY (id) NOT ENFORCED\n" +
                ") WITH (\n" +
                "  'connector' = 'elasticsearch-7',\n" +
                "  'hosts' = '<http://localhost:9200>',\n" +
                "  'index' = 'test'\n" +
                ")"
);
// insert to es from fileTable
tableEnvironment.executeSql(
        "insert into myEsTable " +
                " select " +
                " uuid()," +
                " column_name1," +
                " column_name2," +
                " part_name1," +
                " part_name2 " +
                " from MyUserTable"
);

In pom.xml:

<properties>
        <java.version>1.8</java.version>
        <flink.version>1.14.4</flink.version>
 </properties>

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table-api-java-bridge_2.12</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table-planner_2.12</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-streaming-scala_2.12</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table-common</artifactId>
    <version>1.14.4</version>
    <scope>provided</scope>
</dependency>
<!--        elasticsearch7_2 -->
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-elasticsearch7_2.11</artifactId>
    <version>${flink.version}</version>
</dependency>

Exception:

Table options are:

'connector'='filesystem'

'format'='csv'

'path'='file:///F:/other/temp-data'

'source.monitor-interval'='10'

at org.apache.flink.table.factories.FactoryUtil.createTableSource(FactoryUtil.java:150)

at org.apache.flink.table.planner.plan.schema.CatalogSourceTable.createDynamicTableSource(CatalogSourceTable.java:116)

at org.apache.flink.table.planner.plan.schema.CatalogSourceTable.toRel(CatalogSourceTable.java:82)

at org.apache.calcite.sql2rel.SqlToRelConverter.toRel(SqlToRelConverter.java:3585)

at org.apache.calcite.sql2rel.SqlToRelConverter.convertIdentifier(SqlToRelConverter.java:2507)

at org.apache.calcite.sql2rel.SqlToRelConverter.convertFrom(SqlToRelConverter.java:2144)

at org.apache.calcite.sql2rel.SqlToRelConverter.convertFrom(SqlToRelConverter.java:2093)

at org.apache.calcite.sql2rel.SqlToRelConverter.convertFrom(SqlToRelConverter.java:2050)

at org.apache.calcite.sql2rel.SqlToRelConverter.convertSelectImpl(SqlToRelConverter.java:663)

at org.apache.calcite.sql2rel.SqlToRelConverter.convertSelect(SqlToRelConverter.java:644)

at org.apache.calcite.sql2rel.SqlToRelConverter.convertQueryRecursive(SqlToRelConverter.java:3438)

at org.apache.calcite.sql2rel.SqlToRelConverter.convertQuery(SqlToRelConverter.java:570)

at <http://org.apache.flink.table.planner.calcite.FlinkPlannerImpl.org|org.apache.flink.table.planner.calcite.FlinkPlannerImpl.org>$apache$flink$table$planner$calcite$FlinkPlannerImpl$$rel(FlinkPlannerImpl.scala:177)

at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.rel(FlinkPlannerImpl.scala:169)

at org.apache.flink.table.planner.operations.SqlToOperationConverter.toQueryOperation(SqlToOperationConverter.java:1057)

at org.apache.flink.table.planner.operations.SqlToOperationConverter.convertSqlQuery(SqlToOperationConverter.java:1026)

at org.apache.flink.table.planner.operations.SqlToOperationConverter.convert(SqlToOperationConverter.java:301)

at org.apache.flink.table.planner.operations.SqlToOperationConverter.convertSqlInsert(SqlToOperationConverter.java:639)

at org.apache.flink.table.planner.operations.SqlToOperationConverter.convert(SqlToOperationConverter.java:290)

at org.apache.flink.table.planner.delegation.ParserImpl.parse(ParserImpl.java:101)

at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeSql(TableEnvironmentImpl.java:736)

at com.drink.tea.flink.FlinkTest.run(FlinkTest.java:51)

at com.drink.tea.flink.FlinkTest.main(FlinkTest.java:23)

Caused by: org.apache.flink.table.api.ValidationException: Unsupported options found for 'filesystem'.

Unsupported options:

source.monitor-interval

Supported options:

auto-compaction

compaction.file-size

connector

format

partition.default-name

partition.time-extractor.class

partition.time-extractor.kind

partition.time-extractor.timestamp-pattern

path

property-version

sink.parallelism

sink.partition-commit.delay

sink.partition-commit.policy.class

sink.partition-commit.policy.kind

sink.partition-commit.success-file.name

sink.partition-commit.trigger

sink.partition-commit.watermark-time-zone

sink.rolling-policy.check-interval

sink.rolling-policy.file-size

sink.rolling-policy.rollover-interval

sink.shuffle-by-partition.enable

at org.apache.flink.table.factories.FactoryUtil.validateUnconsumedKeys(FactoryUtil.java:484)

at org.apache.flink.table.factories.FactoryUtil$FactoryHelper.validate(FactoryUtil.java:725)

at org.apache.flink.table.factories.FactoryUtil$TableFactoryHelper.validate(FactoryUtil.java:787)

at org.apache.flink.table.factories.FactoryUtil$FactoryHelper.validateExcept(FactoryUtil.java:749)

at org.apache.flink.table.factories.FactoryUtil$TableFactoryHelper.validateExcept(FactoryUtil.java:787)

at org.apache.flink.table.filesystem.FileSystemTableFactory.validate(FileSystemTableFactory.java:124)

at org.apache.flink.table.filesystem.FileSystemTableFactory.createDynamicTableSource(FileSystemTableFactory.java:69)

at org.apache.flink.table.factories.FactoryUtil.createTableSource(FactoryUtil.java:147)

... 22 more

Hi, this config is only available from Flink 1.15

Thank you for your quick response. It really helped me resolve the issue!🙂

np!