Sxnan commented on code in PR #422:
URL: https://github.com/apache/flink-agents/pull/422#discussion_r2685054419
##########
runtime/src/main/java/org/apache/flink/agents/runtime/python/context/PythonRunnerContextImpl.java:
##########
@@ -48,4 +78,120 @@ public void sendEvent(String type, byte[] event, String
eventString) {
// this method will be invoked by PythonActionExecutor's python
interpreter.
sendEvent(new PythonEvent(event, type, eventString));
}
+
+ /**
+ * Initializes the recovery state for the current action.
+ *
+ * @param actionState the ActionState for the current action (used for
adding CallRecords)
+ * @param actionStatePersister callback to persist ActionState after each
code block completion
+ */
+ public void initRecoveryState(ActionState actionState, Runnable
actionStatePersister) {
+ this.currentCallIndex = 0;
+ this.currentActionState = actionState;
+ this.actionStatePersister = actionStatePersister;
+ this.recoveryCallRecords =
+ actionState != null && actionState.getCallRecords() != null
+ ? new ArrayList<>(actionState.getCallRecords())
+ : new ArrayList<>();
+ }
+
+ /**
+ * Gets the current call index.
+ *
+ * @return the current call index
+ */
+ public int getCurrentCallIndex() {
+ return currentCallIndex;
+ }
+
+ /**
+ * Tries to get a cached CallRecord for recovery.
+ *
+ * <p>This method is called by Python's execute/execute_async to check if
a previous result
+ * exists. If found and validated, the cached result is returned.
+ *
+ * @param functionId the function identifier
+ * @param argsDigest the digest of serialized arguments
+ * @return array containing [isHit (boolean), resultPayload (byte[]),
exceptionPayload
+ * (byte[])], or null if miss
+ */
+ public Object[] tryGetCachedCallRecord(String functionId, String
argsDigest) {
+ mailboxThreadChecker.run();
+
+ if (currentCallIndex < recoveryCallRecords.size()) {
+ CallRecord record = recoveryCallRecords.get(currentCallIndex);
+
+ if (record.matches(functionId, argsDigest)) {
+ LOG.debug(
+ "CallRecord hit at index {}: functionId={},
argsDigest={}",
+ currentCallIndex,
+ functionId,
+ argsDigest);
+ currentCallIndex++;
+ return new Object[] {true, record.getResultPayload(),
record.getExceptionPayload()};
+ } else {
+ // Non-deterministic call detected, clear subsequent records
+ LOG.warn(
Review Comment:
I think strict_mode could be useful for development/testing to catch
non-deterministic issues early.
For now, I'd prefer to keep the current lenient behavior as the default
since it's more resilient in production. We can add this as a follow-up
enhancement. WDYT?
--
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]
