diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/LocalMergeOperator.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/LocalMergeOperator.java index 4852171a6442..ac182593cc4e 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/LocalMergeOperator.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/LocalMergeOperator.java @@ -201,13 +201,12 @@ public void close() throws Exception { } private void flushBuffer() throws Exception { - if (merger.isEmpty()) { - return; + if (!merger.isEmpty()) { + merger.forEach(row -> output.collect(new StreamRecord<>(row))); + merger.clear(); } - merger.forEach(row -> output.collect(new StreamRecord<>(row))); - merger.clear(); - + // emit the watermark even when no rows are buffered, e.g. when all rows are filtered out if (currentWatermark != Long.MIN_VALUE) { super.processWatermark(new Watermark(currentWatermark)); // each watermark should only be emitted once diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/LocalMergeWatermarkITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/LocalMergeWatermarkITCase.java new file mode 100644 index 000000000000..f6b8e990a352 --- /dev/null +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/LocalMergeWatermarkITCase.java @@ -0,0 +1,128 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.paimon.flink; + +import org.apache.paimon.Snapshot; +import org.apache.paimon.flink.sink.FlinkSinkBuilder; +import org.apache.paimon.flink.source.AbstractNonCoordinatedSource; +import org.apache.paimon.flink.source.AbstractNonCoordinatedSourceReader; +import org.apache.paimon.flink.source.SimpleSourceSplit; +import org.apache.paimon.table.FileStoreTable; + +import org.apache.flink.api.common.eventtime.Watermark; +import org.apache.flink.api.common.eventtime.WatermarkStrategy; +import org.apache.flink.api.connector.source.Boundedness; +import org.apache.flink.api.connector.source.ReaderOutput; +import org.apache.flink.api.connector.source.SourceReader; +import org.apache.flink.api.connector.source.SourceReaderContext; +import org.apache.flink.core.execution.JobClient; +import org.apache.flink.core.io.InputStatus; +import org.apache.flink.streaming.api.datastream.DataStream; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.table.data.GenericRowData; +import org.apache.flink.table.data.RowData; +import org.junit.jupiter.api.Test; + +import java.util.Collections; +import java.util.List; + +/** IT cases for the watermark of a sink with local merge. */ +public class LocalMergeWatermarkITCase extends CatalogITCaseBase { + + @Override + protected List ddl() { + return Collections.singletonList( + "CREATE TABLE T (k INT, v INT, PRIMARY KEY (k) NOT ENFORCED) WITH (" + + "'bucket' = '1', 'local-merge-buffer-size' = '5m', " + + "'commit.force-create-snapshot' = 'true')"); + } + + @Test + public void testWatermarkAdvancesWhileRecordsAreFilteredOut() throws Exception { + FileStoreTable table = paimonTable("T"); + StreamExecutionEnvironment env = + streamExecutionEnvironmentBuilder() + .streamingMode() + .parallelism(1) + .checkpointIntervalMs(200) + .build(); + DataStream source = + env.fromSource( + new IncreasingWatermarkSource(), + WatermarkStrategy.noWatermarks(), + "increasing-watermark-source"); + // like a WHERE clause: only the first record reaches the sink, the watermark keeps moving + new FlinkSinkBuilder(table).forRowData(source.filter(row -> row.getInt(0) == 0)).build(); + + // the job is cancelled by AbstractTestBase#cleanupRunningJobs + JobClient client = env.executeAsync(); + long first = waitForWatermark(client, table, Long.MIN_VALUE); + waitForWatermark(client, table, first); + } + + /** Waits for a snapshot whose watermark is greater than {@code previous}, fails on timeout. */ + private static long waitForWatermark(JobClient client, FileStoreTable table, long previous) + throws Exception { + long deadline = System.currentTimeMillis() + 60_000; + while (System.currentTimeMillis() < deadline) { + if (client.getJobStatus().get().isGloballyTerminalState()) { + // surfaces the failure cause + client.getJobExecutionResult().get(); + } + Snapshot snapshot = table.snapshotManager().latestSnapshot(); + if (snapshot != null + && snapshot.watermark() != null + && snapshot.watermark() > previous) { + return snapshot.watermark(); + } + Thread.sleep(100); + } + throw new AssertionError( + String.format("No snapshot watermark greater than %d within 60 s", previous)); + } + + /** Emits a record with an increasing watermark every 100 ms, forever. */ + private static class IncreasingWatermarkSource extends AbstractNonCoordinatedSource { + + private static final long serialVersionUID = 1L; + + @Override + public Boundedness getBoundedness() { + return Boundedness.CONTINUOUS_UNBOUNDED; + } + + @Override + public SourceReader createReader(SourceReaderContext context) { + return new AbstractNonCoordinatedSourceReader() { + + private int step; + + @Override + public InputStatus pollNext(ReaderOutput output) + throws InterruptedException { + output.collect(GenericRowData.of(step, step)); + output.emitWatermark(new Watermark(step * 1000L)); + step++; + Thread.sleep(100); + return InputStatus.MORE_AVAILABLE; + } + }; + } + } +} diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/LocalMergeOperatorTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/LocalMergeOperatorTest.java index eba09e12e59d..353b85bb7c67 100644 --- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/LocalMergeOperatorTest.java +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/LocalMergeOperatorTest.java @@ -100,6 +100,29 @@ public void testHashNormal() throws Exception { result.clear(); } + @Test + public void testEmptyFlushStillEmitsWatermark() throws Exception { + prepareHashOperator(); + List watermarks = new ArrayList<>(); + List rows = new ArrayList<>(); + operator.setOutput( + new TestOutput(row -> rows.add(row.toString())) { + @Override + public void emitWatermark(Watermark mark) { + watermarks.add(mark.getTimestamp()); + } + }); + + // a watermark arrives but no records do + operator.processWatermark(new Watermark(42L)); + assertThat(operator.merger().isEmpty()).isTrue(); + + operator.prepareSnapshotPreBarrier(0); + + assertThat(rows).isEmpty(); + assertThat(watermarks).containsExactly(42L); + } + @Test public void testUserDefineSequence() throws Exception { Map options = new HashMap<>();