weiqingy opened a new pull request, #948: URL: https://github.com/apache/flink-agents/pull/948
Linked issue: none (hotfix) ### Purpose of change `KafkaActionStateStore.close()` closes the producer and the consumer in sequence. If `producer.close()` throws, the consumer is never closed and its network client, background heartbeat thread, and coordinator connection are leaked for the lifetime of the TaskManager JVM. The sibling store in the same package already handles this: `FlussActionStateStore.close()` closes the first resource in a `try` and the second in a `finally`. This applies the same shape to the Kafka store. Behavior on the success path is unchanged, and the producer's exception still propagates to the caller. ### Tests Added `KafkaActionStateStoreTest#testCloseClosesConsumerWhenProducerCloseFails`, which stubs a producer whose `close()` throws and asserts both that the exception still reaches the caller and that `consumer.close()` was invoked. Verified the test actually catches the bug: with the `try`/`finally` reverted it fails on `verify(consumer).close()` with "Actually, there were zero interactions with this mock", and passes with the fix in place. `KafkaActionStateStoreTest` 8/8, the `actionstate` package 83/83, and the full `runtime` module 402/402 pass. `spotless:check` is clean. ### API No. `close()` is an existing `ActionStateStore` method and its signature is unchanged. ### Documentation - [ ] `doc-needed` - [x] `doc-not-needed` - [ ] `doc-included` -- 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]
