#776 Make sure all closeables opened in another thread are closed on successful path.#777
Conversation
…successful path.
WalkthroughThe changes improve thread-owned resource cleanup, coordinate JDBC connection closure with the registry, wait after interrupting timed-out threads, and stop JDBC retries when interrupted. A Scaladoc grammar correction and a registry cleanup test are also included. ChangesThread resource cleanup
JDBC retry interruption
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant ThreadUtils
participant WorkerThread
participant ThreadClosableRegistry
participant QueryExecutorJdbc
ThreadUtils->>WorkerThread: interrupt after timeout
ThreadUtils->>WorkerThread: wait for closeWaitMillis
ThreadUtils->>ThreadClosableRegistry: cleanupThread(threadId)
ThreadClosableRegistry->>QueryExecutorJdbc: close registered connection
QueryExecutorJdbc->>ThreadClosableRegistry: closeCloseable(conn)
ThreadClosableRegistry-->>QueryExecutorJdbc: unregister and close
Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
🧹 Nitpick comments (3)
pramen/core/src/main/scala/za/co/absa/pramen/core/reader/JdbcUrlSelectorImpl.scala (1)
154-167: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueInterruptedException handling correctly bypasses retry and backoff.
The pattern match rethrows
InterruptedExceptionimmediately, while preserving the existing retry/backoff flow for all otherThrowablecases.Thread.sleepat line 162 is outside theTryblock, so an interruption during backoff also propagates directly without retrying — exactly the desired behavior. The downstream caller inQueryExecutorJdbc.executeActionOnConnection(lines 119-147) catches and rethrowsInterruptedExceptionwithout swallowing it, so the propagation chain is intact.One minor note: the inner
case ex: InterruptedExceptionshadows the outerexfromFailure(ex). This is valid Scala but could be slightly clearer with a distinct name (e.g.,ie), though it's not a functional concern.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@pramen/core/src/main/scala/za/co/absa/pramen/core/reader/JdbcUrlSelectorImpl.scala` around lines 154 - 167, Rename the pattern-bound variable in the InterruptedException branch of getNewConnection from ex to a distinct name such as ie, and rethrow that renamed variable; leave the existing retry and backoff handling for other Throwable cases unchanged.pramen/core/src/main/scala/za/co/absa/pramen/core/runner/task/ThreadClosableRegistry.scala (1)
47-62: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueScaladoc is misleading and has grammar/typo issues for this new public method.
The doc block appears copied from
registerCloseable: line 49 claims the resource is closed later "whencleanupThreadis called", butcloseCloseablecloses it immediately; the@paramsays "to register" instead of "to close"; and line 48 reads "unregisters is as an transactional atomic operation". Also note the close happens outside the synchronizedunregisterCloseable, so it is not truly atomic. The log message on line 57 has a typo ("closable").📝 Suggested doc/message fixes
- /** - * Closes a closeable resource and unregisters is as an transactional atomic operation. - * The resource will be automatically closed when [[cleanupThread]] is called for this thread. - * - * `@param` closeable The AutoCloseable resource to register - */ + /** + * Unregisters a closeable resource from the registry and then closes it. + * Close exceptions are logged and swallowed; they are not rethrown. + * + * `@param` closeable The AutoCloseable resource to close + */ def closeCloseable(closeable: AutoCloseable): Unit = { unregisterCloseable(closeable) try { - log.info(s"Closing closable of type ${closeable.getClass.getCanonicalName}...") + log.info(s"Closing closeable of type ${closeable.getClass.getCanonicalName}...")🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@pramen/core/src/main/scala/za/co/absa/pramen/core/runner/task/ThreadClosableRegistry.scala` around lines 47 - 62, Correct the Scaladoc for closeCloseable to describe immediate closing and unregistration, fix the grammar and “an transactional” typo, and change the `@param` text to say the resource is being closed. Avoid claiming the operation is atomic, since close occurs outside unregisterCloseable synchronization. Also fix the log message in closeCloseable to use “closeable” instead of “closable”.pramen/core/src/main/scala/za/co/absa/pramen/core/utils/ThreadUtils.scala (1)
64-83: 🩺 Stability & Availability | 🔵 Trivial | 💤 Low valueThe reworked sequencing is correct: capturing the stack trace before
interrupt(), thenjoin(closeWaitMillis)to let the thread self-clean, and reusing the precomputedisAlivefor the later timeout branch all look right. UsingMath.max(..., MIN_WAIT_AFTER_INTERRUPT_MS)also correctly preventsthread.join(0)from blocking indefinitely whencleanupTimeoutis zero.One operational note: with the default
cleanupTimeoutof 5 minutes,closeWaitMillisbecomes 5 minutes, and if closeables remain the subsequentDetachedRunService.runWithTimeoutThenDetach(cleanupTimeout)can wait up to anothercleanupTimeout. In the worst case the calling (runner) thread can block for roughly2 × cleanupTimeout. Confirm this combined budget is acceptable for the task runner, or consider deriving the post-interrupt wait from a smaller bound.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@pramen/core/src/main/scala/za/co/absa/pramen/core/utils/ThreadUtils.scala` around lines 64 - 83, Review the timeout budget in the thread cleanup flow around the thread-closing method: after the initial wait, closeWaitMillis may consume the full cleanupTimeout before DetachedRunService.runWithTimeoutThenDetach(cleanupTimeout) waits again. Confirm that the potential 2×cleanupTimeout blocking period is acceptable; otherwise derive closeWaitMillis from a smaller bounded post-interrupt timeout while preserving the minimum wait safeguard.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Nitpick comments:
In
`@pramen/core/src/main/scala/za/co/absa/pramen/core/reader/JdbcUrlSelectorImpl.scala`:
- Around line 154-167: Rename the pattern-bound variable in the
InterruptedException branch of getNewConnection from ex to a distinct name such
as ie, and rethrow that renamed variable; leave the existing retry and backoff
handling for other Throwable cases unchanged.
In
`@pramen/core/src/main/scala/za/co/absa/pramen/core/runner/task/ThreadClosableRegistry.scala`:
- Around line 47-62: Correct the Scaladoc for closeCloseable to describe
immediate closing and unregistration, fix the grammar and “an transactional”
typo, and change the `@param` text to say the resource is being closed. Avoid
claiming the operation is atomic, since close occurs outside unregisterCloseable
synchronization. Also fix the log message in closeCloseable to use “closeable”
instead of “closable”.
In `@pramen/core/src/main/scala/za/co/absa/pramen/core/utils/ThreadUtils.scala`:
- Around line 64-83: Review the timeout budget in the thread cleanup flow around
the thread-closing method: after the initial wait, closeWaitMillis may consume
the full cleanupTimeout before
DetachedRunService.runWithTimeoutThenDetach(cleanupTimeout) waits again. Confirm
that the potential 2×cleanupTimeout blocking period is acceptable; otherwise
derive closeWaitMillis from a smaller bounded post-interrupt timeout while
preserving the minimum wait safeguard.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: ccdc12c9-aacc-4580-81b8-164ffec37723
📒 Files selected for processing (6)
pramen/core/src/main/scala/za/co/absa/pramen/core/reader/JdbcUrlSelector.scalapramen/core/src/main/scala/za/co/absa/pramen/core/reader/JdbcUrlSelectorImpl.scalapramen/core/src/main/scala/za/co/absa/pramen/core/runner/task/ThreadClosableRegistry.scalapramen/core/src/main/scala/za/co/absa/pramen/core/utils/ThreadUtils.scalapramen/core/src/main/scala/za/co/absa/pramen/core/utils/hive/QueryExecutorJdbc.scalapramen/core/src/test/scala/za/co/absa/pramen/core/tests/runner/task/ThreadClosableRegistrySuite.scala
Unit Test Coverage
Files
|
1 participant
Closes #776
Overview
Makes sure all closeables opened in another thread are closed on successful path.
Release Notes
Related
Summary by CodeRabbit
Documentation
Tests