Skip to content

[SPARK-52968][SS] Emit additional state store metrics #51679

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wants to merge 16 commits into from
Closed

[SPARK-52968][SS] Emit additional state store metrics #51679

Changes from 1 commit
Commits
Show all changes
16 commits
Select commit Hold shift + click to select a range
4018608
Emit additional state store metrics
Jul 28, 2025
d9cdeaf
PR fixes
Jul 28, 2025
86dc65c
Fix
Jul 28, 2025
4878bf0
Merge branch 'apache:master' into SPARK-52968
dylanwong250 Jul 29, 2025
97bf9ac
Some PR fixes
Jul 29, 2025
dc51917
Merge branch 'master' into SPARK-52968
Jul 29, 2025
c4f256b
Fix flakey test
Jul 29, 2025
3b99287
Only snapshot commit metrics when there is a upload in the commit
Jul 29, 2025
76c2c35
PR fixes
Jul 30, 2025
5fab7d4
Merge branch 'apache:master' into SPARK-52968
dylanwong250 Jul 30, 2025
f9d76b4
Remove clearing metrics
Jul 30, 2025
72ba026
Remove metrics refresh
Jul 30, 2025
a91d851
Fix test method
Jul 31, 2025
8590307
Fix commit metrics bug
Jul 31, 2025
913b6c1
style fix
Aug 1, 2025
bae64c4
Merge branch 'master' into SPARK-52968
Aug 1, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
Fix commit metrics bug
  • Loading branch information
Dylan Wong committed Jul 31, 2025
commit 85903071d6fc732fa259989b1400ef84f42df87a
3 changes: 3 additions & 0 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1045,6 +1045,9 @@ class RocksDB(
* - Sync the checkpoint dir files to DFS
*/
def commit(): (Long, StateStoreCheckpointInfo) = {
// Reset the commit metrics before commit
commitLatencyMs.clear()

val newVersion = loadedVersion + 1
try {
logInfo(log"Flushing updates for ${MDC(LogKeys.VERSION_NUM, newVersion)}")
Expand Down