rob-9 commented on code in PR #1161:
URL: https://github.com/apache/flink-agents/pull/1161#discussion_r4108563928
##########
runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStore.java:
##########
@@ -161,9 +166,11 @@ 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:
If the thread is interrupted while waiting for Kafka here, `get()` throws
`InterruptedException` and clears the thread's interrupt flag. The `catch`
below wraps the exception but leaves that flag cleared. The caller still
receives an error, but code checking the interrupt flag no longer sees the
request to stop.
##########
runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java:
##########
@@ -233,6 +233,87 @@ void testRecoveryMarker() throws Exception {
assertThat((Map<Integer, Long>) secondMarker).containsEntry(1, 3L);
}
+ @Test
+ void testRebuildStateKeepsPendingResultBeforeRecoveryMarker() throws
Exception {
+ ActionState pendingState = new ActionState(testEvent);
+ actionStateStore.put(TEST_KEY, 1L, testAction, testEvent,
pendingState);
Review Comment:
This test stores an empty `ActionState` and checks that the store reads it
back. It never ends up executing a durable call or taking a Flink checkpoint,
so it doesn't fully verify the behavior reported in the issue.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStore.java:
##########
@@ -299,6 +310,11 @@ public void setOwnershipFilter(IntPredicate
ownershipFilter) {
this.ownershipFilter = ownershipFilter;
}
+ @Override
+ public void markCheckpointedSequence(Object key, long seqNum) {
+ latestKeySeqNum.merge(keyEncoder.generateBusinessKeyIdentity(key),
seqNum, Math::max);
Review Comment:
Each completed input adds its key to `latestKeySeqNum`. After checkpoint
completion, pruning removes the action state and its offset but leaves this
entry behind.
A task processing many different keys therefore keeps accumulating entries
in heap, even when none of those keys has remaining action state.
--
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]
