sangkyoonnam commented on code in PR #1177:
URL: https://github.com/apache/flink-agents/pull/1177#discussion_r4254630230
##########
runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStore.java:
##########
@@ -212,6 +214,22 @@ public KafkaActionStateStore(
LOG.info("Initialized KafkaActionStateStore with topic: {}", topic);
}
+ /**
+ * Persist the given action state to Kafka and add it to the attempt-local
cache.
+ *
+ * <p>The record is sent, the producer is flushed, and then the record's
send future is awaited.
+ * Flushing before the wait forces the record out of the accumulator
immediately, so a
+ * synchronous write is not delayed by the producer's {@code linger.ms}.
The send future is
+ * still awaited afterwards because a successful {@code flush()} does not
guarantee that an
+ * individual record was acknowledged: a record-level, asynchronous broker
failure is surfaced
+ * to the caller.
+ *
+ * <p>The state is added to the attempt-local cache only after both the
flush and the
+ * acknowledgement succeed. If the send, the acknowledgement, or the flush
fails, the state is
+ * not cached and a {@link RuntimeException} wrapping the cause is thrown.
If the
+ * acknowledgement wait is interrupted, {@link InterruptedException} is
rethrown with the
+ * thread's interrupt flag restored.
Review Comment:
"The state is not cached" can still read as a claim about what the cache
holds, which is the concern wenjin272 raised.
`DurableExecutionManager.maybePersistTaskResult` takes the cached
`ActionState`, marks it completed, and passes that same object to `put()`. If
that `put()` fails, the cache still holds the state, already marked completed.
Describing only what `put()` does avoids that:
```suggestion
* <p>{@code put()} adds the state to the attempt-local cache only after
both the flush and the
* acknowledgement succeed. If the send, the flush, or the
acknowledgement fails, {@code put()}
* does not add the state and throws a {@link RuntimeException} wrapping
the cause. If the
* acknowledgement wait is interrupted, {@link InterruptedException} is
rethrown with the
* thread's interrupt flag restored.
```
I applied it locally and `tools/lint.sh -c` passes on JDK 11.
--
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]
