This is an automated email from the ASF dual-hosted git repository. gaoyunhaii pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit 9ef8873c0739688a37fcdc365dcd83f321919704 Author: Gen Luo <[email protected]> AuthorDate: Fri Mar 11 19:59:39 2022 +0800 [FLINK-26610][connectors/filesystem] Do not add state handlers for file compaction if compaction is not explicitly disabled. This closes #19061. --- .../docs/connectors/datastream/filesystem.md | 4 +- .../docs/connectors/datastream/filesystem.md | 4 +- .../apache/flink/connector/file/sink/FileSink.java | 41 +++++++++++- .../file/sink/FileSinkCompactionSwitchITCase.java | 73 +++++++++++++++------- 4 files changed, 97 insertions(+), 25 deletions(-) diff --git a/docs/content.zh/docs/connectors/datastream/filesystem.md b/docs/content.zh/docs/connectors/datastream/filesystem.md index 2f5052c..efd6d4c 100644 --- a/docs/content.zh/docs/connectors/datastream/filesystem.md +++ b/docs/content.zh/docs/connectors/datastream/filesystem.md @@ -1025,7 +1025,9 @@ val fileSink: FileSink[Integer] = 这种类型的 `CompactingFileWriter` 会逐条读出输入文件的记录用户，然后和`FileWriter`一样写入输出文件中。`CompactingFileWriter` 的一个例子是 {{< javadoc file="org/apache/flink/connector/file/sink/compactor/RecordWiseFileCompactor.html" name="RecordWiseFileCompactor">}} ，它从给定的文件中读出记录并写出到 `CompactingFileWriter` 中。用户需要指定如何从原始文件中读出记录。 {{< hint info >}} -**重要** 如果启用了文件合并功能，文件可见的时间会被延长。 +**注意事项1** 一旦启用了文件合并功能，此后若需要再关闭，必须在构建`FileSink`时显式调用`disableCompact`方法。 + +**注意事项2** 如果启用了文件合并功能，文件可见的时间会被延长。 {{< /hint >}} <a name="important-considerations"></a> diff --git a/docs/content/docs/connectors/datastream/filesystem.md b/docs/content/docs/connectors/datastream/filesystem.md index 4fce324..4c6a8c2 100644 --- a/docs/content/docs/connectors/datastream/filesystem.md +++ b/docs/content/docs/connectors/datastream/filesystem.md @@ -1026,7 +1026,9 @@ the give list of `Path` and write the result file. It could be classified into t An example is the {{< javadoc file="org/apache/flink/connector/file/sink/compactor/RecordWiseFileCompactor.html" name="RecordWiseFileCompactor">}} that reads records from the source files and then writes them with the `CompactingFileWriter`. Users need to specify how to read records from the source files. {{< hint info >}} -**Important** Once the compaction is enabled, the written files need to wait for longer time before they get visible. +**Important Note 1** Once the compaction is enabled, you must explicitly call `disableCompact` when building the `FileSink` if you want to disable compaction. + +**Important Note 2** When the compaction is enabled, the written files need to wait for longer time before they get visible. {{< /hint >}} ### Important Considerations diff --git a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/FileSink.java b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/FileSink.java index 8d745d9..8fdf987 100644 --- a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/FileSink.java +++ b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/FileSink.java @@ -117,6 +117,11 @@ import static org.apache.flink.util.Preconditions.checkState; * finished} state while any {@code in-progress} files are rolled back, so that they do not contain * data that arrived after the checkpoint from which we restore. * + * <p>FileSink also support compacting small files to accelerate the access speed of the resulted + * files. Compaction could be enabled via {@code enableCompact}. Once enabled, the compaction could + * only be disabled via calling {@code disableCompact} explicitly, otherwise there might be data + * loss. + * * @param <IN> Type of the elements in the input of the sink that are also the elements to be * written to its output */ @@ -197,9 +202,14 @@ public class FileSink<IN> public DataStream<CommittableMessage<FileSinkCommittable>> addPreCommitTopology( DataStream<CommittableMessage<FileSinkCommittable>> committableStream) { FileCompactStrategy strategy = bucketsBuilder.getCompactStrategy(); + if (strategy == null && !bucketsBuilder.isCompactDisabledExplicitly()) { + // compact is never enabled, we may not add the handlers + return committableStream; + } + if (strategy == null) { - // not enabled, handlers will be added to process the remaining states of the compact - // coordinator and the compactor operators. + // not enabled at present, handlers will be added to process the remaining states of the + // compact coordinator and the compactor operators. SingleOutputStreamOperator< Either<CommittableMessage<FileSinkCommittable>, CompactorRequest>> coordinatorOp = @@ -290,6 +300,9 @@ public class FileSink<IN> throws IOException; @Internal + abstract boolean isCompactDisabledExplicitly(); + + @Internal abstract FileCompactStrategy getCompactStrategy(); @Internal @@ -319,6 +332,8 @@ public class FileSink<IN> private OutputFileConfig outputFileConfig; + private boolean isCompactDisabledExplicitly = false; + private FileCompactStrategy compactStrategy; private FileCompactor fileCompactor; @@ -378,6 +393,11 @@ public class FileSink<IN> return self(); } + public T disableCompact() { + this.isCompactDisabledExplicitly = true; + return self(); + } + /** Creates the actual sink. */ public FileSink<IN> build() { return new FileSink<>(this); @@ -416,6 +436,11 @@ public class FileSink<IN> } @Override + boolean isCompactDisabledExplicitly() { + return isCompactDisabledExplicitly; + } + + @Override FileCompactStrategy getCompactStrategy() { return compactStrategy; } @@ -483,6 +508,8 @@ public class FileSink<IN> private OutputFileConfig outputFileConfig; + private boolean isCompactDisabledExplicitly = false; + private FileCompactStrategy compactStrategy; private FileCompactor fileCompactor; @@ -560,6 +587,11 @@ public class FileSink<IN> return self(); } + public T disableCompact() { + this.isCompactDisabledExplicitly = true; + return self(); + } + /** Creates the actual sink. */ public FileSink<IN> build() { return new FileSink<>(this); @@ -598,6 +630,11 @@ public class FileSink<IN> } @Override + boolean isCompactDisabledExplicitly() { + return isCompactDisabledExplicitly; + } + + @Override FileCompactStrategy getCompactStrategy() { return compactStrategy; } diff --git a/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/sink/FileSinkCompactionSwitchITCase.java b/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/sink/FileSinkCompactionSwitchITCase.java index 6e7f8a4..d4a68ca 100644 --- a/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/sink/FileSinkCompactionSwitchITCase.java +++ b/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/sink/FileSinkCompactionSwitchITCase.java @@ -37,6 +37,7 @@ import org.apache.flink.connector.file.sink.utils.IntegerFileSinkTestDataUtils.M import org.apache.flink.connector.file.sink.utils.PartSizeAndCheckpointRollingPolicy; import org.apache.flink.core.execution.SavepointFormatType; import org.apache.flink.core.fs.Path; +import org.apache.flink.runtime.client.JobExecutionException; import org.apache.flink.runtime.jobgraph.JobGraph; import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings; import org.apache.flink.runtime.minicluster.MiniCluster; @@ -64,8 +65,6 @@ import org.junit.ClassRule; import org.junit.Rule; import org.junit.Test; import org.junit.rules.TemporaryFolder; -import org.junit.runner.RunWith; -import org.junit.runners.Parameterized; import java.io.DataInputStream; import java.io.EOFException; @@ -73,7 +72,6 @@ import java.io.File; import java.io.FileInputStream; import java.util.ArrayList; import java.util.Arrays; -import java.util.Collection; import java.util.Collections; import java.util.Comparator; import java.util.HashMap; @@ -86,9 +84,9 @@ import java.util.concurrent.CountDownLatch; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; /** Tests of switching on or off compaction for the {@link FileSink}. */ -@RunWith(Parameterized.class) public class FileSinkCompactionSwitchITCase extends TestLogger { private static final int PARALLELISM = 4; @@ -119,13 +117,6 @@ public class FileSinkCompactionSwitchITCase extends TestLogger { private String latchId; - @Parameterized.Parameter public boolean isOnToOff; - - @Parameterized.Parameters(name = "isOnToOff = {0}") - public static Collection<Object[]> params() { - return Arrays.asList(new Object[] {false}, new Object[] {true}); - } - @Before public void setup() { this.latchId = UUID.randomUUID().toString(); @@ -139,14 +130,53 @@ public class FileSinkCompactionSwitchITCase extends TestLogger { } @Test - public void testSwitchingCompaction() throws Exception { + public void testSwitchNeverEnabledToEnabled() throws Exception { + String path = TEMPORARY_FOLDER.newFolder().getAbsolutePath(); + FileSink<Integer> originFileSink = createFileSink(path, null, false); + FileSink<Integer> restoredFileSink = + createFileSink(path, createFileCompactStrategy(), false); + testSwitchingCompaction(path, originFileSink, restoredFileSink); + } + + @Test + public void testSwitchDisabledToEnabled() throws Exception { String path = TEMPORARY_FOLDER.newFolder().getAbsolutePath(); + FileSink<Integer> originFileSink = createFileSink(path, null, true); + FileSink<Integer> restoredFileSink = + createFileSink(path, createFileCompactStrategy(), false); + testSwitchingCompaction(path, originFileSink, restoredFileSink); + } + + @Test + public void testSwitchEnabledToDisabled() throws Exception { + String path = TEMPORARY_FOLDER.newFolder().getAbsolutePath(); + FileSink<Integer> originFileSink = createFileSink(path, createFileCompactStrategy(), false); + FileSink<Integer> restoredFileSink = createFileSink(path, null, true); + testSwitchingCompaction(path, originFileSink, restoredFileSink); + } + + @Test + public void testSwitchEnabledToDisabledImproperly() throws Exception { + String path = TEMPORARY_FOLDER.newFolder().getAbsolutePath(); + FileSink<Integer> originFileSink = createFileSink(path, createFileCompactStrategy(), false); + FileSink<Integer> restoredFileSink = createFileSink(path, null, false); + try { + testSwitchingCompaction(path, originFileSink, restoredFileSink); + } catch (JobExecutionException expected) { + return; + } + fail("Job is not failing when compaction is disabled improperly"); + } + + private void testSwitchingCompaction( + String path, FileSink<Integer> originFileSink, FileSink<Integer> restoredFileSink) + throws Exception { String cpPath = "file://" + TEMPORARY_FOLDER.newFolder().getAbsolutePath(); SharedReference<ConcurrentHashMap<Integer, Integer>> sendCountMap = sharedObjects.add(new ConcurrentHashMap<>()); - JobGraph jobGraph = createJobGraph(path, cpPath, isOnToOff, false, sendCountMap); - JobGraph restoringJobGraph = createJobGraph(path, cpPath, !isOnToOff, true, sendCountMap); + JobGraph jobGraph = createJobGraph(cpPath, originFileSink, false, sendCountMap); + JobGraph restoringJobGraph = createJobGraph(cpPath, restoredFileSink, true, sendCountMap); final Configuration config = new Configuration(); config.setString(RestOptions.BIND_PORT, "18081-19000"); @@ -185,9 +215,8 @@ public class FileSinkCompactionSwitchITCase extends TestLogger { } private JobGraph createJobGraph( - String path, String cpPath, - boolean compactionEnabled, + FileSink<Integer> fileSink, boolean isFinite, SharedReference<ConcurrentHashMap<Integer, Integer>> sendCountMap) { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); @@ -204,7 +233,7 @@ public class FileSinkCompactionSwitchITCase extends TestLogger { env.addSource(new CountingTestSource(latchId, NUM_RECORDS, isFinite, sendCountMap)) .setParallelism(NUM_SOURCES) - .sinkTo(createFileSink(path, compactionEnabled)) + .sinkTo(fileSink) .uid("sink") .setParallelism(NUM_SINKS); @@ -212,15 +241,17 @@ public class FileSinkCompactionSwitchITCase extends TestLogger { return streamGraph.getJobGraph(); } - private FileSink<Integer> createFileSink(String path, boolean compactionEnabled) { + private FileSink<Integer> createFileSink( + String path, FileCompactStrategy compactStrategy, boolean disableCompact) { DefaultRowFormatBuilder<Integer> sinkBuilder = FileSink.forRowFormat(new Path(path), new IntEncoder()) .withBucketAssigner(new ModuloBucketAssigner(NUM_BUCKETS)) .withRollingPolicy(new PartSizeAndCheckpointRollingPolicy<>(1024, false)); - if (compactionEnabled) { - sinkBuilder = - sinkBuilder.enableCompact(createFileCompactStrategy(), createFileCompactor()); + if (compactStrategy != null) { + sinkBuilder = sinkBuilder.enableCompact(compactStrategy, createFileCompactor()); + } else if (disableCompact) { + sinkBuilder = sinkBuilder.disableCompact(); } return sinkBuilder.build();
