sangkyoonnam commented on code in PR #1161:
URL: https://github.com/apache/flink-agents/pull/1161#discussion_r4242419962
##########
runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStore.java:
##########
@@ -224,13 +229,20 @@ public void put(Object key, long seqNum, Action action,
Event event, ActionState
try {
ProducerRecord<String, ActionState> kafkaRecord =
new ProducerRecord<>(topic, stateKey, state);
- producer.send(kafkaRecord);
+ RecordMetadata metadata = producer.send(kafkaRecord).get();
Review Comment:
Thanks, `d664cd6` does it: flush before `get()`, and the caches only update
after the write succeeds. `testPutDoesNotCacheFailedWrite` covers both a failed
flush and a failed send result. #1177 still conflicts in `put()` and the store
tests, so whichever lands second should keep this ordering and the offset
tracking.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStore.java:
##########
@@ -586,6 +630,12 @@ public Object getRecoveryMarker() {
for (Map.Entry<TopicPartition, Long> entry :
endOffsets.entrySet()) {
recoveryOffsets.put(entry.getKey().partition(),
entry.getValue());
}
+ latestStateOffsets.forEach(
Review Comment:
Thanks, `d87ea7b` does what I had in mind: the marker now takes the
completed-sequence boundaries from the keyed state captured by the same
checkpoint, and `latestKeySeqNum` is gone.
--
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]
