ericyuan915 opened a new issue, #19966:
URL: https://github.com/apache/hudi/issues/19966
### Describe the problem you faced
**Summary**
`org.apache.hudi.sink.v2.HoodieSink` (Sink V2, added in #13423) is disabled
on Flink 1.18 by a runtime guard in `PipelinesV2.sink`:
```java
ValidationUtils.checkArgument(FlinkVersion.current().toString().compareTo(FLINK_1_18_VERSION)
> 0,
"Hudi sink v2 is not supported with Flink version: " +
FlinkVersion.current());
```
(`hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/v2/utils/PipelinesV2.java`,
L78-L79, added in #13610.)
The guard is correct as the code stands today, but the limitation behind it
looks avoidable. Flink 1.18 does have the pre-write-topology hook — it is
simply named `WithPreWriteTopology` instead of `SupportsPreWriteTopology`.
Having the 1.18 adapter extend it would make Sink V2 work on 1.18, and the
guard could then be relaxed.
Since master now carries `hudi-flink1.18.x` as the oldest Flink profile,
1.18 is the only supported version where Sink V2 is unavailable.
**Why it does not work today**
`HoodieSink implements SinkAdapter<RowData>,
SupportsPreWriteTopologyAdapter<RowData>` and builds the entire write pipeline
inside `addPreWriteTopology`; `createWriter()` returns a no-op
`DummySinkWriter`.
- `hudi-flink1.19.x/.../adapter/SupportsPreWriteTopologyAdapter` extends
Flink's real `SupportsPreWriteTopology`, so Flink invokes `addPreWriteTopology`.
- `hudi-flink1.18.x/.../adapter/SupportsPreWriteTopologyAdapter` extends
nothing. Its javadoc states the intent: *"We add the adapter here to just make
the compilation successful for earlier Flink versions (< 1.19)."*
Flink 1.18 dispatches on `WithPreWriteTopology`
(`SinkTransformationTranslator`, release-1.18, L138):
```java
if (sink instanceof WithPreWriteTopology) {
prewritten = adjustTransformations(
prewritten, ((WithPreWriteTopology<T>)
sink)::addPreWriteTopology, true, ...);
}
```
`HoodieSink` is not a `WithPreWriteTopology`, so on 1.18 the hook is never
called, the Hudi topology is never built, and only `DummySinkWriter` runs — the
job would silently write nothing. The guard correctly fences that off.
**Why 1.18 can support it**
Flink 1.18
`org.apache.flink.streaming.api.connector.sink2.WithPreWriteTopology`:
```java
@Experimental
public interface WithPreWriteTopology<InputT> extends Sink<InputT> {
DataStream<InputT> addPreWriteTopology(DataStream<InputT>
inputDataStream);
}
```
Flink 1.19 `SupportsPreWriteTopology` declares the identical method. The
1.19 change split the capability out of `Sink` and deprecated
`WithPreWriteTopology` (which in 1.19 extends both `Sink` and
`SupportsPreWriteTopology`). So the capability is present on 1.18; only the
type name and its `Sink` supertype differ.
**Proposal**
1. In `hudi-flink1.18.x`, make the adapter extend the real Flink interface:
```java
public interface SupportsPreWriteTopologyAdapter<InputT>
extends
org.apache.flink.streaming.api.connector.sink2.WithPreWriteTopology<InputT> {
}
```
`HoodieSink` then reaches `Sink<RowData>` through both `SinkAdapter` and
`WithPreWriteTopology` — legal, since it is the same type argument — and
inherits `addPreWriteTopology` instead of redeclaring it.
2. Relax the version guard in `PipelinesV2.sink` accordingly (or drop it, if
1.18 becomes the supported floor).
3. `ITTestDataStreamV2Write` already exercises `PipelinesV2.sink` and
carries no version gating; the default profile on master is `flink2.2`, so
running that IT under the `flink1.18` profile would cover the change.
**Open questions**
- Does `SinkTransformationTranslator.adjustTransformations` on 1.18 preserve
the operator UIDs that `Pipelines.*` assigns via `opUID(...)` the same way 1.19
does? If it differs, savepoint/checkpoint compatibility is worth validating
before advertising 1.18 support.
- Was the guard a deliberate "we do not want to support Sink V2 below 1.19"
policy decision rather than a technical one? It arrived as a one-line rider
inside the #13610 adapter cleanup, so I may be missing context — hence filing
this as a question before a patch.
Happy to put up the PR if the direction sounds reasonable.
### To Reproduce
N/A
Code issue
### Expected behavior
N/A
Can 1.18 work with Hudi sink v2?
### Environment Description
- Hudi: master (behaviour also present on the 1.2.x line)
- Flink: 1.18.1
### Additional context
### Task Type
Improvement
### Related Issues
- #13423 — [HUDI-9520] Support new sink based on Flink sink V2 API
(introduced `HoodieSink` / `PipelinesV2`)
- #13610 — [HUDI-9617] Clean redundant adapters for supporting multiple
Flink versions (introduced the version guard)
### Stacktrace
```shell
```
--
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]
