twalthr commented on a change in pull request #18176: URL: https://github.com/apache/flink/pull/18176#discussion_r784660922
########## File path: flink-table/flink-table-planner/src/test/java/org/apache/flink/connector/file/table/FileSystemTableSinkStreamingITCase.java ########## @@ -0,0 +1,91 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.connector.file.table; + +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.api.Schema; +import org.apache.flink.table.api.TableDescriptor; +import org.apache.flink.table.planner.runtime.utils.StreamingTestBase; +import org.apache.flink.types.Row; +import org.apache.flink.util.CloseableIterator; + +import org.junit.Test; + +import java.io.File; +import java.nio.file.Files; +import java.nio.file.Paths; +import java.nio.file.StandardOpenOption; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Test of the filesystem source in streaming mode. */ +public class FileSystemTableSinkStreamingITCase extends StreamingTestBase { + + @Test + public void testMonitorContinuously() throws Exception { + // Create temp dir + File testPath = TEMPORARY_FOLDER.newFolder(); + + // Write first csv file out + Files.write( Review comment: I still think that we should clean up the created file once the test finishes or fails. ########## File path: flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemConnectorOptions.java ########## @@ -46,6 +46,19 @@ "The default partition name in case the dynamic partition" + " column value is null/empty string."); + public static final ConfigOption<Duration> SOURCE_MONITOR_INTERVAL = + key("source.monitor-interval") + .durationType() + .noDefaultValue() + .withDescription( + "The interval in which the source checks for new files. " + + "Each file is uniquely identified by its path, and will be processed once, as soon as it's discovered. " + + "The set of files already processed is kept in state during the whole lifecycle of the source, " + + "so it's persisted in checkpoints and savepoints together with the source state. " + + "Shorter intervals mean that files are discovered more quickly, " + + "but also imply more frequent listing or directory traversal of the file system / object store. " + + "If this config option is empty, the provided path will be scanned once, hence the source will be bounded."); Review comment: an empty option would be invalid, how about `If this config option is not set`? ########## File path: flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableSource.java ########## @@ -272,7 +272,14 @@ public ScanRuntimeProvider getScanRuntimeProvider(ScanContext scanContext) { } private SourceProvider createSourceProvider(BulkFormat<RowData, FileSourceSplit> bulkFormat) { - return SourceProvider.of(FileSource.forBulkFileFormat(bulkFormat, paths()).build()); + final FileSource.FileSourceBuilder<RowData> fileSourceBuilder = + FileSource.forBulkFileFormat(bulkFormat, paths()); + + tableOptions + .getOptional(FileSystemConnectorOptions.SOURCE_MONITOR_INTERVAL) + .ifPresent(fileSourceBuilder::monitorContinuously); Review comment: btw what happens if the duration is set but 0? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
