weiqingy commented on code in PR #948:
URL: https://github.com/apache/flink-agents/pull/948#discussion_r3709616701
##########
runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStore.java:
##########
@@ -331,11 +331,14 @@ public Object getRecoveryMarker() {
@Override
public void close() throws Exception {
- if (producer != null) {
- producer.close();
- }
- if (consumer != null) {
- consumer.close();
+ try {
+ if (producer != null) {
+ producer.close();
+ }
+ } finally {
+ if (consumer != null) {
+ consumer.close();
Review Comment:
You're right, and it's worse than the producer exception being demoted. A
`finally` that throws discards the in-flight exception, so the producer failure
wasn't reachable at all.
`close()` now attempts both closes and rethrows the first failure with the
later one attached via `addSuppressed`, matching `ResourceCache.close()` and
`SkillManager.closeRepos()`. Added
`testCloseKeepsProducerFailureWhenBothCloseFail` for your case, plus two more
so all four outcome combinations are covered.
Also rewrote the PR description, since it claimed the producer's exception
still propagates.
One more thing: `FlussActionStateStore.close()` has the same `try`/`finally`
and masks the same way. I left it out for now to keep this scoped. WDYT, do you
have any preference on fixing it here, or separately?
--
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]
