Improve commit metrics, remove currentConsumerId, and add missing docs

- Fix lastCommittedSnapshotId to default to -1 and read actual snapshot
  ID post-commit instead of querying snapshotManager
- Remove currentConsumerId metric from ConsumerProgressCalculator and
  ContinuousFileSplitEnumerator as lastScannedSnapshotId already exists
- Add lastScannedSnapshotId and lastCommittedSnapshotId to metrics docs

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: 2 people

docs/content/maintenance/metrics.md

@@ -62,6 +62,11 @@ Below is lists of Paimon built-in metrics. They are summarized into types of sca
             <td>Histogram</td>
             <td>Distributions of the time taken by the last few scans.</td>
         </tr>
+        <tr>
+            <td>lastScannedSnapshotId</td>
+            <td>Gauge</td>
+            <td>The snapshot ID scanned in the last scan. 0 if no scan has occurred.</td>
+        </tr>
         <tr>
             <td>lastScannedManifests</td>
             <td>Gauge</td>
@@ -181,6 +186,11 @@ Below is lists of Paimon built-in metrics. They are summarized into types of sca
             <td>Gauge</td>
             <td>Total size of the output files for the last compaction.</td>
         </tr>
+        <tr>
+            <td>lastCommittedSnapshotId</td>
+            <td>Gauge</td>
+            <td>The snapshot ID created by the last commit. -1 if no commit has occurred.</td>
+        </tr>
     </tbody>
 </table>
 

paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java

