fix(hive-sync): close the SessionState HiveQL sync starts - #19812
fix(hive-sync): close the SessionState HiveQL sync starts#19812skywalker0618 wants to merge 4 commits into
Conversation
HiveQueryDDLExecutor starts a SessionState in its constructor and never closed it. Hive derives a session's four scratch directory roots from hive.session.id and reclaims them only in SessionState.close(), so every sync left a directory set behind, along with the session's registry and class loaders. A HiveSyncTool is built per sync, so on a long-running streaming job this accumulates for the life of the JVM. close() now closes the session, after the Driver teardown because Driver.destroy() can reach SessionState.get() while releasing locks, and in a finally so a Driver close that throws cannot skip it. The constructor's error path uses the same helper, which also stops a RuntimeException from a failing teardown masking the construction error. Closing the session detaches it from the calling thread, which Hive does unconditionally for whichever session is attached. The single-session SQL path therefore re-asserts its own session before running statements, as Hive documents a thread running several sessions must, instead of relying on a thread local another executor may have replaced or cleared.
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! This PR closes the SessionState that HiveQueryDDLExecutor starts per sync, fixing a scratch-directory / per-session-object leak on long-lived HiveQL-mode jobs. It moves the session close into a finally after the Driver teardown, reuses a closeQuietly helper on the constructor error path, and re-asserts the executor's own thread-local session before running statements. No correctness issues found. A few style/readability suggestions in the inline comments. Please take a look, and this should be ready for a Hudi committer or PMC member to take it from here. Code looks clean overall — one minor naming nit below.
cc @yihua
| } catch (Exception e) { | ||
| log.error("Error while closing SessionState", e); | ||
| } | ||
| } |
There was a problem hiding this comment.
🤖 nit: could you rename the parameter to avoid the same name as the instance field? Something like state would make it immediately clear this is a static helper and not accidentally referencing the field.
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #19812 +/- ##
============================================
+ Coverage 77.88% 78.32% +0.43%
- Complexity 33265 33904 +639
============================================
Files 2533 2542 +9
Lines 140348 141787 +1439
Branches 16913 17586 +673
============================================
+ Hits 109304 111048 +1744
+ Misses 23406 23028 -378
- Partials 7638 7711 +73
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
|
Hi @danny0405 do you mind taking a look at this fix? Found this issue during the stress testing in Uber and fixed it by this change. Thanks! |
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! This PR closes the SessionState that HiveQueryDDLExecutor starts per sync, releasing its scratch directories and per-session objects that previously accumulated for the life of a long-running JVM. The session-close ordering (after Driver teardown, in a finally), the widened catch in the constructor error path, and the SessionState re-assertion in updateHiveSQLs all trace through correctly, and the new tests cover the ordering and failure paths. No issues flagged from this automated pass — a Hudi committer or PMC member can take it from here for a final review.
cc @yihua
| hiveDriver.close(); | ||
| } finally { | ||
| destroyQuietly(hiveDriver); | ||
| try { |
There was a problem hiding this comment.
[P1] Bind the owned session during Driver teardown
Driver.close() and destroy() consult SessionState.get()—the former clears the current session’s lineage state, and the latter can obtain its transaction manager when releasing locks. If executor B was constructed or run after executor A, calling A.close() here tears down A’s Driver while B’s session is current, potentially clearing or operating on B’s state; A.sessionState.close() then also detaches B. Capture the previous session, bind sessionState before Driver/session teardown, and restore the previous session afterward.
| // thread writes: the one constructed most recently wins, and one that is closed clears it. | ||
| // Re-assert ours, as Hive documents a thread running several sessions must, so these | ||
| // statements run under the session that owns hiveDriver. | ||
| SessionState.setCurrentSessionState(sessionState); |
There was a problem hiding this comment.
[P2] Restore the previous thread-local session
This assignment remains in the thread local after updateHiveSQLs returns. With interleaved executors—or an embedding application that already owns a Hive session—subsequent Hive work silently runs under this executor’s database, configuration, and transaction state. Make the binding scoped: save SessionState.get(), bind this executor for the Driver calls, then restore the saved session (or detach when it was null) in a finally block.
Driver.close() and Driver.destroy() both act on whichever SessionState the thread currently holds, so an executor closed while a later-constructed executor's session is attached would clear that session's lineage state, release locks through its transaction manager, and then detach it.
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! The PR closes the SessionState that HiveQueryDDLExecutor starts, adding a closeDriverAndSession() helper that binds the owned session during Driver teardown and restores the previously-attached session afterward, plus a session re-assertion in updateHiveSQLs. I traced the teardown ordering and the thread-local session save/restore path; the remaining concerns (binding the owned session during Driver teardown, and restoring the previous thread-local in updateHiveSQLs) are already captured in the prior-round comments from @danny0405, so nothing new to add from this automated pass. Please take a look at any inline comments, and this should be ready for a Hudi committer or PMC member to take it from here.
Leaving this executor's session in the thread local made the caller's later Hive work run under our database, configuration and transaction state, which is wrong for an interleaved executor and for an application that embeds the sync and owns its own session.
github-actions
Bot
added
size:L
| if (previousSession != null) { | ||
| SessionState.setCurrentSessionState(previousSession); | ||
| } else { | ||
| SessionState.detachSession(); |
There was a problem hiding this comment.
[P2] Restore the caller context class loader too
Hive SessionState.setCurrentSessionState changes Thread.currentThread().getContextClassLoader() to the session conf loader, while detachSession() only removes the Hive thread-local. Thus, when previousSession is null, this finally block leaves the caller thread using this executor class loader; closeDriverAndSession has the same problem and can leave the thread pointing at a loader that SessionState.close() just closed. That both leaks the session loader and can break later class loading on long-running worker threads. Please capture the original context class loader before binding and restore it in the finally path, including the null-session case, with an assertion in the lifecycle tests.
| } | ||
| destroyQuietly(this.hiveDriver); | ||
| try { | ||
| closeDriverAndSession(); |
There was a problem hiding this comment.
[P1] Preserve the session displaced by construction
SessionState.start(this.sessionState) overwrites any session already attached to the caller thread, but the constructor never saves or restores that session. In the real lifecycle, updateHiveSQLs therefore observes this executor session as previousSession and restores it, and this teardown observes the same session, closes/detaches it, and leaves the caller with no session; the original embedding session is never recovered. The new tests miss this because executorWith bypasses the constructor and manually attaches otherSession afterward. Now that each SQL/teardown operation explicitly binds the owned session, could the constructor capture the pre-existing session before start, restore it after initialization and on failure, and add a constructor-path regression test?
5 participants
Describe the issue this Pull Request addresses
HiveQueryDDLExecutorstarts aSessionStatein its constructor and never closes it. Hive derivesa session's four scratch directory roots from
hive.session.idand reclaims them only inSessionState.close(), so every sync leaves a directory set behind, along with the session'sfunction registry and the class loaders it created. A
HiveSyncTool, and therefore an executor andits session, is built per sync, so on a long-running streaming job this accumulates for the life of
the JVM:
close()released the metastore client and the Driver, never the session.Observed on a Flink job syncing a 5,000-partition table in HiveQL mode: one unclosed session per
sync cycle, 63 of 63 and 76 of 76 across two runs, each leaving its scratch directory set on local
disk.
Summary and Changelog
Users running
hive_sync.mode=hiveqlon a long-lived job stop accumulating scratch directories andper-session objects; the session is now released with the rest of the executor's resources.
close()closes theSessionState. It runs after the Driver teardown, becauseSessionState.close()detaches the session from the calling thread andDriver.destroy()canreach
SessionState.get()while releasing locks, and it runs in afinallyso a Driverclose()that throws cannot skip it.
closeQuietlyhelper instead of its own inlineblock. Besides removing the duplication, it widens the caught type from
IOExceptiontoException, so aRuntimeExceptionduring teardown can no longer mask the construction failurethat is about to be thrown.
updateHiveSQLsre-asserts its own session before running statements. This is required by thechange above:
SessionState.close()callsdetachSession(), which clears the thread local forwhichever session is attached, so an executor can no longer assume the session it started in its
constructor is still current when a second executor on the same thread is constructed or closed.
Driver.compile()dereferencesSessionState.get()unconditionally, and Hive documents that athread running several sessions must set the current session when switching between them.
TestHiveQueryDDLExecutorSessioncovers the session lifecycle: closed after the Driver and inthat order, closed even when the Driver
close()throws, a failing session close swallowed so itdoes not become the caller's problem, and SQL running under the session the executor started.
Impact
No public API or config change. Behaviour change is that the HiveQL sync path now releases its Hive
session when the sync tool is closed, so its scratch directories are reclaimed instead of being left
until the JVM exits.
Risk Level
low
Verification:
TestHiveSyncToolpasses in full, 275 tests, which exercises the HiveQL pathend-to-end against the embedded metastore, plus
TestHiveDriverPool,TestHiveQueryDDLExecutorFailures,TestHoodieHiveSyncClientCloseand the new tests. The sessionre-assert was checked by removing it and confirming the new test fails with
SessionState.get()returning null, which is the same NPE in
Driver.compile()that the full suite surfaced in teardownbefore the re-assert was added.
Documentation Update
none
Contributor's checklist