This is an automated email from the ASF dual-hosted git repository.
MartijnVisser pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 3196a998cf0 [FLINK-30166][tests] Refactor tests that use the
deprecated StreamingFileSink to FileSink (#28565)
3196a998cf0 is described below
commit 3196a998cf00a58bb49d4441baf6bbb5cd45d5b5
Author: Martijn Visser <[email protected]>
AuthorDate: Sat Jul 4 09:45:48 2026 +0200
[FLINK-30166][tests] Refactor tests that use the deprecated
StreamingFileSink to FileSink (#28565)
Includes extracting repeated bulk-writer block into OrcBulkWriterTestUtil
Generated-by: Claude Code (Opus 4.8)
---
.../flink/connector/file/sink/FileSinkProgram.java | 50 ++++++-------------
.../flink-stream-sql-test/pom.xml | 6 +++
.../flink/sql/tests/StreamSQLTestProgram.java | 8 ++--
flink-end-to-end-tests/run-nightly-tests.sh | 6 +--
.../test-scripts/test_file_sink.sh | 6 +--
...FileSinkITCase.java => AvroFileSinkITCase.java} | 19 ++++----
flink-formats/flink-compress/pom.xml | 3 +-
.../compress/CompressWriterFactoryTest.java | 56 +++++++---------------
.../formats/compress/CompressionFactoryITCase.java | 10 ++--
.../org/apache/flink/orc/OrcFileSystemITCase.java | 26 ++--------
.../flink/orc/util/OrcBulkWriterTestUtil.java | 17 +++++++
.../flink/orc/writer/OrcBulkRowDataWriterTest.java | 34 ++++---------
.../flink/orc/writer/OrcBulkWriterITCase.java | 8 ++--
.../apache/flink/orc/writer/OrcBulkWriterTest.java | 35 ++++----------
...kITCase.java => AvroParquetFileSinkITCase.java} | 20 ++++----
...ITCase.java => ParquetProtoFileSinkITCase.java} | 12 ++---
flink-formats/flink-sequence-file/pom.xml | 2 +-
...SinkITCase.java => SequenceFileSinkITCase.java} | 11 ++---
18 files changed, 124 insertions(+), 205 deletions(-)
diff --git
a/flink-end-to-end-tests/flink-file-sink-test/src/main/java/org/apache/flink/connector/file/sink/FileSinkProgram.java
b/flink-end-to-end-tests/flink-file-sink-test/src/main/java/org/apache/flink/connector/file/sink/FileSinkProgram.java
index e22e0e1d162..14689dec6fa 100644
---
a/flink-end-to-end-tests/flink-file-sink-test/src/main/java/org/apache/flink/connector/file/sink/FileSinkProgram.java
+++
b/flink-end-to-end-tests/flink-file-sink-test/src/main/java/org/apache/flink/connector/file/sink/FileSinkProgram.java
@@ -34,7 +34,6 @@ import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.filesystem.BucketAssigner;
import
org.apache.flink.streaming.api.functions.sink.filesystem.bucketassigners.SimpleVersionedStringSerializer;
-import
org.apache.flink.streaming.api.functions.sink.filesystem.legacy.StreamingFileSink;
import
org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.OnCheckpointRollingPolicy;
import org.apache.flink.streaming.api.functions.source.legacy.SourceFunction;
import org.apache.flink.util.ParameterTool;
@@ -44,7 +43,7 @@ import java.time.Duration;
import java.util.Collections;
/**
- * Test program for the {@link StreamingFileSink} and {@link FileSink}.
+ * Test program for the {@link FileSink}.
*
* <p>Uses a source that steadily emits a deterministic set of records over 60
seconds, after which
* it idles and waits for job cancellation. Every record has a unique index
that is written to the
@@ -60,7 +59,6 @@ public enum FileSinkProgram {
public static void main(final String[] args) throws Exception {
final ParameterTool params = ParameterTool.fromArgs(args);
final String outputPath = params.getRequired("outputPath");
- final String sinkToTest = params.getRequired("sinkToTest");
final StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
@@ -78,38 +76,20 @@ public enum FileSinkProgram {
// generate data, shuffle, sink
DataStream<Tuple2<Integer, Integer>> source = env.addSource(new
Generator(10, 10, 60));
- if (sinkToTest.equalsIgnoreCase("StreamingFileSink")) {
- final StreamingFileSink<Tuple2<Integer, Integer>> sink =
- StreamingFileSink.forRowFormat(
- new Path(outputPath),
- (Encoder<Tuple2<Integer, Integer>>)
- (element, stream) -> {
- PrintStream out = new
PrintStream(stream);
- out.println(element.f1);
- })
- .withBucketAssigner(new KeyBucketAssigner())
-
.withRollingPolicy(OnCheckpointRollingPolicy.build())
- .build();
-
- source.keyBy(x -> x.f0).addSink(sink);
- } else if (sinkToTest.equalsIgnoreCase("FileSink")) {
- FileSink<Tuple2<Integer, Integer>> sink =
- FileSink.forRowFormat(
- new Path(outputPath),
- (Encoder<Tuple2<Integer, Integer>>)
- (element, stream) -> {
- PrintStream out = new
PrintStream(stream);
- out.println(element.f1);
- })
- .withBucketAssigner(new KeyBucketAssigner())
-
.withRollingPolicy(OnCheckpointRollingPolicy.build())
- .build();
- source.keyBy(x -> x.f0).sinkTo(sink);
- } else {
- throw new UnsupportedOperationException("Unsupported sink type: "
+ sinkToTest);
- }
-
- env.execute("StreamingFileSinkProgram");
+ FileSink<Tuple2<Integer, Integer>> sink =
+ FileSink.forRowFormat(
+ new Path(outputPath),
+ (Encoder<Tuple2<Integer, Integer>>)
+ (element, stream) -> {
+ PrintStream out = new
PrintStream(stream);
+ out.println(element.f1);
+ })
+ .withBucketAssigner(new KeyBucketAssigner())
+ .withRollingPolicy(OnCheckpointRollingPolicy.build())
+ .build();
+ source.keyBy(x -> x.f0).sinkTo(sink);
+
+ env.execute("FileSinkProgram");
}
/** Use first field for buckets. */
diff --git a/flink-end-to-end-tests/flink-stream-sql-test/pom.xml
b/flink-end-to-end-tests/flink-stream-sql-test/pom.xml
index 3abb2714225..4e14bd2eaff 100644
--- a/flink-end-to-end-tests/flink-stream-sql-test/pom.xml
+++ b/flink-end-to-end-tests/flink-stream-sql-test/pom.xml
@@ -41,6 +41,12 @@
<version>${project.version}</version>
<scope>provided</scope>
</dependency>
+ <!-- FileSink -->
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-files</artifactId>
+ <version>${project.version}</version>
+ </dependency>
</dependencies>
<build>
diff --git
a/flink-end-to-end-tests/flink-stream-sql-test/src/main/java/org/apache/flink/sql/tests/StreamSQLTestProgram.java
b/flink-end-to-end-tests/flink-stream-sql-test/src/main/java/org/apache/flink/sql/tests/StreamSQLTestProgram.java
index b34020afcb9..43f15fcce86 100644
---
a/flink-end-to-end-tests/flink-stream-sql-test/src/main/java/org/apache/flink/sql/tests/StreamSQLTestProgram.java
+++
b/flink-end-to-end-tests/flink-stream-sql-test/src/main/java/org/apache/flink/sql/tests/StreamSQLTestProgram.java
@@ -30,6 +30,7 @@ import
org.apache.flink.api.java.typeutils.ResultTypeQueryable;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestartStrategyOptions;
+import org.apache.flink.connector.file.sink.FileSink;
import org.apache.flink.core.fs.Path;
import org.apache.flink.core.io.SimpleVersionedSerializer;
import org.apache.flink.runtime.state.FunctionInitializationContext;
@@ -39,7 +40,6 @@ import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.filesystem.BucketAssigner;
import
org.apache.flink.streaming.api.functions.sink.filesystem.bucketassigners.SimpleVersionedStringSerializer;
-import
org.apache.flink.streaming.api.functions.sink.filesystem.legacy.StreamingFileSink;
import
org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.OnCheckpointRollingPolicy;
import org.apache.flink.streaming.api.functions.source.legacy.SourceFunction;
import org.apache.flink.table.api.DataTypes;
@@ -163,8 +163,8 @@ public class StreamSQLTestProgram {
DataTypes.ROW(
DataTypes.INT(),
DataTypes.TIMESTAMP().bridgedTo(Timestamp.class)));
- final StreamingFileSink<Row> sink =
- StreamingFileSink.forRowFormat(
+ final FileSink<Row> sink =
+ FileSink.forRowFormat(
new Path(outputPath),
(Encoder<Row>)
(element, stream) -> {
@@ -181,7 +181,7 @@ public class StreamSQLTestProgram {
.map(new KillMapper())
.setParallelism(1)
// add sink function
- .addSink(sink)
+ .sinkTo(sink)
.setParallelism(1);
sEnv.execute();
diff --git a/flink-end-to-end-tests/run-nightly-tests.sh
b/flink-end-to-end-tests/run-nightly-tests.sh
index eab0f865604..4b7eb708336 100755
--- a/flink-end-to-end-tests/run-nightly-tests.sh
+++ b/flink-end-to-end-tests/run-nightly-tests.sh
@@ -184,10 +184,8 @@ function run_group_3 {
run_test "Run kubernetes SQL application test"
"$END_TO_END_DIR/test-scripts/test_kubernetes_sql_application.sh"
run_test "Run kubernetes Materialized Table test"
"$END_TO_END_DIR/test-scripts/test_kubernetes_materialized_table.sh"
- run_test "Streaming File Sink end-to-end test"
"$END_TO_END_DIR/test-scripts/test_file_sink.sh local StreamingFileSink"
"skip_check_exceptions"
- run_test "Streaming File Sink s3 end-to-end test"
"$END_TO_END_DIR/test-scripts/test_file_sink.sh s3 StreamingFileSink"
"skip_check_exceptions"
- run_test "New File Sink end-to-end test"
"$END_TO_END_DIR/test-scripts/test_file_sink.sh local FileSink"
"skip_check_exceptions"
- run_test "New File Sink s3 end-to-end test"
"$END_TO_END_DIR/test-scripts/test_file_sink.sh s3 FileSink"
"skip_check_exceptions"
+ run_test "File Sink end-to-end test"
"$END_TO_END_DIR/test-scripts/test_file_sink.sh local" "skip_check_exceptions"
+ run_test "File Sink s3 end-to-end test"
"$END_TO_END_DIR/test-scripts/test_file_sink.sh s3" "skip_check_exceptions"
run_test "Wordcount Hadoop S3 SeaweedFS read-write end-to-end test"
"$END_TO_END_DIR/test-scripts/test_batch_wordcount.sh hadoop_seaweedfs"
run_test "Wordcount Presto S3 SeaweedFS read end-to-end test"
"$END_TO_END_DIR/test-scripts/test_batch_wordcount.sh presto_seaweedfs_read"
diff --git a/flink-end-to-end-tests/test-scripts/test_file_sink.sh
b/flink-end-to-end-tests/test-scripts/test_file_sink.sh
index f35369f95f3..5de4731ca4c 100755
--- a/flink-end-to-end-tests/test-scripts/test_file_sink.sh
+++ b/flink-end-to-end-tests/test-scripts/test_file_sink.sh
@@ -18,7 +18,6 @@
################################################################################
OUT_TYPE="${1:-local}" # other type: s3
-SINK_TO_TEST="${2:-"StreamingFileSink"}"
source "$(dirname "$0")"/common.sh
@@ -153,8 +152,7 @@ function run_file_sink_test {
"${FLINK_DIR}/bin/taskmanager.sh" start
echo "Submitting job."
- CLIENT_OUTPUT=$("$FLINK_DIR/bin/flink" run -d "${TEST_PROGRAM_JAR}"
--outputPath "${JOB_OUTPUT_PATH}" \
- --sinkToTest "${SINK_TO_TEST}")
+ CLIENT_OUTPUT=$("$FLINK_DIR/bin/flink" run -d "${TEST_PROGRAM_JAR}"
--outputPath "${JOB_OUTPUT_PATH}")
JOB_ID=$(echo "${CLIENT_OUTPUT}" | grep "Job has been submitted with JobID"
| sed 's/.* //g')
if [[ -z $JOB_ID ]]; then
@@ -194,7 +192,7 @@ function run_file_sink_test {
get_complete_result > "${TEST_DATA_DIR}/complete_result"
- check_result_hash "File Streaming Sink" "$TEST_DATA_DIR/complete_result"
"6727342fdd3aae2129e61fc8f433fb6f"
+ check_result_hash "File Sink" "$TEST_DATA_DIR/complete_result"
"6727342fdd3aae2129e61fc8f433fb6f"
}
# usual runtime is ~6 minutes
diff --git
a/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroStreamingFileSinkITCase.java
b/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroFileSinkITCase.java
similarity index 93%
rename from
flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroStreamingFileSinkITCase.java
rename to
flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroFileSinkITCase.java
index e4259e65143..a549db848a0 100644
---
a/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroStreamingFileSinkITCase.java
+++
b/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroFileSinkITCase.java
@@ -21,13 +21,13 @@ package org.apache.flink.formats.avro;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.connector.datagen.source.TestDataGenerators;
+import org.apache.flink.connector.file.sink.FileSink;
import org.apache.flink.core.fs.Path;
import org.apache.flink.formats.avro.generated.Address;
import org.apache.flink.formats.avro.typeutils.GenericRecordAvroTypeInfo;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import
org.apache.flink.streaming.api.functions.sink.filesystem.bucketassigners.UniqueBucketAssigner;
-import
org.apache.flink.streaming.api.functions.sink.filesystem.legacy.StreamingFileSink;
import org.apache.flink.test.util.AbstractTestBase;
import org.apache.avro.Schema;
@@ -56,11 +56,10 @@ import java.util.concurrent.TimeUnit;
import static org.assertj.core.api.Assertions.assertThat;
/**
- * Simple integration test case for writing bulk encoded files with the {@link
StreamingFileSink}
- * with Avro.
+ * Simple integration test case for writing bulk encoded files with the {@link
FileSink} with Avro.
*/
@Timeout(value = 20, unit = TimeUnit.SECONDS)
-class AvroStreamingFileSinkITCase extends AbstractTestBase {
+class AvroFileSinkITCase extends AbstractTestBase {
@Test
void testWriteAvroSpecific(@TempDir File folder) throws Exception {
@@ -82,8 +81,8 @@ class AvroStreamingFileSinkITCase extends AbstractTestBase {
WatermarkStrategy.noWatermarks(),
"Test Source");
- stream.addSink(
- StreamingFileSink.forBulkFormat(Path.fromLocalFile(folder),
avroWriterFactory)
+ stream.sinkTo(
+ FileSink.forBulkFormat(Path.fromLocalFile(folder),
avroWriterFactory)
.withBucketAssigner(new UniqueBucketAssigner<>("test"))
.build());
env.execute();
@@ -108,8 +107,8 @@ class AvroStreamingFileSinkITCase extends AbstractTestBase {
WatermarkStrategy.noWatermarks(),
"Test Source");
- stream.addSink(
- StreamingFileSink.forBulkFormat(Path.fromLocalFile(folder),
avroWriterFactory)
+ stream.sinkTo(
+ FileSink.forBulkFormat(Path.fromLocalFile(folder),
avroWriterFactory)
.withBucketAssigner(new UniqueBucketAssigner<>("test"))
.build());
env.execute();
@@ -134,8 +133,8 @@ class AvroStreamingFileSinkITCase extends AbstractTestBase {
WatermarkStrategy.noWatermarks(),
"Test Source");
- stream.addSink(
- StreamingFileSink.forBulkFormat(Path.fromLocalFile(folder),
avroWriterFactory)
+ stream.sinkTo(
+ FileSink.forBulkFormat(Path.fromLocalFile(folder),
avroWriterFactory)
.withBucketAssigner(new UniqueBucketAssigner<>("test"))
.build());
env.execute();
diff --git a/flink-formats/flink-compress/pom.xml
b/flink-formats/flink-compress/pom.xml
index 06392a9c0cd..7545fbea180 100644
--- a/flink-formats/flink-compress/pom.xml
+++ b/flink-formats/flink-compress/pom.xml
@@ -89,10 +89,9 @@ under the License.
<dependency>
<groupId>org.apache.flink</groupId>
- <artifactId>flink-streaming-java</artifactId>
+ <artifactId>flink-connector-files</artifactId>
<version>${project.version}</version>
<scope>test</scope>
- <type>test-jar</type>
</dependency>
<dependency>
diff --git
a/flink-formats/flink-compress/src/test/java/org/apache/flink/formats/compress/CompressWriterFactoryTest.java
b/flink-formats/flink-compress/src/test/java/org/apache/flink/formats/compress/CompressWriterFactoryTest.java
index 227b21ec2da..3ff7534b282 100644
---
a/flink-formats/flink-compress/src/test/java/org/apache/flink/formats/compress/CompressWriterFactoryTest.java
+++
b/flink-formats/flink-compress/src/test/java/org/apache/flink/formats/compress/CompressWriterFactoryTest.java
@@ -18,13 +18,9 @@
package org.apache.flink.formats.compress;
-import org.apache.flink.core.fs.Path;
+import org.apache.flink.api.common.serialization.BulkWriter;
+import org.apache.flink.core.fs.local.LocalDataOutputStream;
import org.apache.flink.formats.compress.extractor.DefaultExtractor;
-import
org.apache.flink.streaming.api.functions.sink.filesystem.bucketassigners.UniqueBucketAssigner;
-import
org.apache.flink.streaming.api.functions.sink.filesystem.legacy.StreamingFileSink;
-import org.apache.flink.streaming.api.operators.StreamSink;
-import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
-import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.compress.CompressionCodec;
@@ -132,54 +128,36 @@ class CompressWriterFactoryTest {
.withHadoopCompression(codec, conf);
List<String> lines = Arrays.asList("line1", "line2", "line3");
- File directory = prepareCompressedFile(codec, writer, lines);
+ File partFile = writeCompressedFile(codec, writer, lines);
- validateResults(directory, lines, new
CompressionCodecFactory(conf).getCodecByName(codec));
+ validateResult(partFile, lines, new
CompressionCodecFactory(conf).getCodecByName(codec));
}
- private File prepareCompressedFile(
+ private File writeCompressedFile(
String codec, CompressWriterFactory<String> writer, List<String>
lines)
throws Exception {
final File outDir = tmpDir.resolve(codec).toFile();
assertThat(outDir.mkdirs()).isTrue();
+ final File partFile = new File(outDir, "part-0-0");
- StreamingFileSink<String> sink =
- StreamingFileSink.forBulkFormat(new Path(outDir.toURI()),
writer)
- .withBucketAssigner(new UniqueBucketAssigner<>("test"))
- .build();
-
- try (OneInputStreamOperatorTestHarness<String, Object> testHarness =
- new OneInputStreamOperatorTestHarness<>(new
StreamSink<>(sink), 1, 1, 0)) {
- testHarness.setup();
- testHarness.open();
-
- int time = 0;
+ // finish() writes the codec trailer and flushes the compressed bytes;
closing the
+ // stream releases the file.
+ try (LocalDataOutputStream out = new LocalDataOutputStream(partFile)) {
+ final BulkWriter<String> bulkWriter = writer.create(out);
for (String line : lines) {
- testHarness.processElement(new StreamRecord<>(line, ++time));
+ bulkWriter.addElement(line);
}
-
- testHarness.snapshot(1, ++time);
- testHarness.notifyOfCompletedCheckpoint(1);
+ bulkWriter.finish();
}
- return outDir;
+ return partFile;
}
- private void validateResults(File folder, List<String> expected,
CompressionCodec codec)
+ private void validateResult(File partFile, List<String> expected,
CompressionCodec codec)
throws Exception {
- File[] buckets = folder.listFiles();
- assertThat(buckets).isNotNull();
- assertThat(buckets).hasSize(1);
-
- final File[] partFiles = buckets[0].listFiles();
- assertThat(partFiles).isNotNull();
- assertThat(partFiles).hasSize(1);
-
- for (File partFile : partFiles) {
- assertThat(partFile.length()).isGreaterThan(0);
- final List<String> fileContent = readFile(partFile, codec);
- assertThat(fileContent).isEqualTo(expected);
- }
+ assertThat(partFile.length()).isGreaterThan(0);
+ final List<String> fileContent = readFile(partFile, codec);
+ assertThat(fileContent).isEqualTo(expected);
}
private List<String> readFile(File file, CompressionCodec codec) throws
Exception {
diff --git
a/flink-formats/flink-compress/src/test/java/org/apache/flink/formats/compress/CompressionFactoryITCase.java
b/flink-formats/flink-compress/src/test/java/org/apache/flink/formats/compress/CompressionFactoryITCase.java
index 1ae3369ee6d..77ed70f4c39 100644
---
a/flink-formats/flink-compress/src/test/java/org/apache/flink/formats/compress/CompressionFactoryITCase.java
+++
b/flink-formats/flink-compress/src/test/java/org/apache/flink/formats/compress/CompressionFactoryITCase.java
@@ -21,12 +21,12 @@ package org.apache.flink.formats.compress;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.connector.datagen.source.TestDataGenerators;
+import org.apache.flink.connector.file.sink.FileSink;
import org.apache.flink.core.fs.Path;
import org.apache.flink.formats.compress.extractor.DefaultExtractor;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import
org.apache.flink.streaming.api.functions.sink.filesystem.bucketassigners.UniqueBucketAssigner;
-import
org.apache.flink.streaming.api.functions.sink.filesystem.legacy.StreamingFileSink;
import org.apache.flink.test.junit5.MiniClusterExtension;
import org.apache.hadoop.conf.Configuration;
@@ -47,8 +47,8 @@ import java.util.stream.Collectors;
import static org.assertj.core.api.Assertions.assertThat;
/**
- * Integration test case for writing bulk encoded files with the {@link
StreamingFileSink} and
- * Hadoop Compression Codecs.
+ * Integration test case for writing bulk encoded files with the {@link
FileSink} and Hadoop
+ * Compression Codecs.
*/
@ExtendWith(MiniClusterExtension.class)
class CompressionFactoryITCase {
@@ -75,8 +75,8 @@ class CompressionFactoryITCase {
"Test Source");
stream.map(str -> str)
- .addSink(
- StreamingFileSink.forBulkFormat(
+ .sinkTo(
+ FileSink.forBulkFormat(
testPath,
CompressWriters.forExtractor(new
DefaultExtractor<String>())
.withHadoopCompression(TEST_CODEC_NAME))
diff --git
a/flink-formats/flink-orc/src/test/java/org/apache/flink/orc/OrcFileSystemITCase.java
b/flink-formats/flink-orc/src/test/java/org/apache/flink/orc/OrcFileSystemITCase.java
index 7cc9f5452b4..c42a7170455 100644
---
a/flink-formats/flink-orc/src/test/java/org/apache/flink/orc/OrcFileSystemITCase.java
+++
b/flink-formats/flink-orc/src/test/java/org/apache/flink/orc/OrcFileSystemITCase.java
@@ -18,11 +18,9 @@
package org.apache.flink.orc;
+import org.apache.flink.orc.util.OrcBulkWriterTestUtil;
import org.apache.flink.orc.vector.RowDataVectorizer;
import org.apache.flink.orc.writer.OrcBulkWriterFactory;
-import
org.apache.flink.streaming.api.functions.sink.filesystem.legacy.StreamingFileSink;
-import org.apache.flink.streaming.api.operators.StreamSink;
-import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.data.GenericArrayData;
import org.apache.flink.table.data.GenericMapData;
@@ -325,31 +323,13 @@ public class OrcFileSystemITCase extends
BatchFileSystemITCaseBase {
Properties writerProps = new Properties();
writerProps.setProperty("orc.compress", "LZ4");
- final OrcBulkWriterFactory<RowData> writer =
+ final OrcBulkWriterFactory<RowData> writerFactory =
new OrcBulkWriterFactory<>(
new RowDataVectorizer(schema, fieldTypes),
writerProps,
new Configuration());
- StreamingFileSink<RowData> sink =
- StreamingFileSink.forBulkFormat(
- new
org.apache.flink.core.fs.Path(outDir.toURI()), writer)
- .withBucketCheckInterval(10000)
- .build();
-
- try (OneInputStreamOperatorTestHarness<RowData, Object> testHarness =
- new OneInputStreamOperatorTestHarness<>(new
StreamSink<>(sink), 1, 1, 0)) {
-
- testHarness.setup();
- testHarness.open();
-
- int time = 0;
- for (final RowData record : data) {
- testHarness.processElement(record, ++time);
- }
- testHarness.snapshot(1, ++time);
- testHarness.notifyOfCompletedCheckpoint(1);
- }
+ OrcBulkWriterTestUtil.writeRecordsToFile(new File(outDir, "part-0-0"),
writerFactory, data);
return outDir.getAbsolutePath();
}
diff --git
a/flink-formats/flink-orc/src/test/java/org/apache/flink/orc/util/OrcBulkWriterTestUtil.java
b/flink-formats/flink-orc/src/test/java/org/apache/flink/orc/util/OrcBulkWriterTestUtil.java
index f5c8f604e18..fd56bd07d17 100644
---
a/flink-formats/flink-orc/src/test/java/org/apache/flink/orc/util/OrcBulkWriterTestUtil.java
+++
b/flink-formats/flink-orc/src/test/java/org/apache/flink/orc/util/OrcBulkWriterTestUtil.java
@@ -18,6 +18,8 @@
package org.apache.flink.orc.util;
+import org.apache.flink.api.common.serialization.BulkWriter;
+import org.apache.flink.core.fs.local.LocalDataOutputStream;
import org.apache.flink.orc.data.Record;
import org.apache.hadoop.conf.Configuration;
@@ -43,6 +45,21 @@ public class OrcBulkWriterTestUtil {
public static final String USER_METADATA_KEY = "userKey";
public static final ByteBuffer USER_METADATA_VALUE =
ByteBuffer.wrap("hello".getBytes());
+ /**
+ * Writes the records to the given file through a {@link BulkWriter}
created from the factory.
+ * finish() writes the ORC footer and closes the underlying stream.
+ */
+ public static <T> void writeRecordsToFile(
+ File file, BulkWriter.Factory<T> writerFactory, List<T> records)
throws IOException {
+ try (LocalDataOutputStream out = new LocalDataOutputStream(file)) {
+ final BulkWriter<T> writer = writerFactory.create(out);
+ for (final T record : records) {
+ writer.addElement(record);
+ }
+ writer.finish();
+ }
+ }
+
public static void validate(File files, List<Record> expected) throws
IOException {
final File[] buckets = files.listFiles();
assertThat(buckets).isNotNull();
diff --git
a/flink-formats/flink-orc/src/test/java/org/apache/flink/orc/writer/OrcBulkRowDataWriterTest.java
b/flink-formats/flink-orc/src/test/java/org/apache/flink/orc/writer/OrcBulkRowDataWriterTest.java
index 82837e43756..6d32680e9bb 100644
---
a/flink-formats/flink-orc/src/test/java/org/apache/flink/orc/writer/OrcBulkRowDataWriterTest.java
+++
b/flink-formats/flink-orc/src/test/java/org/apache/flink/orc/writer/OrcBulkRowDataWriterTest.java
@@ -18,12 +18,8 @@
package org.apache.flink.orc.writer;
-import org.apache.flink.core.fs.Path;
+import org.apache.flink.orc.util.OrcBulkWriterTestUtil;
import org.apache.flink.orc.vector.RowDataVectorizer;
-import
org.apache.flink.streaming.api.functions.sink.filesystem.bucketassigners.UniqueBucketAssigner;
-import
org.apache.flink.streaming.api.functions.sink.filesystem.legacy.StreamingFileSink;
-import org.apache.flink.streaming.api.operators.StreamSink;
-import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
import org.apache.flink.table.data.ArrayData;
import org.apache.flink.table.data.GenericArrayData;
import org.apache.flink.table.data.GenericMapData;
@@ -86,34 +82,20 @@ class OrcBulkRowDataWriterTest {
final Properties writerProps = new Properties();
writerProps.setProperty("orc.compress", "LZ4");
- final OrcBulkWriterFactory<RowData> writer =
+ final OrcBulkWriterFactory<RowData> writerFactory =
new OrcBulkWriterFactory<>(
new RowDataVectorizer(schema, fieldTypes),
writerProps,
new Configuration());
- StreamingFileSink<RowData> sink =
- StreamingFileSink.forBulkFormat(new Path(outDir.toURI()),
writer)
- .withBucketAssigner(new UniqueBucketAssigner<>("test"))
- .withBucketCheckInterval(10000)
- .build();
+ // Mimic a single bucket directory so the validate(...) helper applies.
+ final File bucketDir = new File(outDir, "test");
+ assertThat(bucketDir.mkdirs()).isTrue();
- try (OneInputStreamOperatorTestHarness<RowData, Object> testHarness =
- new OneInputStreamOperatorTestHarness<>(new
StreamSink<>(sink), 1, 1, 0)) {
+ OrcBulkWriterTestUtil.writeRecordsToFile(
+ new File(bucketDir, "part-0-0"), writerFactory, input);
- testHarness.setup();
- testHarness.open();
-
- int time = 0;
- for (final RowData record : input) {
- testHarness.processElement(record, ++time);
- }
-
- testHarness.snapshot(1, ++time);
- testHarness.notifyOfCompletedCheckpoint(1);
-
- validate(outDir, input);
- }
+ validate(outDir, input);
}
@BeforeEach
diff --git
a/flink-formats/flink-orc/src/test/java/org/apache/flink/orc/writer/OrcBulkWriterITCase.java
b/flink-formats/flink-orc/src/test/java/org/apache/flink/orc/writer/OrcBulkWriterITCase.java
index 2cbb68ff524..53a0a44071f 100644
---
a/flink-formats/flink-orc/src/test/java/org/apache/flink/orc/writer/OrcBulkWriterITCase.java
+++
b/flink-formats/flink-orc/src/test/java/org/apache/flink/orc/writer/OrcBulkWriterITCase.java
@@ -21,6 +21,7 @@ package org.apache.flink.orc.writer;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.connector.datagen.source.TestDataGenerators;
+import org.apache.flink.connector.file.sink.FileSink;
import org.apache.flink.core.fs.Path;
import org.apache.flink.orc.data.Record;
import org.apache.flink.orc.util.OrcBulkWriterTestUtil;
@@ -28,7 +29,6 @@ import org.apache.flink.orc.vector.RecordVectorizer;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import
org.apache.flink.streaming.api.functions.sink.filesystem.bucketassigners.UniqueBucketAssigner;
-import
org.apache.flink.streaming.api.functions.sink.filesystem.legacy.StreamingFileSink;
import org.apache.hadoop.conf.Configuration;
import org.junit.jupiter.api.Test;
@@ -39,7 +39,7 @@ import java.util.Arrays;
import java.util.List;
import java.util.Properties;
-/** Integration test for writing data in ORC bulk format using
StreamingFileSink. */
+/** Integration test for writing data in ORC bulk format using {@link
FileSink}. */
class OrcBulkWriterITCase {
private final String schema = "struct<_col0:string,_col1:int>";
@@ -67,8 +67,8 @@ class OrcBulkWriterITCase {
"Test Source");
stream.map(str -> str)
- .addSink(
- StreamingFileSink.forBulkFormat(new
Path(outDir.toURI()), factory)
+ .sinkTo(
+ FileSink.forBulkFormat(new Path(outDir.toURI()),
factory)
.withBucketAssigner(new
UniqueBucketAssigner<>("test"))
.build());
diff --git
a/flink-formats/flink-orc/src/test/java/org/apache/flink/orc/writer/OrcBulkWriterTest.java
b/flink-formats/flink-orc/src/test/java/org/apache/flink/orc/writer/OrcBulkWriterTest.java
index e87785902dc..75b9a851da5 100644
---
a/flink-formats/flink-orc/src/test/java/org/apache/flink/orc/writer/OrcBulkWriterTest.java
+++
b/flink-formats/flink-orc/src/test/java/org/apache/flink/orc/writer/OrcBulkWriterTest.java
@@ -18,14 +18,9 @@
package org.apache.flink.orc.writer;
-import org.apache.flink.core.fs.Path;
import org.apache.flink.orc.data.Record;
import org.apache.flink.orc.util.OrcBulkWriterTestUtil;
import org.apache.flink.orc.vector.RecordVectorizer;
-import
org.apache.flink.streaming.api.functions.sink.filesystem.bucketassigners.UniqueBucketAssigner;
-import
org.apache.flink.streaming.api.functions.sink.filesystem.legacy.StreamingFileSink;
-import org.apache.flink.streaming.api.operators.StreamSink;
-import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
import org.apache.hadoop.conf.Configuration;
import org.junit.jupiter.api.Test;
@@ -36,6 +31,8 @@ import java.util.Arrays;
import java.util.List;
import java.util.Properties;
+import static org.assertj.core.api.Assertions.assertThat;
+
/** Unit test for the ORC BulkWriter implementation. */
class OrcBulkWriterTest {
@@ -48,31 +45,17 @@ class OrcBulkWriterTest {
final Properties writerProps = new Properties();
writerProps.setProperty("orc.compress", "LZ4");
- final OrcBulkWriterFactory<Record> writer =
+ final OrcBulkWriterFactory<Record> writerFactory =
new OrcBulkWriterFactory<>(
new RecordVectorizer(schema), writerProps, new
Configuration());
- StreamingFileSink<Record> sink =
- StreamingFileSink.forBulkFormat(new Path(outDir.toURI()),
writer)
- .withBucketAssigner(new UniqueBucketAssigner<>("test"))
- .withBucketCheckInterval(10000)
- .build();
-
- try (OneInputStreamOperatorTestHarness<Record, Object> testHarness =
- new OneInputStreamOperatorTestHarness<>(new
StreamSink<>(sink), 1, 1, 0)) {
-
- testHarness.setup();
- testHarness.open();
-
- int time = 0;
- for (final Record record : input) {
- testHarness.processElement(record, ++time);
- }
+ // Mimic a single bucket directory so the shared
OrcBulkWriterTestUtil.validate applies.
+ final File bucketDir = new File(outDir, "test");
+ assertThat(bucketDir.mkdirs()).isTrue();
- testHarness.snapshot(1, ++time);
- testHarness.notifyOfCompletedCheckpoint(1);
+ OrcBulkWriterTestUtil.writeRecordsToFile(
+ new File(bucketDir, "part-0-0"), writerFactory, input);
- OrcBulkWriterTestUtil.validate(outDir, input);
- }
+ OrcBulkWriterTestUtil.validate(outDir, input);
}
}
diff --git
a/flink-formats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/avro/AvroParquetStreamingFileSinkITCase.java
b/flink-formats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/avro/AvroParquetFileSinkITCase.java
similarity index 95%
rename from
flink-formats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/avro/AvroParquetStreamingFileSinkITCase.java
rename to
flink-formats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/avro/AvroParquetFileSinkITCase.java
index 6e4cb845ccd..0e43da01127 100644
---
a/flink-formats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/avro/AvroParquetStreamingFileSinkITCase.java
+++
b/flink-formats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/avro/AvroParquetFileSinkITCase.java
@@ -21,13 +21,13 @@ package org.apache.flink.formats.parquet.avro;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.connector.datagen.source.TestDataGenerators;
+import org.apache.flink.connector.file.sink.FileSink;
import org.apache.flink.core.fs.Path;
import org.apache.flink.formats.avro.typeutils.GenericRecordAvroTypeInfo;
import org.apache.flink.formats.parquet.generated.Address;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import
org.apache.flink.streaming.api.functions.sink.filesystem.bucketassigners.UniqueBucketAssigner;
-import
org.apache.flink.streaming.api.functions.sink.filesystem.legacy.StreamingFileSink;
import org.apache.flink.test.junit5.MiniClusterExtension;
import org.apache.avro.Schema;
@@ -57,11 +57,11 @@ import java.util.List;
import static org.assertj.core.api.Assertions.assertThat;
/**
- * Simple integration test case for writing bulk encoded files with the {@link
StreamingFileSink}
- * with Parquet.
+ * Simple integration test case for writing bulk encoded files with the {@link
FileSink} with
+ * Parquet.
*/
@ExtendWith(MiniClusterExtension.class)
-class AvroParquetStreamingFileSinkITCase {
+class AvroParquetFileSinkITCase {
@Test
void testWriteParquetAvroSpecific(@TempDir File folder) throws Exception {
@@ -83,8 +83,8 @@ class AvroParquetStreamingFileSinkITCase {
WatermarkStrategy.noWatermarks(),
"Test Source");
- stream.addSink(
- StreamingFileSink.forBulkFormat(
+ stream.sinkTo(
+ FileSink.forBulkFormat(
Path.fromLocalFile(folder),
AvroParquetWriters.forSpecificRecord(Address.class))
.withBucketAssigner(new UniqueBucketAssigner<>("test"))
@@ -113,8 +113,8 @@ class AvroParquetStreamingFileSinkITCase {
WatermarkStrategy.noWatermarks(),
"Test Source");
- stream.addSink(
- StreamingFileSink.forBulkFormat(
+ stream.sinkTo(
+ FileSink.forBulkFormat(
Path.fromLocalFile(folder),
AvroParquetWriters.forGenericRecord(schema))
.withBucketAssigner(new UniqueBucketAssigner<>("test"))
@@ -147,8 +147,8 @@ class AvroParquetStreamingFileSinkITCase {
WatermarkStrategy.noWatermarks(),
"Test Source");
- stream.addSink(
- StreamingFileSink.forBulkFormat(
+ stream.sinkTo(
+ FileSink.forBulkFormat(
Path.fromLocalFile(folder),
AvroParquetWriters.forReflectRecord(Datum.class))
.withBucketAssigner(new UniqueBucketAssigner<>("test"))
diff --git
a/flink-formats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/protobuf/ParquetProtoStreamingFileSinkITCase.java
b/flink-formats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/protobuf/ParquetProtoFileSinkITCase.java
similarity index 95%
rename from
flink-formats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/protobuf/ParquetProtoStreamingFileSinkITCase.java
rename to
flink-formats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/protobuf/ParquetProtoFileSinkITCase.java
index f4e42634827..7a8718b9d23 100644
---
a/flink-formats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/protobuf/ParquetProtoStreamingFileSinkITCase.java
+++
b/flink-formats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/protobuf/ParquetProtoFileSinkITCase.java
@@ -21,11 +21,11 @@ package org.apache.flink.formats.parquet.protobuf;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.connector.datagen.source.TestDataGenerators;
+import org.apache.flink.connector.file.sink.FileSink;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import
org.apache.flink.streaming.api.functions.sink.filesystem.bucketassigners.UniqueBucketAssigner;
-import
org.apache.flink.streaming.api.functions.sink.filesystem.legacy.StreamingFileSink;
import org.apache.flink.test.junit5.MiniClusterExtension;
import com.google.protobuf.Message;
@@ -47,11 +47,11 @@ import static
org.apache.flink.formats.parquet.protobuf.SimpleRecord.SimpleProto
import static org.assertj.core.api.Assertions.assertThat;
/**
- * Simple integration test case for writing bulk encoded files with the {@link
StreamingFileSink}
- * with Parquet.
+ * Simple integration test case for writing bulk encoded files with the {@link
FileSink} with
+ * Parquet.
*/
@ExtendWith(MiniClusterExtension.class)
-class ParquetProtoStreamingFileSinkITCase {
+class ParquetProtoFileSinkITCase {
@Test
void testParquetProtoWriters(@TempDir File folder) throws Exception {
@@ -73,8 +73,8 @@ class ParquetProtoStreamingFileSinkITCase {
WatermarkStrategy.noWatermarks(),
"Test Source");
- stream.addSink(
- StreamingFileSink.forBulkFormat(
+ stream.sinkTo(
+ FileSink.forBulkFormat(
Path.fromLocalFile(folder),
ParquetProtoWriters.forType(SimpleProtoRecord.class))
.withBucketAssigner(new UniqueBucketAssigner<>("test"))
diff --git a/flink-formats/flink-sequence-file/pom.xml
b/flink-formats/flink-sequence-file/pom.xml
index 509eda9d4a7..a77bb6fb64a 100644
--- a/flink-formats/flink-sequence-file/pom.xml
+++ b/flink-formats/flink-sequence-file/pom.xml
@@ -89,7 +89,7 @@ under the License.
<dependency>
<groupId>org.apache.flink</groupId>
- <artifactId>flink-streaming-java</artifactId>
+ <artifactId>flink-connector-files</artifactId>
<version>${project.version}</version>
<scope>test</scope>
</dependency>
diff --git
a/flink-formats/flink-sequence-file/src/test/java/org/apache/flink/formats/sequencefile/SequenceStreamingFileSinkITCase.java
b/flink-formats/flink-sequence-file/src/test/java/org/apache/flink/formats/sequencefile/SequenceFileSinkITCase.java
similarity index 95%
rename from
flink-formats/flink-sequence-file/src/test/java/org/apache/flink/formats/sequencefile/SequenceStreamingFileSinkITCase.java
rename to
flink-formats/flink-sequence-file/src/test/java/org/apache/flink/formats/sequencefile/SequenceFileSinkITCase.java
index 37b9626877c..7ced5df29f6 100644
---
a/flink-formats/flink-sequence-file/src/test/java/org/apache/flink/formats/sequencefile/SequenceStreamingFileSinkITCase.java
+++
b/flink-formats/flink-sequence-file/src/test/java/org/apache/flink/formats/sequencefile/SequenceFileSinkITCase.java
@@ -24,11 +24,11 @@ import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.datagen.source.TestDataGenerators;
+import org.apache.flink.connector.file.sink.FileSink;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import
org.apache.flink.streaming.api.functions.sink.filesystem.bucketassigners.UniqueBucketAssigner;
-import
org.apache.flink.streaming.api.functions.sink.filesystem.legacy.StreamingFileSink;
import org.apache.flink.test.junit5.MiniClusterExtension;
import org.apache.hadoop.conf.Configuration;
@@ -48,11 +48,10 @@ import java.util.List;
import static org.assertj.core.api.Assertions.assertThat;
/**
- * Integration test case for writing bulk encoded files with the {@link
StreamingFileSink} with
- * SequenceFile.
+ * Integration test case for writing bulk encoded files with the {@link
FileSink} with SequenceFile.
*/
@ExtendWith(MiniClusterExtension.class)
-class SequenceStreamingFileSinkITCase {
+class SequenceFileSinkITCase {
private final Configuration configuration = new Configuration();
@@ -83,8 +82,8 @@ class SequenceStreamingFileSinkITCase {
return new Tuple2<>(new
LongWritable(value.f0), new Text(value.f1));
}
})
- .addSink(
- StreamingFileSink.forBulkFormat(
+ .sinkTo(
+ FileSink.forBulkFormat(
testPath,
new SequenceFileWriterFactory<>(
configuration,
