This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 8f5f9ba6f92c feat(flink): enable Sink V2 on Flink 1.18 via
WithPreWriteTopology (#19982)
8f5f9ba6f92c is described below
commit 8f5f9ba6f92c0cd02b175d846b98dc2f4dea518f
Author: ericyuan915 <[email protected]>
AuthorDate: Thu Sep 17 01:44:53 2026 -0700
feat(flink): enable Sink V2 on Flink 1.18 via WithPreWriteTopology (#19982)
HoodieSink builds its whole write pipeline in addPreWriteTopology, which
Flink
only invokes for a sink implementing the pre-write-topology interface of the
running version. The Flink 1.18 SupportsPreWriteTopologyAdapter extended
nothing -- it existed only to keep the shared hudi-flink module compiling
-- so
on 1.18 the hook was never called and PipelinesV2.sink fenced Sink V2 off
behind a runtime version check.
Flink 1.18 does have the capability, named WithPreWriteTopology, which
SinkTransformationTranslator dispatches on there. Flink 1.19 split the
identical method out into SupportsPreWriteTopology and deprecated
WithPreWriteTopology, so only the type name and its Sink supertype differ.
Extend the real interface in the 1.18 adapter and inherit
addPreWriteTopology
instead of redeclaring it. HoodieSink then reaches Sink<RowData> through
both
SinkAdapter and WithPreWriteTopology, which is legal: same type argument,
and
SinkAdapter's createWriter(InitContext) default overrides the abstract
declaration inherited from Sink. Drop the version guard accordingly --
hudi-flink1.18.x is the oldest Flink profile on master, so it is
unreachable.
Add TestPipelinesV2#testSinkLetsFlinkExpandTheHudiWritePipeline, which runs
PipelinesV2.sink through Flink's own SinkTransformationTranslator by
generating
the StreamGraph, then asserts the Hudi write operator is in the graph with
the
sink uid prefixed to its own. It names no version-specific interface, so it
covers the pre-write hook on every Flink profile.
closes #19966
---
.../org/apache/hudi/sink/v2/utils/PipelinesV2.java | 5 ----
.../apache/hudi/sink/v2/utils/TestPipelinesV2.java | 28 ++++++++++++++++++++++
.../adapter/SupportsPreWriteTopologyAdapter.java | 21 +++++++---------
3 files changed, 36 insertions(+), 18 deletions(-)
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/v2/utils/PipelinesV2.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/v2/utils/PipelinesV2.java
index 5cf19bfd958a..60b3d875dbd1 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/v2/utils/PipelinesV2.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/v2/utils/PipelinesV2.java
@@ -19,7 +19,6 @@
package org.apache.hudi.sink.v2.utils;
import org.apache.hudi.client.model.HoodieFlinkInternalRow;
-import org.apache.hudi.common.util.ValidationUtils;
import org.apache.hudi.configuration.FlinkOptions;
import org.apache.hudi.configuration.OptionsResolver;
import org.apache.hudi.exception.HoodieException;
@@ -37,7 +36,6 @@ import org.apache.hudi.sink.v2.HoodieSink;
import org.apache.hudi.sink.v2.clustering.ClusteringCommitSinkV2;
import org.apache.hudi.sink.v2.compact.CompactionCommitSinkV2;
-import org.apache.flink.FlinkVersion;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
@@ -55,7 +53,6 @@ import static org.apache.hudi.sink.utils.Pipelines.opUID;
public class PipelinesV2 {
private static final String SINK_V2_NAME = "sink_v2";
- private static final String FLINK_1_18_VERSION = "1.18";
/**
* Construct a write pipeline based on {@link HoodieSink}, which is
implemented
@@ -75,8 +72,6 @@ public class PipelinesV2 {
RowType rowType,
boolean overwrite,
boolean isBounded) {
-
ValidationUtils.checkArgument(FlinkVersion.current().toString().compareTo(FLINK_1_18_VERSION)
> 0,
- "Hudi sink v2 is not supported with Flink version: " +
FlinkVersion.current());
HoodieSink hoodieSink = new HoodieSink(conf, rowType, overwrite,
isBounded);
return dataStream.sinkTo(hoodieSink)
.setParallelism(getParallelismForSinkV2(conf))
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/v2/utils/TestPipelinesV2.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/v2/utils/TestPipelinesV2.java
index a56cc694cc91..5ade4b2b1c21 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/v2/utils/TestPipelinesV2.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/v2/utils/TestPipelinesV2.java
@@ -31,6 +31,7 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.graph.StreamNode;
import org.apache.flink.streaming.api.transformations.PartitionTransformation;
import org.apache.flink.streaming.runtime.partitioner.CustomPartitionerWrapper;
import org.apache.flink.streaming.runtime.partitioner.StreamPartitioner;
@@ -83,6 +84,33 @@ class TestPipelinesV2 {
.matches("uid_sink_v2(?:_\\d+)?_sink_v2_test"));
}
+ @Test
+ void testSinkLetsFlinkExpandTheHudiWritePipeline() {
+ conf.set(FlinkOptions.OPERATION, "insert");
+ conf.set(FlinkOptions.TABLE_NAME, "sink_v2_test");
+ conf.set(FlinkOptions.WRITE_TASKS, 4);
+
+ DataStreamSink<RowData> sink = PipelinesV2.sink(
+ input, conf, TestConfigurations.ROW_TYPE, false, true);
+ String sinkUid = sink.getTransformation().getUid();
+
+ // HoodieSink builds the whole write pipeline in addPreWriteTopology, and
Flink only calls that
+ // for a sink implementing the pre-write topology interface of the running
version
+ // (WithPreWriteTopology on 1.18, SupportsPreWriteTopology since 1.19).
Without it the job
+ // keeps nothing but the no-op sink writer and silently writes no data.
+ List<StreamNode> writeNodes =
input.getExecutionEnvironment().getStreamGraph()
+ .getStreamNodes().stream()
+ .filter(node -> node.getOperatorName().contains("hoodie_append_write"))
+ .collect(Collectors.toList());
+
+ assertEquals(1, writeNodes.size(), "Flink did not expand the pre-write
topology of the sink");
+ // the sink uid is prepended to the uid of every operator the sink expands
into, which is what
+ // keeps the expanded pipeline addressable across state restores.
+ assertTrue(writeNodes.get(0).getTransformationUID().startsWith(sinkUid +
": "),
+ "Expected the write operator uid to be prefixed with the sink uid " +
sinkUid
+ + " but got " + writeNodes.get(0).getTransformationUID());
+ }
+
@Test
void testServiceTopologiesUseExpectedSingletonOperatorsAndPartitioners()
throws Exception {
conf.set(FlinkOptions.CLUSTERING_TASKS, 3);
diff --git
a/hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/adapter/SupportsPreWriteTopologyAdapter.java
b/hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/adapter/SupportsPreWriteTopologyAdapter.java
index 45e552ed2231..f6cfe0aa83a4 100644
---
a/hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/adapter/SupportsPreWriteTopologyAdapter.java
+++
b/hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/adapter/SupportsPreWriteTopologyAdapter.java
@@ -18,20 +18,15 @@
package org.apache.hudi.adapter;
-import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.connector.sink2.WithPreWriteTopology;
/**
- * {@code SupportsPreWriteTopology} is introduced for Sink V2 since Flink 1.19,
- * We add the adapter here to just make the compilation successful for earlier
- * Flink versions (< 1.19).
+ * Adapter clazz for {@code SupportsPreWriteTopology}.
+ *
+ * <p>{@code SupportsPreWriteTopology} only exists since Flink 1.19, where it
was split out of
+ * {@link WithPreWriteTopology}; both declare the same {@code
addPreWriteTopology} method, and
+ * Flink 1.18 dispatches on {@link WithPreWriteTopology} in {@code
SinkTransformationTranslator}.
+ * Extending it here keeps the pre-write topology hook wired for Flink 1.18 as
well.
*/
-public interface SupportsPreWriteTopologyAdapter<InputT> {
- /**
- * Adds an arbitrary topology before the writer. The topology may be used to
repartition the
- * data.
- *
- * @param inputDataStream the stream of input records.
- * @return the custom topology before {@code SinkWriter}.
- */
- DataStream<InputT> addPreWriteTopology(DataStream<InputT> inputDataStream);
+public interface SupportsPreWriteTopologyAdapter<InputT> extends
WithPreWriteTopology<InputT> {
}
