Hey folks! I'm working on migrating the Delta Lake connector from Flink 1.16 to 2.0 and hit an interesting blocker that I'd love some guidance on. Quick context: I've made substantial progress (15 commits, 105+ tests migrated, everything compiles), but there's one missing piece that's blocking integration tests. The Problem: My custom

DeltaSinkInternal

implements

Sink<IN>

and has these methods:

public Optional<Committer<T>> createCommitter() throws IOException
public Optional<SimpleVersionedSerializer<T>> getCommittableSerializer()

The framework discovers and invokes

createWriter()

perfectly, but

createCommitter()

is never called. When I tried adding

@Override

, I got a compilation error saying these methods aren't part of the

Sink<IN>

interface. The Question: How does a Sink expose its Committer in Flink 2.0? Is there a marker interface I'm missing? How does

FileSink

do it? I suspect there's a specific pattern or interface (maybe something that replaced

TwoPhaseCommittingSink

from 1.x) that I need to implement, but I can't find documentation about it. Related: • Delta Lake issue: https://github.com/delta-io/delta/issues/5228 • My implementation branch: https://github.com/kdsnunes/delta/tree/flink/add-flink-2.0-support • (I'm also trying to get access to Flink's Jira to create an issue there, but it's pending approval) Any pointers would be super appreciated! Has anyone migrated a connector with Committer support to 2.0? Thanks in advance! 🙏

With 2.0 you need to use the SupportXXX interfaces, in particular

SupportsCommitter

@Arvid Heise Thanks to your guidance on using SupportsCommitter<T>, I've successfully resolved the Committer discovery issue in the Delta Lake connector migration to Flink 2.0

Global committer is a bit more hidden in V2:

private static class TestSinkV2WithPostCommitTopology<InputT>
            extends TestSinkV2TwoPhaseCommittingSink<InputT>
            implements SupportsPostCommitTopology<String> {
        public TestSinkV2WithPostCommitTopology(
                DefaultSinkWriter<InputT> writer,
                SimpleVersionedSerializer<String> committableSerializer,
                DefaultCommitter committer) {
            super(writer, committableSerializer, committer);
        }

        @Override
        public void addPostCommitTopology(DataStream<CommittableMessage<String>> committables) {
            StandardSinkTopologies.addGlobalCommitter(
                    committables, DefaultCommitter::new, this::getCommittableSerializer);
        }
    }

So you add a post commit topology and can do whatever you want, in particular you can use something that simulates the old global committer.

@Arvid Heise Thank you so much for pointing me to

SupportsPostCommitTopology

and

StandardSinkTopologies.addGlobalCommitter

! That was exactly what I needed. I've successfully implemented the pattern and the global coordination is now working correctly - committables from all parallel DeltaCommitter instances are being aggregated and committed atomically to the Delta Log through the post-commit topology. The migration from GlobalCommitter to this new pattern is much clearer now. Really appreciate the help! 🙏

@K Nunes you'll be a hero in my books if you get Flink 2 support going 🙂