Skip to content

[hotfix][checkpoint] Refactor common behavior into ChannelStateFilteringHandler#getGateFilterHandler and GateFilterHandler#getVirtualChannelBy to reduce code duplication #28954

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
RocMarshal wants to merge 1 commit into
base: master
Choose a base branch
from
Draft

[hotfix][checkpoint] Refactor common behavior into ChannelStateFilteringHandler#getGateFilterHandler and GateFilterHandler#getVirtualChannelBy to reduce code duplication #28954

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
106 changes: 45 additions & 61 deletions ...c/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateFilteringHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@
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.util.Preconditions;

import javax.annotation.Nullable;

Expand All @@ -46,8 +47,6 @@
import java.util.List;
import java.util.Map;

import static org.apache.flink.util.Preconditions.checkNotNull;

/**
* Filters recovered channel state buffers during the channel-state-unspilling phase, removing
* records that do not belong to the current subtask after rescaling.
Expand All @@ -63,7 +62,7 @@ public class ChannelStateFilteringHandler implements Closeable {
private final GateFilterHandler<?>[] gateHandlers;

ChannelStateFilteringHandler(GateFilterHandler<?>[] gateHandlers) {
this.gateHandlers = checkNotNull(gateHandlers);
this.gateHandlers = Preconditions.checkNotNull(gateHandlers);
}

/**
Expand Down Expand Up @@ -114,21 +113,7 @@ public void filterAndRewrite(
DataOutputSerializer outputSerializer)
throws IOException {

if (gateIndex < 0 || gateIndex >= gateHandlers.length) {
throw new IllegalStateException(
"Invalid gateIndex: "
+ gateIndex
+ ", number of gates: "
+ gateHandlers.length);
}

GateFilterHandler<?> gateHandler = gateHandlers[gateIndex];
if (gateHandler == null) {
throw new IllegalStateException(
"No handler for gateIndex "
+ gateIndex
+ ". This gate is not a network input and should not have recovered buffers.");
}
GateFilterHandler<?> gateHandler = getGateFilterHandler(gateIndex);
gateHandler.filterAndRewrite(
oldSubtaskIndex, oldChannelIndex, sourceBuffer, outputSerializer);
}
Expand Down Expand Up @@ -156,21 +141,7 @@ public List<Buffer> filterAndRewrite(
BufferSupplier bufferSupplier)
throws IOException, InterruptedException {

if (gateIndex < 0 || gateIndex >= gateHandlers.length) {
throw new IllegalStateException(
"Invalid gateIndex: "
+ gateIndex
+ ", number of gates: "
+ gateHandlers.length);
}

GateFilterHandler<?> gateHandler = gateHandlers[gateIndex];
if (gateHandler == null) {
throw new IllegalStateException(
"No handler for gateIndex "
+ gateIndex
+ ". This gate is not a network input and should not have recovered buffers.");
}
GateFilterHandler<?> gateHandler = getGateFilterHandler(gateIndex);
return gateHandler.filterAndRewrite(
oldSubtaskIndex, oldChannelIndex, sourceBuffer, bufferSupplier);
}
Expand All @@ -194,6 +165,25 @@ public void close() {
}
}

private GateFilterHandler<?> getGateFilterHandler(int gateIndex) {
if (gateIndex < 0 || gateIndex >= gateHandlers.length) {
throw new IllegalStateException(
"Invalid gateIndex: "
+ gateIndex
+ ", number of gates: "
+ gateHandlers.length);
}

GateFilterHandler<?> gateHandler = gateHandlers[gateIndex];
if (gateHandler == null) {
throw new IllegalStateException(
"No handler for gateIndex "
+ gateIndex
+ ". This gate is not a network input and should not have recovered buffers.");
}
return gateHandler;
}

// -------------------------------------------------------------------------------------------
// Private static helper methods
// -------------------------------------------------------------------------------------------
Expand All @@ -210,12 +200,10 @@ private static <T> GateFilterHandler<T> createGateHandler(
InflightDataRescalingDescriptor rescalingDescriptor,
int gateIndex) {
RecordFilterContext.InputFilterConfig inputConfig = filterContext.getInputConfig(gateIndex);
if (inputConfig == null) {
throw new IllegalStateException(
"No InputFilterConfig for gateIndex "
+ gateIndex
+ ". This indicates a bug in RecordFilterContext initialization.");
}
Preconditions.checkState(
inputConfig != null,
"No InputFilterConfig for gateIndex %s. This indicates a bug in RecordFilterContext initialization.",
gateIndex);

InputGate gate = inputGates[gateIndex];
int[] oldSubtaskIndexes = rescalingDescriptor.getOldSubtaskIndexes(gateIndex);
Expand Down Expand Up @@ -314,8 +302,8 @@ static class GateFilterHandler<T> {
GateFilterHandler(
Map<SubtaskConnectionDescriptor, VirtualChannel<T>> virtualChannels,
StreamElementSerializer<T> serializer) {
this.virtualChannels = checkNotNull(virtualChannels);
this.serializer = checkNotNull(serializer);
this.virtualChannels = Preconditions.checkNotNull(virtualChannels);
this.serializer = Preconditions.checkNotNull(serializer);
this.deserializationDelegate = new NonReusingDeserializationDelegate<>(serializer);
this.outputSerializer = new DataOutputSerializer(128);
}
Expand All @@ -334,16 +322,7 @@ void filterAndRewrite(

boolean sourceBufferOwnershipTransferred = false;
try {
SubtaskConnectionDescriptor key =
new SubtaskConnectionDescriptor(oldSubtaskIndex, oldChannelIndex);
VirtualChannel<T> vc = virtualChannels.get(key);
if (vc == null) {
throw new IllegalStateException(
"No VirtualChannel found for key: "
+ key
+ "; known channels are "
+ virtualChannels.keySet());
}
VirtualChannel<T> vc = getVirtualChannelBy(oldSubtaskIndex, oldChannelIndex);

vc.setNextBuffer(sourceBuffer);
sourceBufferOwnershipTransferred = true;
Expand Down Expand Up @@ -401,16 +380,7 @@ List<Buffer> filterAndRewrite(
List<Buffer> resultBuffers = new ArrayList<>();
Buffer currentBuffer = null;
try {
SubtaskConnectionDescriptor key =
new SubtaskConnectionDescriptor(oldSubtaskIndex, oldChannelIndex);
VirtualChannel<T> vc = virtualChannels.get(key);
if (vc == null) {
throw new IllegalStateException(
"No VirtualChannel found for key: "
+ key
+ "; known channels are "
+ virtualChannels.keySet());
}
VirtualChannel<T> vc = getVirtualChannelBy(oldSubtaskIndex, oldChannelIndex);

vc.setNextBuffer(sourceBuffer);
sourceBufferOwnershipTransferred = true;
Expand Down Expand Up @@ -462,6 +432,20 @@ List<Buffer> filterAndRewrite(
}
}

private VirtualChannel<T> getVirtualChannelBy(int oldSubtaskIndex, int oldChannelIndex) {

@weiqingy weiqingy Aug 11, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: The comment at line 363 marks everything below it as code that goes away when FLINK-38544 lands. This new helper sits inside that section, but one of its two callers is at line 325, in the overload that stays. Up to now nothing inside the section was called from outside it, so that deletion could be a clean cut. Would moving the helper above line 362 be worth it to keep that true?

SubtaskConnectionDescriptor key =
new SubtaskConnectionDescriptor(oldSubtaskIndex, oldChannelIndex);
VirtualChannel<T> vc = virtualChannels.get(key);
if (vc == null) {
throw new IllegalStateException(
"No VirtualChannel found for key: "
+ key
+ "; known channels are "
+ virtualChannels.keySet());
}
return vc;
}

/**
* Serializes a single stream element into the current buffer using the length-prefixed
* format (4-byte big-endian length + record bytes) expected by Flink's record
Expand Down