[lake/flink] Introduce committer operator

This closes #990.

Author: luoyuxia

fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/tiering/committer/CommittableMessage.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright (c) 2025 Alibaba Group Holding Ltd.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.alibaba.fluss.flink.tiering.committer;
+
+import java.io.Serializable;
+
+/** A class wrapping {@link Committable} to commit to lake. */
+public class CommittableMessage<Committable> implements Serializable {
+    private static final long serialVersionUID = 1L;
+
+    private final Committable committable;
+
+    public CommittableMessage(Committable committable) {
+        this.committable = committable;
+    }
+
+    public Committable committable() {
+        return committable;
+    }
+}

fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/tiering/committer/CommittableMessageTypeInfo.java

@@ -0,0 +1,143 @@
+/*
+ * Copyright (c) 2025 Alibaba Group Holding Ltd.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.alibaba.fluss.flink.tiering.committer;
+
+import com.alibaba.fluss.flink.tiering.source.TableBucketWriteResult;
+import com.alibaba.fluss.lakehouse.serializer.SimpleVersionedSerializer;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.io.SimpleVersionedSerializerTypeSerializerProxy;
+import org.apache.flink.util.function.SerializableSupplier;
+
+import java.io.IOException;
+
+/** A {@link TypeInformation} for {@link CommittableMessage}. */
+public class CommittableMessageTypeInfo<Committable>
+        extends TypeInformation<CommittableMessage<Committable>> {
+
+    private final SerializableSupplier<SimpleVersionedSerializer<Committable>>
+            committableSerializerFactory;
+
+    private CommittableMessageTypeInfo(
+            SerializableSupplier<SimpleVersionedSerializer<Committable>>
+                    committableSerializerFactory) {
+        this.committableSerializerFactory = committableSerializerFactory;
+    }
+
+    public static <Committable> TypeInformation<CommittableMessage<Committable>> of(
+            SerializableSupplier<SimpleVersionedSerializer<Committable>>
+                    committableSerializerFactory) {
+        return new CommittableMessageTypeInfo<>(committableSerializerFactory);
+    }
+
+    @Override
+    public boolean isBasicType() {
+        return false;
+    }
+
+    @Override
+    public boolean isTupleType() {
+        return false;
+    }
+
+    @Override
+    public int getArity() {
+        return 1;
+    }
+
+    @Override
+    public int getTotalFields() {
+        return 1;
+    }
+
+    @SuppressWarnings({"unchecked", "rawtypes"})
+    @Override
+    public Class<CommittableMessage<Committable>> getTypeClass() {
+        return (Class) TableBucketWriteResult.class;
+    }
+
+    @Override
+    public boolean isKeyType() {
+        return false;
+    }
+
+    @Override
+    public TypeSerializer<CommittableMessage<Committable>> createSerializer(
+            ExecutionConfig executionConfig) {
+        // no copy, so that data from writer is directly going into upstream operator while chaining
+        SimpleVersionedSerializer<Committable> committableSerializer =
+                committableSerializerFactory.get();
+        return new SimpleVersionedSerializerTypeSerializerProxy<CommittableMessage<Committable>>(
+                () ->
+                        new org.apache.flink.core.io.SimpleVersionedSerializer<
+                                CommittableMessage<Committable>>() {
+                            @Override
+                            public int getVersion() {
+                                return committableSerializer.getVersion();
+                            }
+
+                            @Override
+                            public byte[] serialize(
+                                    CommittableMessage<Committable> committableCommittableMessage)
+                                    throws IOException {
+                                return committableSerializer.serialize(
+                                        committableCommittableMessage.committable());
+                            }
+
+                            @Override
+                            public CommittableMessage<Committable> deserialize(
+                                    int version, byte[] serialized) throws IOException {
+                                return new CommittableMessage<>(
+                                        committableSerializer.deserialize(version, serialized));
+                            }
+                        }) {
+            // nothing
+            @Override
+            public CommittableMessage<Committable> copy(CommittableMessage<Committable> from) {
+                return from;
+            }
+
+            @Override
+            public CommittableMessage<Committable> copy(
+                    CommittableMessage<Committable> from, CommittableMessage<Committable> reuse) {
+                return from;
+            }
+        };
+    }
+
+    @Override
+    public String toString() {
+        return "LakeCommittableTypeInfo";
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        return obj instanceof CommittableMessageTypeInfo;
+    }
+
+    @Override
+    public int hashCode() {
+        return getClass().hashCode();
+    }
+
+    @Override
+    public boolean canEqual(Object obj) {
+        return obj instanceof CommittableMessageTypeInfo;
+    }
+}

fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/tiering/committer/FlussTableLakeSnapshot.java

@@ -0,0 +1,62 @@
+/*
+ * Copyright (c) 2025 Alibaba Group Holding Ltd.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.alibaba.fluss.flink.tiering.committer;
+
+import com.alibaba.fluss.metadata.TableBucket;
+
+import java.util.Map;
+
+/** A lake snapshot for a Fluss table. */
+public class FlussTableLakeSnapshot {
+
+    private final long tableId;
+
+    private final long lakeSnapshotId;
+
+    private final Map<TableBucket, Long> logEndOffsets;
+
+    public FlussTableLakeSnapshot(
+            long tableId, long lakeSnapshotId, Map<TableBucket, Long> logEndOffsets) {
+        this.tableId = tableId;
+        this.lakeSnapshotId = lakeSnapshotId;
+        this.logEndOffsets = logEndOffsets;
+    }
+
+    public long tableId() {
+        return tableId;
+    }
+
+    public long lakeSnapshotId() {
+        return lakeSnapshotId;
+    }
+
+    public Map<TableBucket, Long> logEndOffsets() {
+        return logEndOffsets;
+    }
+
+    @Override
+    public String toString() {
+        return "FlussTableLakeSnapshot{"
+                + "tableId="
+                + tableId
+                + ", lakeSnapshotId="
+                + lakeSnapshotId
+                + ", logEndOffsets="
+                + logEndOffsets
+                + '}';
+    }
+}

fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/tiering/committer/FlussTableLakeSnapshotCommitter.java

@@ -0,0 +1,104 @@
+/*
+ * Copyright (c) 2025 Alibaba Group Holding Ltd.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.alibaba.fluss.flink.tiering.committer;
+
+import com.alibaba.fluss.client.metadata.MetadataUpdater;
+import com.alibaba.fluss.config.ConfigOptions;
+import com.alibaba.fluss.config.Configuration;
+import com.alibaba.fluss.metadata.TableBucket;
+import com.alibaba.fluss.metrics.registry.MetricRegistry;
+import com.alibaba.fluss.rpc.GatewayClientProxy;
+import com.alibaba.fluss.rpc.RpcClient;
+import com.alibaba.fluss.rpc.gateway.CoordinatorGateway;
+import com.alibaba.fluss.rpc.messages.CommitLakeTableSnapshotRequest;
+import com.alibaba.fluss.rpc.messages.PbLakeTableOffsetForBucket;
+import com.alibaba.fluss.rpc.messages.PbLakeTableSnapshotInfo;
+import com.alibaba.fluss.rpc.metrics.ClientMetricGroup;
+import com.alibaba.fluss.utils.ExceptionUtils;
+
+import java.io.IOException;
+import java.util.Map;
+
+/** Committer to commit {@link FlussTableLakeSnapshot} of lake to Fluss. */
+public class FlussTableLakeSnapshotCommitter implements AutoCloseable {
+
+    private final Configuration flussConf;
+
+    private CoordinatorGateway coordinatorGateway;
+    private RpcClient rpcClient;
+
+    public FlussTableLakeSnapshotCommitter(Configuration flussConf) {
+        this.flussConf = flussConf;
+    }
+
+    public void open() {
+        // init coordinator gateway
+        String clientId = flussConf.getString(ConfigOptions.CLIENT_ID);
+        MetricRegistry metricRegistry = MetricRegistry.create(flussConf, null);
+        // don't care about metrics, but pass a ClientMetricGroup to make compiler happy
+        rpcClient = RpcClient.create(flussConf, new ClientMetricGroup(metricRegistry, clientId));
+        MetadataUpdater metadataUpdater = new MetadataUpdater(flussConf, rpcClient);
+        this.coordinatorGateway =
+                GatewayClientProxy.createGatewayProxy(
+                        metadataUpdater::getCoordinatorServer, rpcClient, CoordinatorGateway.class);
+    }
+
+    public void commit(FlussTableLakeSnapshot flussTableLakeSnapshot) throws IOException {
+        try {
+            CommitLakeTableSnapshotRequest request =
+                    toCommitLakeTableSnapshotRequest(flussTableLakeSnapshot);
+            coordinatorGateway.commitLakeTableSnapshot(request).get();
+        } catch (Exception e) {
+            throw new IOException(
+                    String.format(
+                            "Fail to commit table lake snapshot %s to Fluss.",
+                            flussTableLakeSnapshot),
+                    ExceptionUtils.stripExecutionException(e));
+        }
+    }
+
+    private CommitLakeTableSnapshotRequest toCommitLakeTableSnapshotRequest(
+            FlussTableLakeSnapshot flussTableLakeSnapshot) {
+        CommitLakeTableSnapshotRequest commitLakeTableSnapshotRequest =
+                new CommitLakeTableSnapshotRequest();
+        PbLakeTableSnapshotInfo pbLakeTableSnapshotInfo =
+                commitLakeTableSnapshotRequest.addTablesReq();
+
+        pbLakeTableSnapshotInfo.setTableId(flussTableLakeSnapshot.tableId());
+        pbLakeTableSnapshotInfo.setSnapshotId(flussTableLakeSnapshot.lakeSnapshotId());
+        for (Map.Entry<TableBucket, Long> bucketEndOffsetEntry :
+                flussTableLakeSnapshot.logEndOffsets().entrySet()) {
+            PbLakeTableOffsetForBucket pbLakeTableOffsetForBucket =
+                    pbLakeTableSnapshotInfo.addBucketsReq();
+            TableBucket tableBucket = bucketEndOffsetEntry.getKey();
+            long endOffset = bucketEndOffsetEntry.getValue();
+            if (tableBucket.getPartitionId() != null) {
+                pbLakeTableOffsetForBucket.setPartitionId(tableBucket.getPartitionId());
+            }
+            pbLakeTableOffsetForBucket.setBucketId(tableBucket.getBucket());
+            pbLakeTableOffsetForBucket.setLogEndOffset(endOffset);
+        }
+        return commitLakeTableSnapshotRequest;
+    }
+
+    @Override
+    public void close() throws Exception {
+        if (rpcClient != null) {
+            rpcClient.close();
+        }
+    }
+}