KSQL-15025 | Address misc items for ksql - #11059
Open
Vedarth Sharma (VedarthConfluent) wants to merge 2 commits into
Open
KSQL-15025 | Address misc items for ksql#11059Vedarth Sharma (VedarthConfluent) wants to merge 2 commits into
Vedarth Sharma (VedarthConfluent) wants to merge 2 commits into
Conversation
…/2 clients and print-consumer on terminal paths Two related resource leaks on terminal paths: (1) KsqlClient.close() walked only the two HTTP/1 client fields (httpNonTlsClient, httpTlsClient). The HTTP/2 variants (httpNonTlsClientHttp2 and httpTlsClientHttp2) each own a Netty channel pool and event-loop threads; leaving them un-closed leaks one pool per construct/close cycle. Test harnesses, hot reloads, and any reconnect path that recreates the client accumulates dangling pools. Fix: close both HTTP/2 clients in close() with the same swallow-and- continue pattern as the HTTP/1 closes. (2) BlockingPrintPublisher.doSend() exited its outer loop when messagesWritten >= limit (LIMIT clause) but only closed topicConsumer when isClosed() was true. The KafkaConsumer (Selector thread + broker connections) leaked until the connection's end handler eventually fired, which on abrupt client disconnect may not happen. Fix: also close when the limit has been reached. The SEND_MAX_BATCH_SIZE re-schedule path keeps messagesWritten < limit, so it is not affected. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
|
🎉 All Contributor License Agreements have been signed. Ready to merge. |
…sher closed at LIMIT, not just close the consumer The previous commit closed topicConsumer when the doSend() loop terminated because LIMIT was reached. That correctly stopped the KafkaConsumer leak, but introduced a latent race: messagesWritten is a local variable reset on every doSend() invocation. If maybeSend() triggered another doSend() after LIMIT (which happens because BasePublisher's demand bookkeeping is asynchronous), the second doSend() would have messagesWritten=0, pass the loop guard, and try to topicConsumer.poll() on the now-closed consumer. Fix: when LIMIT is reached, set `closed = true` before scheduling sendComplete. The next doSend() invocation - if there is one - exits at the !isClosed() guard without touching the consumer. The bottom close-block can revert to the original `if (isClosed())` condition because closed=true now reflects both shutdown paths. This shows up in CI on PR #11059 as failures in the PRINT '...' LIMIT N integration tests in ApiIntegrationTest and RestApiTest. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Description
What behavior do you want to change, why, how does your patch achieve the changes?
Address misc items for ksql
Testing done
Describe the testing strategy. Unit and integration tests are expected for any behavior changes.
Reviewer checklist