@@ -155,6 +155,7 @@ public class FileStoreCommitImpl implements FileStoreCommit {
     private boolean ignoreEmptyCommit;
     private CommitMetrics commitMetrics;
     private boolean appendCommitCheckConflict = false;
+    private long lastCommittedSnapshotId = -1L;
 
     public FileStoreCommitImpl(
             SnapshotCommit snapshotCommit,
@@ -360,7 +361,6 @@ public int commit(ManifestCommittable committable, boolean checkAppendFiles) {
                     tableName,
                     commitDuration);
             if (this.commitMetrics != null) {
-                Long latestSnapshotId = snapshotManager.latestSnapshotId();
                 reportCommit(
                         changes.appendTableFiles,
                         changes.appendChangelog,
@@ -369,7 +369,7 @@ public int commit(ManifestCommittable committable, boolean checkAppendFiles) {
                         commitDuration,
                         generatedSnapshot,
                         attempts,
-                        latestSnapshotId == null ? 0L : latestSnapshotId);
+                        lastCommittedSnapshotId);
             }
         }
         return generatedSnapshot;
@@ -528,7 +528,6 @@ public int overwritePartition(
             long commitDuration = (System.nanoTime() - started) / 1_000_000;
             LOG.info("Finished overwrite to table {}, duration {} ms", tableName, commitDuration);
             if (this.commitMetrics != null) {
-                Long latestSnapshotId = snapshotManager.latestSnapshotId();
                 reportCommit(
                         changes.appendTableFiles,
                         emptyList(),
@@ -537,7 +536,7 @@ public int overwritePartition(
                         commitDuration,
                         generatedSnapshot,
                         attempts,
-                        latestSnapshotId == null ? 0L : latestSnapshotId);
+                        lastCommittedSnapshotId);
             }
         }
         return generatedSnapshot;
@@ -1044,6 +1043,7 @@ CommitResult tryCommitOnce(
         if (strictModeChecker != null) {
             strictModeChecker.update(newSnapshotId);
         }
+        lastCommittedSnapshotId = newSnapshotId;
         CommitCallback.Context context =
                 new CommitCallback.Context(
                         finalBaseFiles, finalDeltaFiles, indexFiles, newSnapshot, identifier);

paimon-core/src/main/java/org/apache/paimon/operation/metrics/CommitMetrics.java

@@ -129,7 +129,7 @@ private void registerGenericCommitMetrics() {
                 () -> latestCommit == null ? 0L : latestCommit.getCompactionOutputFileSize());
         metricGroup.gauge(
                 LAST_COMMITTED_SNAPSHOT_ID,
-                () -> latestCommit == null ? 0L : latestCommit.getLastCommittedSnapshotId());
+                () -> latestCommit == null ? -1L : latestCommit.getLastCommittedSnapshotId());
     }
 
     public void reportCommit(CommitStats commitStats) {

paimon-core/src/test/java/org/apache/paimon/operation/metrics/CommitMetricsTest.java

@@ -121,7 +121,7 @@ public void testMetricsAreUpdated() {
         assertThat(lastChangelogRecordsAppended.getValue()).isEqualTo(0);
         assertThat(lastDeltaRecordsCompacted.getValue()).isEqualTo(0);
         assertThat(lastChangelogRecordsCompacted.getValue()).isEqualTo(0);
-        assertThat(lastCommittedSnapshotId.getValue()).isEqualTo(0);
+        assertThat(lastCommittedSnapshotId.getValue()).isEqualTo(-1);
 
         // report once
         reportOnce(commitMetrics);

paimon-core/src/test/java/org/apache/paimon/operation/metrics/CommitStatsTest.java

@@ -79,7 +79,7 @@ public void testFailedAppendSnapshot() {
                         0,
                         0,
                         1,
-                        0L);
+                        -1L);
         assertThat(commitStats.getTableFilesAdded()).isEqualTo(0);
         assertThat(commitStats.getTableFilesDeleted()).isEqualTo(0);
         assertThat(commitStats.getTableFilesAppended()).isEqualTo(0);
@@ -95,7 +95,7 @@ public void testFailedAppendSnapshot() {
         assertThat(commitStats.getNumBucketsWritten()).isEqualTo(0);
         assertThat(commitStats.getDuration()).isEqualTo(0);
         assertThat(commitStats.getAttempts()).isEqualTo(1);
-        assertThat(commitStats.getLastCommittedSnapshotId()).isEqualTo(0);
+        assertThat(commitStats.getLastCommittedSnapshotId()).isEqualTo(-1);
     }
 
     @Test

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/ConsumerProgressCalculator.java

@@ -37,8 +37,6 @@ public class ConsumerProgressCalculator {
 
     private final Map<Integer, Long> consumingSnapshotPerReader;
 
-    private long currentConsumerId = 0L;
-
     public ConsumerProgressCalculator(int parallelism) {
         this.minNextSnapshotPerCheckpoint = new TreeMap<>();
         this.assignedSnapshotPerReader = new HashMap<>(parallelism);
@@ -61,10 +59,8 @@ public void notifySnapshotState(
             int parallelism) {
         computeMinNextSnapshotId(readersAwaitingSplit, unassignedCalculationFunction, parallelism)
                 .ifPresent(
-                        minNextSnapshotId -> {
-                            minNextSnapshotPerCheckpoint.put(checkpointId, minNextSnapshotId);
-                            currentConsumerId = minNextSnapshotId;
-                        });
+                        minNextSnapshotId ->
+                                minNextSnapshotPerCheckpoint.put(checkpointId, minNextSnapshotId));
     }
 
     public OptionalLong notifyCheckpointComplete(long checkpointId) {
@@ -75,10 +71,6 @@ public OptionalLong notifyCheckpointComplete(long checkpointId) {
         return max;
     }
 
-    public long getCurrentConsumerId() {
-        return currentConsumerId;
-    }
-
     /** Calculate the minimum snapshot currently being consumed by all readers. */
     private Optional<Long> computeMinNextSnapshotId(
             Set<Integer> readersAwaitingSplit,

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/ContinuousFileSplitEnumerator.java

@@ -100,8 +100,6 @@ public class ContinuousFileSplitEnumerator
      */
     public static final String SOURCE_PARALLELISM_UPPER_BOUND = "sourceParallelismUpperBound";
 
-    public static final String CURRENT_CONSUMER_ID = "currentConsumerId";
-
     public ContinuousFileSplitEnumerator(
             SplitEnumeratorContext<FileStoreSourceSplit> context,
             Collection<FileStoreSourceSplit> remainSplits,
@@ -156,8 +154,6 @@ private void registerMetrics() {
         try {
             context.metricGroup()
                     .gauge(SOURCE_PARALLELISM_UPPER_BOUND, () -> sourceParallelismUpperBound);
-            context.metricGroup()
-                    .gauge(CURRENT_CONSUMER_ID, consumerProgressCalculator::getCurrentConsumerId);
         } catch (Exception e) {
             LOG.warn("Failed to register enumerator metrics.", e);
         }