purushah commented on issue #1029:
URL: https://github.com/apache/flink-agents/issues/1029#issuecomment-5324129234
Reproduction artifacts, as promised in the issue body. Both were run against
`main` @ 3070ee21.
<details><summary><b>Standalone operator-level repro (runs with the runtime
test classpath, no Kafka needed)</b></summary>
```java
/*
* 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.flink.agents.runtime.operator;
import org.apache.flink.agents.api.Event;
import org.apache.flink.agents.api.InputEvent;
import org.apache.flink.agents.api.OutputEvent;
import org.apache.flink.agents.api.context.MemoryObject;
import org.apache.flink.agents.api.context.RunnerContext;
import org.apache.flink.agents.plan.AgentConfiguration;
import org.apache.flink.agents.plan.AgentPlan;
import org.apache.flink.agents.plan.JavaFunction;
import org.apache.flink.agents.plan.actions.Action;
import org.apache.flink.agents.runtime.actionstate.InMemoryActionStateStore;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.runtime.checkpoint.OperatorSubtaskState;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import
org.apache.flink.streaming.util.KeyedOneInputStreamOperatorTestHarness;
import org.apache.flink.util.ExceptionUtils;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
/**
* Standalone reproduction of: "Durable-execution replay of newObject()
memory updates corrupts
* short-term memory or crash-loops recovery".
*
* <p>Runs the real {@link ActionExecutionOperator} on the local Flink
operator runtime (real keyed
* state backend, real checkpoint snapshot/restore) with durable execution
enabled, using the same
* in-memory ActionStateStore the project's own replay tests use.
*
* <p>The agent action does nothing exotic — it is the documented short-term
memory usage pattern:
*
* <pre>
* MemoryObject mem = context.getShortTermMemory();
* mem.newObject("user");
* mem.set("user.score", input + 1);
* </pre>
*
* <p>Scenario 1 — recovery into empty state: a checkpoint was taken before
the input; after the
* failure the input is re-delivered, the action is found completed in the
ActionState store and is
* skipped, and its recorded memory updates are replayed. The newObject
update was recorded as
* MemoryUpdate(path, null), indistinguishable from set(path, null), so
replay materializes "user"
* as a null VALUE leaf and the subsequent replay of set("user.score", ...)
throws
* UnsupportedOperationException (child write under a value leaf).
*
* <p>Scenario 2 — recovery over restored state: input #1 created the "user"
object and a checkpoint
* persisted it; input #2 for the same key completed (its ActionState
survives) but the job failed
* before the next checkpoint. On recovery input #2 is re-delivered and its
updates are replayed
* against the restored state: replaying the newObject update as set("user",
null) hits the existing
* OBJECT node and throws IllegalArgumentException("Cannot overwrite object
with value") — every
* recovery attempt fails the same way, i.e. a permanent recovery crash loop.
*
* <p>Run with the runtime test classpath, e.g.:
*
* <pre>
* mvn -pl runtime test-compile
* mvn -pl runtime org.codehaus.mojo:exec-maven-plugin:3.1.0:java \
* -Dexec.classpathScope=test \
*
-Dexec.mainClass=org.apache.flink.agents.runtime.operator.NewObjectReplayCrashRepro
* </pre>
*/
public final class NewObjectReplayCrashRepro {
private NewObjectReplayCrashRepro() {}
/** The agent action: documented nested short-term-memory usage. */
public static void nestedMemoryAction(Event event, RunnerContext
context) {
Long input = (Long) InputEvent.fromEvent(event).getInput();
try {
MemoryObject mem = context.getShortTermMemory();
mem.newObject("user");
mem.set("user.score", input + 1);
context.sendEvent(new OutputEvent(input + 1));
} catch (Exception e) {
ExceptionUtils.rethrow(e);
}
}
private static AgentPlan agentPlan() throws Exception {
Action action =
new Action(
"nestedMemoryAction",
new JavaFunction(
NewObjectReplayCrashRepro.class,
"nestedMemoryAction",
new Class<?>[] {Event.class,
RunnerContext.class}),
Collections.singletonList(InputEvent.EVENT_TYPE));
Map<String, Action> actions = new HashMap<>();
actions.put(action.getName(), action);
return new AgentPlan(actions, new HashMap<>(), new
AgentConfiguration());
}
private static KeyedOneInputStreamOperatorTestHarness<Long, Long,
Object> harness(
AgentPlan plan, InMemoryActionStateStore store) throws Exception
{
// All inputs share one Flink key so scenario 2 replays over the
state the earlier input
// left behind for the same key.
return new KeyedOneInputStreamOperatorTestHarness<>(
new ActionExecutionOperatorFactory<Long, Object>(plan, true,
store),
(KeySelector<Long, Long>) value -> 0L,
TypeInformation.of(Long.class));
}
private static void process(
KeyedOneInputStreamOperatorTestHarness<Long, Long, Object> h,
long value)
throws Exception {
h.processElement(new StreamRecord<>(value));
((ActionExecutionOperator<Long, Object>)
h.getOperator()).waitInFlightEventsFinished();
}
public static void main(String[] args) throws Exception {
int failures = 0;
failures += scenario1RecoveryIntoEmptyState() ? 0 : 1;
failures += scenario2RecoveryOverRestoredState() ? 0 : 1;
System.out.println();
if (failures == 0) {
System.out.println("RESULT: both recovery scenarios succeeded
(bug is fixed).");
} else {
System.out.println(
"RESULT: " + failures + " of 2 recovery scenarios FAILED
(bug reproduced).");
System.exit(1);
}
}
private static boolean scenario1RecoveryIntoEmptyState() throws
Exception {
System.out.println("=== Scenario 1: recovery into empty keyed state
===");
AgentPlan plan = agentPlan();
InMemoryActionStateStore store = new InMemoryActionStateStore(false);
try (KeyedOneInputStreamOperatorTestHarness<Long, Long, Object> h =
harness(plan, store)) {
h.open();
process(h, 7L);
System.out.println("first run: action executed, output emitted,
job 'fails' now");
}
// Recovery from a checkpoint taken before the input: keyed state is
empty, the completed
// ActionState survives, so the operator skips the action and
replays its memory updates.
try (KeyedOneInputStreamOperatorTestHarness<Long, Long, Object> h =
harness(plan, store)) {
h.open();
process(h, 7L);
List<StreamRecord<Object>> out = (List<StreamRecord<Object>>)
h.getRecordOutput();
System.out.println("recovery: replay succeeded, output = " +
out);
return true;
} catch (Exception e) {
System.out.println("recovery FAILED during memory-update
replay:");
e.printStackTrace(System.out);
return false;
}
}
private static boolean scenario2RecoveryOverRestoredState() throws
Exception {
System.out.println();
System.out.println("=== Scenario 2: recovery over restored state
(crash loop) ===");
AgentPlan plan = agentPlan();
InMemoryActionStateStore store = new InMemoryActionStateStore(false);
OperatorSubtaskState snapshot;
try (KeyedOneInputStreamOperatorTestHarness<Long, Long, Object> h =
harness(plan, store)) {
h.open();
process(h, 7L); // input #1 creates the "user" object
snapshot = h.snapshot(1L, 1L); // checkpoint persists it
process(h, 9L); // input #2 completes; job 'fails' before the
next checkpoint
System.out.println("first run: two inputs processed, checkpoint
taken in between");
}
// Recovery: restored state already contains "user" as an OBJECT;
input #2 is re-delivered
// and its recorded memory updates are replayed against that state.
try (KeyedOneInputStreamOperatorTestHarness<Long, Long, Object> h =
harness(plan, store)) {
h.initializeState(snapshot);
h.open();
process(h, 9L);
List<StreamRecord<Object>> out = (List<StreamRecord<Object>>)
h.getRecordOutput();
System.out.println("recovery: replay succeeded, output = " +
out);
return true;
} catch (Exception e) {
System.out.println(
"recovery FAILED during memory-update replay (would
repeat on every restart"
+ " => permanent crash loop):");
e.printStackTrace(System.out);
return false;
}
}
}
```
</details>
<details><summary><b>Local Flink job log — before fix (crash loop, job
dead)</b></summary>
```
alice: total=1 (+1)
alice: total=3 (+2)
[FailOnceMap] checkpoint 1 completed
alice: total=6 (+3)
Exception in thread "main"
org.apache.flink.runtime.client.JobExecutionException: Job execution failed.
at
org.apache.flink.runtime.jobmaster.JobResult.toJobExecutionResult(JobResult.java:180)
at
org.apache.flink.runtime.minicluster.MiniClusterJobClient.lambda$getJobExecutionResult$3(MiniClusterJobClient.java:140)
at
java.base/java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:646)
at
java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:510)
at
java.base/java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:2147)
at
org.apache.flink.runtime.rpc.pekko.PekkoInvocationHandler.lambda$invokeRpc$1(PekkoInvocationHandler.java:268)
at
java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:863)
at
java.base/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:841)
at
java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:510)
at
java.base/java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:2147)
at
org.apache.flink.util.concurrent.FutureUtils.doForward(FutureUtils.java:1317)
at
org.apache.flink.runtime.concurrent.ClassLoadingUtils.lambda$guardCompletionWithContextClassLoader$1(ClassLoadingUtils.java:93)
at
org.apache.flink.runtime.concurrent.ClassLoadingUtils.runWithContextClassLoader(ClassLoadingUtils.java:68)
at
org.apache.flink.runtime.concurrent.ClassLoadingUtils.lambda$guardCompletionWithContextClassLoader$2(ClassLoadingUtils.java:92)
at
java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:863)
at
java.base/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:841)
at
java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:510)
at
java.base/java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:2147)
at
org.apache.flink.runtime.concurrent.pekko.ScalaFutureUtils$1.onComplete(ScalaFutureUtils.java:47)
at org.apache.pekko.dispatch.OnComplete.internal(Future.scala:338)
at org.apache.pekko.dispatch.OnComplete.internal(Future.scala:335)
at org.apache.pekko.dispatch.japi$CallbackBridge.apply(Future.scala:259)
at org.apache.pekko.dispatch.japi$CallbackBridge.apply(Future.scala:256)
at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:74)
at
org.apache.flink.runtime.concurrent.pekko.ScalaFutureUtils$DirectExecutionContext.execute(ScalaFutureUtils.java:65)
at
scala.concurrent.impl.CallbackRunnable.executeWithValue(Promise.scala:82)
at
scala.concurrent.impl.Promise$DefaultPromise.$anonfun$tryComplete$1(Promise.scala:298)
at
scala.concurrent.impl.Promise$DefaultPromise.$anonfun$tryComplete$1$adapted(Promise.scala:298)
at
scala.concurrent.impl.Promise$DefaultPromise.tryComplete(Promise.scala:298)
at org.apache.pekko.pattern.PromiseActorRef.$bang(AskSupport.scala:625)
at
org.apache.pekko.pattern.PipeToSupport$PipeableFuture$$anonfun$pipeTo$1.applyOrElse(PipeToSupport.scala:33)
at scala.concurrent.Future.$anonfun$andThen$1(Future.scala:536)
at scala.concurrent.impl.Promise.$anonfun$transform$1(Promise.scala:42)
at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:74)
at
org.apache.pekko.dispatch.BatchingExecutor$AbstractBatch.processBatch(BatchingExecutor.scala:72)
at
org.apache.pekko.dispatch.BatchingExecutor$BlockableBatch.$anonfun$run$1(BatchingExecutor.scala:109)
at
scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)
at
scala.concurrent.BlockContext$.withBlockContext(BlockContext.scala:85)
at
org.apache.pekko.dispatch.BatchingExecutor$BlockableBatch.run(BatchingExecutor.scala:109)
at
org.apache.pekko.dispatch.TaskInvocation.run(AbstractDispatcher.scala:59)
at
org.apache.pekko.dispatch.ForkJoinExecutorConfigurator$PekkoForkJoinTask.exec(ForkJoinExecutorConfigurator.scala:62)
at
java.base/java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:373)
at
java.base/java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(ForkJoinPool.java:1182)
at
java.base/java.util.concurrent.ForkJoinPool.scan(ForkJoinPool.java:1655)
at
java.base/java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1622)
at
java.base/java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:165)
Caused by: org.apache.flink.runtime.JobException: Recovery is suppressed by
FixedDelayRestartBackoffTimeStrategy(maxNumberRestartAttempts=3,
backoffTimeMS=2000)
at
org.apache.flink.runtime.executiongraph.failover.ExecutionFailureHandler.handleFailure(ExecutionFailureHandler.java:213)
at
org.apache.flink.runtime.executiongraph.failover.ExecutionFailureHandler.handleFailureAndReport(ExecutionFailureHandler.java:163)
at
org.apache.flink.runtime.executiongraph.failover.ExecutionFailureHandler.getFailureHandlingResult(ExecutionFailureHandler.java:118)
at
org.apache.flink.runtime.scheduler.DefaultScheduler.recordTaskFailure(DefaultScheduler.java:300)
at
org.apache.flink.runtime.scheduler.DefaultScheduler.handleTaskFailure(DefaultScheduler.java:291)
at
org.apache.flink.runtime.scheduler.DefaultScheduler.onTaskFailed(DefaultScheduler.java:284)
at
org.apache.flink.runtime.scheduler.SchedulerBase.onTaskExecutionStateUpdate(SchedulerBase.java:836)
at
org.apache.flink.runtime.scheduler.SchedulerBase.updateTaskExecutionState(SchedulerBase.java:813)
```
</details>
<details><summary><b>Local Flink job log — after fix (identical failure,
clean recovery)</b></summary>
```
alice: total=1 (+1)
alice: total=3 (+2)
[FailOnceMap] checkpoint 1 completed
alice: total=6 (+3)
alice: total=3 (+2)
alice: total=6 (+3)
alice: total=10 (+4)
alice: total=15 (+5)
alice: total=21 (+6)
[FailOnceMap] checkpoint 2 completed
alice: total=28 (+7)
alice: total=36 (+8)
alice: total=45 (+9)
alice: total=55 (+10)
alice: total=66 (+11)
alice: total=78 (+12)
[FailOnceMap] checkpoint 3 completed
alice: total=91 (+13)
alice: total=105 (+14)
alice: total=120 (+15)
alice: total=136 (+16)
alice: total=153 (+17)
alice: total=171 (+18)
alice: total=190 (+19)
[FailOnceMap] checkpoint 4 completed
alice: total=210 (+20)
[FailOnceMap] checkpoint 5 completed
JOB FINISHED SUCCESSFULLY
```
</details>
--
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]
