gemini-code-assist[bot] commented on code in PR #39191:
URL: https://github.com/apache/beam/pull/39191#discussion_r3507357304
##########
runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/ExecutableStageDoFnOperator.java:
##########
@@ -1009,9 +1010,17 @@ public <KeyT> void onTimer(
Preconditions.checkNotNull(remoteBundle, "Call to onTimer outside of a
bundle");
if (StateAndTimerBundleCheckpointHandler.isSdfTimer(timerId)) {
StateNamespace namespace = StateNamespaces.window(windowCoder, window);
- WindowedValue stateValue =
- keyedStateInternals.state(namespace, StateTags.value(timerId,
residualCoder)).read();
+ // Residuals from pre-#27648 savepoints used per-residual ValueState
tags and are not
+ // migrated; an older savepoint drops any in-flight SDF residual on
this experimental path.
+ MapState<String, WindowedValue<InputT>> residualState =
+ keyedStateInternals.state(
+ namespace,
StateAndTimerBundleCheckpointHandler.residualStateTag(residualCoder));
+ WindowedValue<InputT> stateValue = residualState.get(timerId).read();
processElement(stateValue);
Review Comment:
If an older savepoint is used, `residualState.get(timerId).read()` will
return `null`. Passing `null` to `processElement` will cause a
`NullPointerException`. We should add a null check to safely drop the residual
as intended.
```suggestion
WindowedValue<InputT> stateValue = residualState.get(timerId).read();
if (stateValue != null) {
processElement(stateValue);
}
```
##########
runners/flink/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkExecutableStageFunction.java:
##########
@@ -284,11 +284,13 @@ public void mapPartition(
List<WindowedValue<InputT>> residuals = new ArrayList<>();
TimerInternals.TimerData timer;
while ((timer = sdfTimerInternals.removeNextProcessingTimer()) !=
null) {
- WindowedValue stateValue =
- sdfStateInternals
- .state(timer.getNamespace(),
StateTags.value(timer.getTimerId(), inputCoder))
- .read();
-
+ MapState<String, WindowedValue<InputT>> residualState =
+ sdfStateInternals.state(
+ timer.getNamespace(),
+
BundleCheckpointHandlers.StateAndTimerBundleCheckpointHandler.residualStateTag(
+ inputCoder));
+ WindowedValue<InputT> stateValue =
residualState.get(timer.getTimerId()).read();
+ residualState.remove(timer.getTimerId());
residuals.add(stateValue);
Review Comment:
If an older savepoint is used, the residual state won't be found under the
new `MapState` tag, and `residualState.get(timer.getTimerId()).read()` will
return `null`. Adding `null` to `residuals` will lead to a
`NullPointerException` during `processElements`. We should add a null check to
safely drop the residual as intended.
```java
WindowedValue<InputT> stateValue =
residualState.get(timer.getTimerId()).read();
residualState.remove(timer.getTimerId());
if (stateValue != null) {
residuals.add(stateValue);
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
