This is an automated email from the ASF dual-hosted git repository.
1996fanrui pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 3593f080f09 [FLINK-40517][checkpointing] Aggregate cross-channel
watermarks on the recovery unspilling thread for fan-in rescale
3593f080f09 is described below
commit 3593f080f09fb9f974dca62279a98cc16505acb9
Author: Rui Fan <[email protected]>
AuthorDate: Mon Aug 31 17:44:43 2026 +0200
[FLINK-40517][checkpointing] Aggregate cross-channel watermarks on the
recovery unspilling thread for fan-in rescale
---
.../channel/ChannelStateFilteringHandler.java | 108 ++++++++++++++-
.../GateFilterHandlerBufferOwnershipTest.java | 4 +-
.../checkpoint/channel/GateFilterHandlerTest.java | 152 +++++++++++++++++++--
.../InputChannelRecoveredStateHandlerTest.java | 3 +-
4 files changed, 251 insertions(+), 16 deletions(-)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateFilteringHandler.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateFilteringHandler.java
index 0b6976068d7..be9c91e4955 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateFilteringHandler.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateFilteringHandler.java
@@ -30,12 +30,14 @@ import org.apache.flink.runtime.io.network.buffer.Buffer;
import org.apache.flink.runtime.io.network.partition.consumer.InputGate;
import org.apache.flink.runtime.plugable.DeserializationDelegate;
import org.apache.flink.runtime.plugable.NonReusingDeserializationDelegate;
+import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.runtime.io.recovery.RecordFilter;
import org.apache.flink.streaming.runtime.io.recovery.RecordFilterContext;
import org.apache.flink.streaming.runtime.io.recovery.VirtualChannel;
import
org.apache.flink.streaming.runtime.io.recovery.VirtualChannelRecordFilterFactory;
import org.apache.flink.streaming.runtime.streamrecord.StreamElement;
import org.apache.flink.streaming.runtime.streamrecord.StreamElementSerializer;
+import org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus;
import javax.annotation.Nullable;
@@ -47,6 +49,7 @@ import java.util.List;
import java.util.Map;
import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkState;
/**
* Filters recovered channel state buffers during the channel-state-unspilling
phase, removing
@@ -221,7 +224,47 @@ public class ChannelStateFilteringHandler implements
Closeable {
return null;
}
- return new GateFilterHandler<>(gateVirtualChannels, elementSerializer);
+ return new GateFilterHandler<>(gateVirtualChannels, elementSerializer,
channelMapping);
+ }
+
+ /**
+ * Maps each old-channel key to the virtual channels that fold into its
new channel, so
+ * watermarks/statuses can be aggregated across old channels merged on a
fan-in rescale. Without
+ * a rescale each group is a singleton and aggregation is verbatim
pass-through.
+ */
+ private static <T>
+ Map<SubtaskConnectionDescriptor, List<VirtualChannel<T>>>
buildWatermarkMergeGroups(
+ Map<SubtaskConnectionDescriptor, VirtualChannel<T>>
gateVirtualChannels,
+ RescaleMappings channelMapping) {
+ RescaleMappings oldToNewMapping = channelMapping.invert();
+ Map<Integer, List<VirtualChannel<T>>> byNewChannel = new HashMap<>();
+ Map<SubtaskConnectionDescriptor, List<VirtualChannel<T>>> mergeGroups
= new HashMap<>();
+ gateVirtualChannels.forEach(
+ (key, vc) -> {
+ List<VirtualChannel<T>> group =
+ byNewChannel.computeIfAbsent(
+ newChannelIndexOf(key, oldToNewMapping),
+ idx -> new ArrayList<>());
+ group.add(vc);
+ mergeGroups.put(key, group);
+ });
+ return mergeGroups;
+ }
+
+ /**
+ * Resolves the new channel index the given old-channel descriptor is
routed to. Its {@code
+ * outputSubtaskIndex} field carries the old channel index.
+ */
+ private static int newChannelIndexOf(
+ SubtaskConnectionDescriptor oldChannelKey, RescaleMappings
oldToNewMapping) {
+ int oldChannelIndex = oldChannelKey.getOutputSubtaskIndex();
+ int[] mapped = oldToNewMapping.getMappedIndexes(oldChannelIndex);
+ checkState(
+ mapped.length == 1,
+ "One old channel is expected to fold into exactly one new
channel, but %s mapped to %s new channels.",
+ oldChannelKey,
+ mapped.length);
+ return mapped[0];
}
/**
@@ -252,14 +295,24 @@ public class ChannelStateFilteringHandler implements
Closeable {
static class GateFilterHandler<T> {
private final Map<SubtaskConnectionDescriptor, VirtualChannel<T>>
virtualChannels;
+
+ /**
+ * For each old-channel key, the virtual channels folding into the
same new channel, used to
+ * aggregate watermarks/statuses across old channels merged on a
fan-in rescale.
+ */
+ private final Map<SubtaskConnectionDescriptor, List<VirtualChannel<T>>>
+ watermarkMergeGroups;
+
private final StreamElementSerializer<T> serializer;
private final DeserializationDelegate<StreamElement>
deserializationDelegate;
GateFilterHandler(
Map<SubtaskConnectionDescriptor, VirtualChannel<T>>
virtualChannels,
- StreamElementSerializer<T> serializer) {
+ StreamElementSerializer<T> serializer,
+ RescaleMappings channelMapping) {
this.virtualChannels = checkNotNull(virtualChannels);
this.serializer = checkNotNull(serializer);
+ this.watermarkMergeGroups =
buildWatermarkMergeGroups(virtualChannels, channelMapping);
this.deserializationDelegate = new
NonReusingDeserializationDelegate<>(serializer);
}
@@ -288,13 +341,21 @@ public class ChannelStateFilteringHandler implements
Closeable {
+ virtualChannels.keySet());
}
+ List<VirtualChannel<T>> mergeGroup =
watermarkMergeGroups.get(key);
+ checkNotNull(mergeGroup, "No watermark merge group for key:
%s", key);
+
vc.setNextBuffer(sourceBuffer);
sourceBufferOwnershipTransferred = true;
while (true) {
DeserializationResult result =
vc.getNextRecord(deserializationDelegate);
if (result.isFullRecord()) {
-
serializeElement(deserializationDelegate.getInstance(), outputSerializer);
+ // vc.getNextRecord has already updated the source
channel's lastWatermark /
+ // watermarkStatus, so aggregation below reads the
up-to-date group state.
+ emitAggregated(
+ deserializationDelegate.getInstance(),
+ mergeGroup,
+ outputSerializer);
}
if (result.isBufferConsumed()) {
break;
@@ -308,6 +369,47 @@ public class ChannelStateFilteringHandler implements
Closeable {
}
}
+ /**
+ * Writes one element, aggregating non-records across the {@code
mergeGroup}: emit the group
+ * min watermark (suppressed until every merged channel has one), and
{@code ACTIVE} status
+ * while any merged channel is active. Records and latency markers
pass through verbatim.
+ */
+ private void emitAggregated(
+ StreamElement element,
+ List<VirtualChannel<T>> mergeGroup,
+ DataOutputSerializer outputSerializer)
+ throws IOException {
+ if (element.isWatermark()) {
+ Watermark minWatermark = null;
+ for (VirtualChannel<T> channel : mergeGroup) {
+ Watermark candidate = channel.getLastWatermark();
+ if (minWatermark == null
+ || candidate.getTimestamp() <
minWatermark.getTimestamp()) {
+ minWatermark = candidate;
+ }
+ }
+ checkState(minWatermark != null, "Should always have a
watermark");
+ // min == UNINITIALIZED only when some merged old channel has
no watermark yet;
+ // hold the group's watermark back until every one of them has
produced one.
+ if (!minWatermark.equals(Watermark.UNINITIALIZED)) {
+ serializeElement(minWatermark, outputSerializer);
+ }
+ } else if (element.isWatermarkStatus()) {
+ boolean anyActive = false;
+ for (VirtualChannel<T> channel : mergeGroup) {
+ if (channel.getWatermarkStatus().isActive()) {
+ anyActive = true;
+ break;
+ }
+ }
+ serializeElement(
+ anyActive ? WatermarkStatus.ACTIVE :
element.asWatermarkStatus(),
+ outputSerializer);
+ } else {
+ serializeElement(element, outputSerializer);
+ }
+ }
+
/**
* Appends one stream element as a length-prefixed record. Reserves
the 4B prefix,
* serializes the element, then backfills the length, because {@code
outputSerializer}
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/GateFilterHandlerBufferOwnershipTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/GateFilterHandlerBufferOwnershipTest.java
index ae7b722f683..206ed08e204 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/GateFilterHandlerBufferOwnershipTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/GateFilterHandlerBufferOwnershipTest.java
@@ -21,6 +21,7 @@ import
org.apache.flink.api.common.typeutils.base.LongSerializer;
import org.apache.flink.core.memory.DataOutputSerializer;
import org.apache.flink.core.memory.MemorySegment;
import org.apache.flink.core.memory.MemorySegmentFactory;
+import org.apache.flink.runtime.checkpoint.RescaleMappings;
import org.apache.flink.runtime.io.network.api.SubtaskConnectionDescriptor;
import
org.apache.flink.runtime.io.network.api.serialization.RecordDeserializer;
import
org.apache.flink.runtime.io.network.api.serialization.SpillingAdaptiveSpanningRecordDeserializer;
@@ -142,7 +143,8 @@ class GateFilterHandlerBufferOwnershipTest {
StreamElementSerializer<Long> serializer =
new StreamElementSerializer<>(LongSerializer.INSTANCE);
- return new ChannelStateFilteringHandler.GateFilterHandler<>(channels,
serializer);
+ return new ChannelStateFilteringHandler.GateFilterHandler<>(
+ channels, serializer, RescaleMappings.SYMMETRIC_IDENTITY);
}
private Buffer createBufferWithRecords(Long... values) {
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/GateFilterHandlerTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/GateFilterHandlerTest.java
index 1646908727a..44875cf86f0 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/GateFilterHandlerTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/GateFilterHandlerTest.java
@@ -21,6 +21,7 @@ import
org.apache.flink.api.common.typeutils.base.LongSerializer;
import org.apache.flink.core.memory.DataOutputSerializer;
import org.apache.flink.core.memory.MemorySegment;
import org.apache.flink.core.memory.MemorySegmentFactory;
+import org.apache.flink.runtime.checkpoint.RescaleMappings;
import org.apache.flink.runtime.io.network.api.SubtaskConnectionDescriptor;
import
org.apache.flink.runtime.io.network.api.serialization.RecordDeserializer;
import
org.apache.flink.runtime.io.network.api.serialization.SpillingAdaptiveSpanningRecordDeserializer;
@@ -29,19 +30,23 @@ import
org.apache.flink.runtime.io.network.buffer.FreeingBufferRecycler;
import org.apache.flink.runtime.io.network.buffer.NetworkBuffer;
import org.apache.flink.runtime.plugable.DeserializationDelegate;
import org.apache.flink.runtime.plugable.NonReusingDeserializationDelegate;
+import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.runtime.io.recovery.RecordFilter;
import org.apache.flink.streaming.runtime.io.recovery.VirtualChannel;
import org.apache.flink.streaming.runtime.streamrecord.StreamElement;
import org.apache.flink.streaming.runtime.streamrecord.StreamElementSerializer;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus;
import org.junit.jupiter.api.Test;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.stream.Stream;
import static org.assertj.core.api.Assertions.assertThat;
@@ -129,33 +134,104 @@ class GateFilterHandlerTest {
assertThat(sourceBuffer.isRecycled()).isTrue();
}
+ /**
+ * On a fan-in rescale two old channels A and B fold into one new channel:
A's high watermark is
+ * held while B has none, then the merged output carries the group
minimum, not A's higher one.
+ */
+ @Test
+ void testFanInWatermarkMinMergeAndHold() throws Exception {
+ SubtaskConnectionDescriptor keyA = new SubtaskConnectionDescriptor(0,
0);
+ SubtaskConnectionDescriptor keyB = new SubtaskConnectionDescriptor(0,
1);
+ ChannelStateFilteringHandler.GateFilterHandler<Long> handler =
+ createMergedHandler(keyA, keyB);
+
+ DataOutputSerializer output = new DataOutputSerializer(BUFFER_SIZE);
+
+ // A delivers a record then a high watermark; B has not produced a
watermark yet, so the
+ // watermark must be suppressed at this point.
+ handler.filterAndRewrite(
+ 0,
+ 0,
+ createBufferWithElements(new StreamRecord<>(1L), new
Watermark(100L)),
+ output);
+ // B delivers a record and a lower watermark; now the group can emit
min(100, 30) = 30.
+ handler.filterAndRewrite(
+ 0, 1, createBufferWithElements(new StreamRecord<>(2L), new
Watermark(30L)), output);
+
+ List<StreamElement> elements = readElementsFromSerializer(output);
+
+ // Both records pass through the filter.
+ assertThat(elements).filteredOn(StreamElement::isRecord).hasSize(2);
+ // The only watermark emitted is the min (30) — the premature 100 is
never written.
+ assertThat(elements)
+ .filteredOn(StreamElement::isWatermark)
+ .extracting(e -> e.asWatermark().getTimestamp())
+ .containsExactly(30L);
+ }
+
+ /**
+ * An {@code IDLE} watermark status from a single merged old channel must
not idle the whole new
+ * channel while another merged old channel is still active: the
aggregation emits {@code
+ * ACTIVE}.
+ */
+ @Test
+ void testFanInWatermarkStatusAnyActive() throws Exception {
+ SubtaskConnectionDescriptor keyA = new SubtaskConnectionDescriptor(0,
0);
+ SubtaskConnectionDescriptor keyB = new SubtaskConnectionDescriptor(0,
1);
+ ChannelStateFilteringHandler.GateFilterHandler<Long> handler =
+ createMergedHandler(keyA, keyB);
+
+ DataOutputSerializer output = new DataOutputSerializer(BUFFER_SIZE);
+
+ // A goes idle, but B is still active (default), so the merged status
stays ACTIVE.
+ handler.filterAndRewrite(0, 0,
createBufferWithElements(WatermarkStatus.IDLE), output);
+
+ List<StreamElement> elements = readElementsFromSerializer(output);
+ assertThat(elements).hasSize(1);
+ assertThat(elements.get(0).isWatermarkStatus()).isTrue();
+ assertThat(elements.get(0).asWatermarkStatus().isActive()).isTrue();
+ }
+
//
-------------------------------------------------------------------------------------------
// Helper methods
//
-------------------------------------------------------------------------------------------
- private ChannelStateFilteringHandler.GateFilterHandler<Long> createHandler(
- RecordFilter<Long> filter) {
- RecordDeserializer<DeserializationDelegate<StreamElement>>
deserializer =
- new SpillingAdaptiveSpanningRecordDeserializer<>(
- new String[] {System.getProperty("java.io.tmpdir")});
- VirtualChannel<Long> vc = new VirtualChannel<>(deserializer, filter);
+ /**
+ * Builds a handler whose two old channels {@code keyA} / {@code keyB}
fold into a single new
+ * channel (they share one watermark merge group), mirroring a fan-in
rescale.
+ */
+ private ChannelStateFilteringHandler.GateFilterHandler<Long>
createMergedHandler(
+ SubtaskConnectionDescriptor keyA, SubtaskConnectionDescriptor
keyB) {
+ VirtualChannel<Long> vcA = newVirtualChannel();
+ VirtualChannel<Long> vcB = newVirtualChannel();
Map<SubtaskConnectionDescriptor, VirtualChannel<Long>> channels = new
HashMap<>();
- channels.put(KEY, vc);
+ channels.put(keyA, vcA);
+ channels.put(keyB, vcB);
StreamElementSerializer<Long> serializer =
new StreamElementSerializer<>(LongSerializer.INSTANCE);
- return new ChannelStateFilteringHandler.GateFilterHandler<>(channels,
serializer);
+ // new channel 0 folds old channels {0, 1}, so both keys share one
watermark merge group.
+ RescaleMappings channelMapping = RescaleMappings.of(Stream.of(new
int[] {0, 1}), 2);
+ return new ChannelStateFilteringHandler.GateFilterHandler<>(
+ channels, serializer, channelMapping);
}
- private Buffer createBufferWithRecords(Long... values) throws IOException {
+ private VirtualChannel<Long> newVirtualChannel() {
+ RecordDeserializer<DeserializationDelegate<StreamElement>>
deserializer =
+ new SpillingAdaptiveSpanningRecordDeserializer<>(
+ new String[] {System.getProperty("java.io.tmpdir")});
+ return new VirtualChannel<>(deserializer, RecordFilter.acceptAll());
+ }
+
+ private Buffer createBufferWithElements(StreamElement... elements) throws
IOException {
StreamElementSerializer<Long> serializer =
new StreamElementSerializer<>(LongSerializer.INSTANCE);
DataOutputSerializer output = new DataOutputSerializer(BUFFER_SIZE);
- for (Long value : values) {
+ for (StreamElement element : elements) {
DataOutputSerializer recordOutput = new DataOutputSerializer(64);
- serializer.serialize(new StreamRecord<>(value), recordOutput);
+ serializer.serialize(element, recordOutput);
int recordLength = recordOutput.length();
output.writeInt(recordLength);
output.write(recordOutput.getSharedBuffer(), 0, recordLength);
@@ -170,6 +246,60 @@ class GateFilterHandlerTest {
return buffer;
}
+ private List<StreamElement>
readElementsFromSerializer(DataOutputSerializer output)
+ throws Exception {
+ List<StreamElement> elements = new ArrayList<>();
+ StreamElementSerializer<Long> serializer =
+ new StreamElementSerializer<>(LongSerializer.INSTANCE);
+ DeserializationDelegate<StreamElement> delegate =
+ new NonReusingDeserializationDelegate<>(serializer);
+
+ byte[] bodyBytes = output.getCopyOfBuffer();
+ if (bodyBytes.length == 0) {
+ return elements;
+ }
+ MemorySegment memSeg =
MemorySegmentFactory.allocateUnpooledSegment(bodyBytes.length);
+ memSeg.put(0, bodyBytes);
+ NetworkBuffer buf = new NetworkBuffer(memSeg,
FreeingBufferRecycler.INSTANCE);
+ buf.setSize(bodyBytes.length);
+
+
SpillingAdaptiveSpanningRecordDeserializer<DeserializationDelegate<StreamElement>>
+ deserializer =
+ new SpillingAdaptiveSpanningRecordDeserializer<>(
+ new String[]
{System.getProperty("java.io.tmpdir")});
+ deserializer.setNextBuffer(buf);
+
+ RecordDeserializer.DeserializationResult result;
+ do {
+ result = deserializer.getNextRecord(delegate);
+ if (result.isFullRecord()) {
+ elements.add(delegate.getInstance());
+ }
+ } while (!result.isBufferConsumed());
+ return elements;
+ }
+
+ private ChannelStateFilteringHandler.GateFilterHandler<Long> createHandler(
+ RecordFilter<Long> filter) {
+ RecordDeserializer<DeserializationDelegate<StreamElement>>
deserializer =
+ new SpillingAdaptiveSpanningRecordDeserializer<>(
+ new String[] {System.getProperty("java.io.tmpdir")});
+ VirtualChannel<Long> vc = new VirtualChannel<>(deserializer, filter);
+
+ Map<SubtaskConnectionDescriptor, VirtualChannel<Long>> channels = new
HashMap<>();
+ channels.put(KEY, vc);
+
+ StreamElementSerializer<Long> serializer =
+ new StreamElementSerializer<>(LongSerializer.INSTANCE);
+ return new ChannelStateFilteringHandler.GateFilterHandler<>(
+ channels, serializer, RescaleMappings.SYMMETRIC_IDENTITY);
+ }
+
+ private Buffer createBufferWithRecords(Long... values) throws IOException {
+ return createBufferWithElements(
+
Arrays.stream(values).map(StreamRecord::new).toArray(StreamElement[]::new));
+ }
+
private Buffer createEmptyBuffer() {
MemorySegment segment =
MemorySegmentFactory.allocateUnpooledSegment(BUFFER_SIZE);
return new NetworkBuffer(segment, FreeingBufferRecycler.INSTANCE);
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/InputChannelRecoveredStateHandlerTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/InputChannelRecoveredStateHandlerTest.java
index 61fe42a6a4c..44050135904 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/InputChannelRecoveredStateHandlerTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/InputChannelRecoveredStateHandlerTest.java
@@ -508,7 +508,8 @@ class InputChannelRecoveredStateHandlerTest extends
RecoveredChannelStateHandler
channels.put(new SubtaskConnectionDescriptor(1,
channelInfo.getInputChannelIdx()), vc);
ChannelStateFilteringHandler.GateFilterHandler<Long> gateHandler =
- new
ChannelStateFilteringHandler.GateFilterHandler<>(channels, serializer);
+ new ChannelStateFilteringHandler.GateFilterHandler<>(
+ channels, serializer,
RescaleMappings.SYMMETRIC_IDENTITY);
return new ChannelStateFilteringHandler(
new ChannelStateFilteringHandler.GateFilterHandler<?>[]
{gateHandler});
}
