This is an automated email from the ASF dual-hosted git repository.
luoyuxia pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new 671d2bbba [lake/tiering] Add pendingRecordLag metric for lake tiering
(#3469)
671d2bbba is described below
commit 671d2bbbacf16dfc54cdc46e74e3dc6d9eb43716
Author: Junbo Wang <[email protected]>
AuthorDate: Fri Jul 10 10:19:44 2026 +0800
[lake/tiering] Add pendingRecordLag metric for lake tiering (#3469)
---
.../java/org/apache/fluss/metrics/MetricNames.java | 1 +
.../org/apache/fluss/server/log/LogTablet.java | 82 ++++++++++++++++++++++
.../org/apache/fluss/server/replica/Replica.java | 9 ++-
.../replica/NotifyReplicaLakeTableOffsetTest.java | 40 +++++++++++
.../apache/fluss/server/replica/ReplicaTest.java | 64 +++++++++++++++++
.../maintenance/observability/monitor-metrics.md | 7 +-
6 files changed, 197 insertions(+), 6 deletions(-)
diff --git
a/fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java
b/fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java
index f56ca967c..dc93a5e3d 100644
--- a/fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java
+++ b/fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java
@@ -231,6 +231,7 @@ public class MetricNames {
// for tablet
public static final String LAKE_PENDING_RECORDS = "pendingRecords";
+ public static final String LAKE_PENDING_RECORDS_LAG = "pendingRecordsLag";
// for log tablet
public static final String LOG_NUM_SEGMENTS = "numSegments";
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/log/LogTablet.java
b/fluss-server/src/main/java/org/apache/fluss/server/log/LogTablet.java
index 23c60c82e..150979d4c 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/log/LogTablet.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/log/LogTablet.java
@@ -132,6 +132,9 @@ public final class LogTablet {
private volatile long lakeLogStartOffset = Long.MAX_VALUE;
private volatile long lakeLogEndOffset = -1L;
private volatile long lakeMaxTimestamp = -1;
+ // Best-effort estimate under concurrent high-watermark and lake-progress
updates.
+ // Metric reads are allowed to observe transient intermediate states.
+ private volatile long estimatedPendingStartTimeMs = -1L;
private LogTablet(
File dataDir,
@@ -277,6 +280,34 @@ public final class LogTablet {
return lakeMaxTimestamp;
}
+ /**
+ * Returns the timestamp lag between the latest local log record and the
latest tiered lake log
+ * record.
+ */
+ public long getTimestampLag() {
+ return lakeMaxTimestamp < 0L ? -1L : localMaxTimestamp() -
lakeMaxTimestamp;
+ }
+
+ /**
+ * Returns the elapsed time, in milliseconds, since the oldest committed
record in this bucket
+ * became pending for lake tiering.
+ *
+ * <p>Returns 0 when there are no committed records pending lake tiering
or when the pending
+ * start time estimate is not initialized yet.
+ */
+ public long getPendingRecordsLag(long currentTimeMs) {
+ if (estimatedPendingStartTimeMs < 0L ||
!hasPendingLakeTieringRecords()) {
+ return 0L;
+ }
+ return Math.max(0L, currentTimeMs - estimatedPendingStartTimeMs);
+ }
+
+ /** Returns the estimated start time of the current pending lake-tiering
backlog for tests. */
+ @VisibleForTesting
+ public long getEstimatedPendingStartTimeMs() {
+ return estimatedPendingStartTimeMs;
+ }
+
public int getWriterIdCount() {
return writerStateManager.writerIdCount();
}
@@ -482,6 +513,7 @@ public final class LogTablet {
if (newHighWatermark.getMessageOffset() < 0) {
throw new IllegalArgumentException("High watermark offset should
be non-negative");
}
+ long previousHighWatermark = highWatermarkMetadata.getMessageOffset();
synchronized (lock) {
if (newHighWatermark.getMessageOffset() <
highWatermarkMetadata.getMessageOffset()) {
LOG.warn(
@@ -493,6 +525,7 @@ public final class LogTablet {
highWatermarkMetadata = newHighWatermark;
// TODO log offset listener to update log offset.
}
+ onHighWatermarkUpdated(previousHighWatermark,
newHighWatermark.getMessageOffset());
LOG.trace(
"Setting high watermark {} for bucket {}",
newHighWatermark,
@@ -599,13 +632,62 @@ public final class LogTablet {
public void updateLakeLogEndOffset(long lakeLogEndOffset) {
if (lakeLogEndOffset > this.lakeLogEndOffset) {
this.lakeLogEndOffset = lakeLogEndOffset;
+ // Lake-tiering progress advanced via the end offset; re-estimate
the pending start
+ // time so the lag is corrected (and cleared once caught up) even
when the lake max
+ // timestamp is not updated in the same notification.
+ onLakeProgressUpdated();
}
}
public void updateLakeMaxTimestamp(long lakeMaxTimestamp) {
if (lakeMaxTimestamp > this.lakeMaxTimestamp) {
this.lakeMaxTimestamp = lakeMaxTimestamp;
+ onLakeProgressUpdated();
+ }
+ }
+
+ private void onHighWatermarkUpdated(long previousHighWatermark, long
currentHighWatermark) {
+ if (!isDataLakeEnabled) {
+ return;
+ }
+ long firstPendingOffset = Math.max(lakeLogEndOffset,
localLogStartOffset());
+ boolean hadPendingBeforeUpdate = firstPendingOffset <
previousHighWatermark;
+ boolean hasPendingAfterUpdate = firstPendingOffset <
currentHighWatermark;
+ if (!hasPendingAfterUpdate) {
+ clearPendingStartTime();
+ } else if (!hadPendingBeforeUpdate) {
+ markPendingStartTimeNow();
+ }
+ }
+
+ private void onLakeProgressUpdated() {
+ if (!hasPendingLakeTieringRecords()) {
+ clearPendingStartTime();
+ return;
+ }
+ long timestampLag = getTimestampLag();
+ if (timestampLag < 0L) {
+ return;
}
+ long candidatePendingStartTimeMs = Math.max(0L, clock.milliseconds() -
timestampLag);
+ advancePendingStartTime(candidatePendingStartTimeMs);
+ }
+
+ private boolean hasPendingLakeTieringRecords() {
+ return Math.max(lakeLogEndOffset, localLogStartOffset()) <
getHighWatermark();
+ }
+
+ private void clearPendingStartTime() {
+ estimatedPendingStartTimeMs = -1L;
+ }
+
+ private void markPendingStartTimeNow() {
+ estimatedPendingStartTimeMs = clock.milliseconds();
+ }
+
+ private void advancePendingStartTime(long candidatePendingStartTimeMs) {
+ estimatedPendingStartTimeMs =
+ Math.max(estimatedPendingStartTimeMs,
candidatePendingStartTimeMs);
}
public void loadWriterSnapshot(long lastOffset) throws IOException {
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java
b/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java
index e11dd69c1..1b0407e81 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java
@@ -598,11 +598,10 @@ public final class Replica {
return getLogHighWatermark() - lakeLogEndOffset;
});
lakeTieringMetricGroup.gauge(
- MetricNames.LOG_LAKE_TIMESTAMP_LAG,
- () ->
- logTablet.getLakeMaxTimestamp() < 0L
- ? -1
- : logTablet.localMaxTimestamp() -
logTablet.getLakeMaxTimestamp());
+ MetricNames.LOG_LAKE_TIMESTAMP_LAG,
logTablet::getTimestampLag);
+ lakeTieringMetricGroup.gauge(
+ MetricNames.LAKE_PENDING_RECORDS_LAG,
+ () -> logTablet.getPendingRecordsLag(clock.milliseconds()));
}
private void onBecomeNewFollower(int standbyReplica) {
diff --git
a/fluss-server/src/test/java/org/apache/fluss/server/replica/NotifyReplicaLakeTableOffsetTest.java
b/fluss-server/src/test/java/org/apache/fluss/server/replica/NotifyReplicaLakeTableOffsetTest.java
index 0d0b25204..8b5795644 100644
---
a/fluss-server/src/test/java/org/apache/fluss/server/replica/NotifyReplicaLakeTableOffsetTest.java
+++
b/fluss-server/src/test/java/org/apache/fluss/server/replica/NotifyReplicaLakeTableOffsetTest.java
@@ -18,18 +18,27 @@
package org.apache.fluss.server.replica;
import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.record.MemoryLogRecords;
import org.apache.fluss.rpc.messages.NotifyLakeTableOffsetResponse;
import org.apache.fluss.server.entity.LakeBucketOffset;
import org.apache.fluss.server.entity.NotifyLakeTableOffsetData;
import org.assertj.core.api.AssertionsForClassTypes;
+import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
+import java.time.Duration;
import java.util.Collections;
import java.util.concurrent.CompletableFuture;
+import static org.apache.fluss.record.LogRecordBatch.CURRENT_LOG_MAGIC_VALUE;
+import static org.apache.fluss.record.TestData.DATA1;
+import static org.apache.fluss.record.TestData.DATA1_ROW_TYPE;
import static org.apache.fluss.record.TestData.DATA1_TABLE_ID;
+import static org.apache.fluss.record.TestData.DEFAULT_SCHEMA_ID;
+import static
org.apache.fluss.testutils.DataTestUtils.createRecordsWithoutBaseLogOffset;
+import static org.assertj.core.api.Assertions.assertThat;
/** Test for notify replica lakehouse data info. */
class NotifyReplicaLakeTableOffsetTest extends ReplicaTestBase {
@@ -48,6 +57,37 @@ class NotifyReplicaLakeTableOffsetTest extends
ReplicaTestBase {
notifyAndVerify(tb, replica, 2, 20L, 30L, System.currentTimeMillis());
}
+ @Test
+ void testNotifyLakeTableOffsetClearsPendingRecordsLagWhenCaughtUp() throws
Exception {
+ TableBucket tb = makeTableBucket(false);
+ makeLogTableAsLeader(tb, false);
+ Replica replica = replicaManager.getReplicaOrException(tb);
+ replica.updateIsDataLakeEnabled(true);
+
+ long initialTimestamp = manualClock.milliseconds();
+ MemoryLogRecords records =
+ createRecordsWithoutBaseLogOffset(
+ DATA1_ROW_TYPE,
+ DEFAULT_SCHEMA_ID,
+ 0,
+ initialTimestamp,
+ CURRENT_LOG_MAGIC_VALUE,
+ DATA1,
+ replica.getLogFormat());
+ replica.appendRecordsToLeader(records, 0);
+
replica.getLogTablet().updateHighWatermark(replica.getLocalLogEndOffset());
+
+ manualClock.advanceTime(Duration.ofSeconds(30));
+
assertThat(replica.getLogTablet().getPendingRecordsLag(manualClock.milliseconds()))
+ .isEqualTo(Duration.ofSeconds(30).toMillis());
+
+ notifyAndVerify(tb, replica, 1, 0L, replica.getLocalLogEndOffset(),
initialTimestamp);
+
+
assertThat(replica.getLogTablet().getPendingRecordsLag(manualClock.milliseconds()))
+ .isZero();
+
assertThat(replica.getLogTablet().getEstimatedPendingStartTimeMs()).isEqualTo(-1L);
+ }
+
private void notifyAndVerify(
TableBucket tb,
Replica replica,
diff --git
a/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTest.java
b/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTest.java
index 4eefc0e2f..d37e91dc6 100644
---
a/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTest.java
+++
b/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTest.java
@@ -25,6 +25,10 @@ import org.apache.fluss.metadata.SchemaGetter;
import org.apache.fluss.metadata.SchemaInfo;
import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.metrics.Gauge;
+import org.apache.fluss.metrics.MetricNames;
+import org.apache.fluss.metrics.groups.AbstractMetricGroup;
+import org.apache.fluss.metrics.groups.MetricGroup;
import org.apache.fluss.record.ChangeType;
import org.apache.fluss.record.KvRecordBatch;
import org.apache.fluss.record.KvRecordTestUtils;
@@ -748,6 +752,57 @@ final class ReplicaTest extends ReplicaTestBase {
verifyGetKeyValues(kvTablet, expectedKeyValues);
}
+ @Test
+ void testPendingRecordsLagStartsFromNewBacklogAfterIdle() throws Exception
{
+ Replica logReplica =
+ makeLogReplica(DATA1_PHYSICAL_TABLE_PATH, new
TableBucket(DATA1_TABLE_ID, 1));
+ makeLogReplicaAsLeader(logReplica);
+ logReplica.updateIsDataLakeEnabled(true);
+
+ long initialTimestamp = manualClock.milliseconds();
+ MemoryLogRecords firstBatch =
+ createRecordsWithoutBaseLogOffset(
+ DATA1_ROW_TYPE,
+ DEFAULT_SCHEMA_ID,
+ 0,
+ initialTimestamp,
+ CURRENT_LOG_MAGIC_VALUE,
+ DATA1,
+ LogFormat.ARROW);
+ logReplica.appendRecordsToLeader(firstBatch, 0);
+
logReplica.getLogTablet().updateHighWatermark(logReplica.getLocalLogEndOffset());
+
logReplica.getLogTablet().updateLakeLogEndOffset(logReplica.getLocalLogEndOffset());
+ logReplica.getLogTablet().updateLakeMaxTimestamp(initialTimestamp);
+
+ assertThat(getLakeTieringGaugeValue(logReplica,
MetricNames.LAKE_PENDING_RECORDS_LAG))
+ .isZero();
+
+ manualClock.advanceTime(Duration.ofMinutes(10));
+ long newBatchTimestamp = manualClock.milliseconds();
+ MemoryLogRecords secondBatch =
+ createRecordsWithoutBaseLogOffset(
+ DATA1_ROW_TYPE,
+ DEFAULT_SCHEMA_ID,
+ 0,
+ newBatchTimestamp,
+ CURRENT_LOG_MAGIC_VALUE,
+ DATA1,
+ LogFormat.ARROW);
+ logReplica.appendRecordsToLeader(secondBatch, 0);
+
logReplica.getLogTablet().updateHighWatermark(logReplica.getLocalLogEndOffset());
+
+ assertThat(getLakeTieringGaugeValue(logReplica,
MetricNames.LOG_LAKE_TIMESTAMP_LAG))
+ .isEqualTo(Duration.ofMinutes(10).toMillis());
+ assertThat(getLakeTieringGaugeValue(logReplica,
MetricNames.LAKE_PENDING_RECORDS_LAG))
+ .isZero();
+ assertThat(logReplica.getLogTablet().getEstimatedPendingStartTimeMs())
+ .isEqualTo(manualClock.milliseconds());
+
+ manualClock.advanceTime(Duration.ofSeconds(5));
+ assertThat(getLakeTieringGaugeValue(logReplica,
MetricNames.LAKE_PENDING_RECORDS_LAG))
+ .isEqualTo(Duration.ofSeconds(5).toMillis());
+ }
+
@Test
void testUpdateIsDataLakeEnabled() throws Exception {
Replica logReplica =
@@ -770,6 +825,15 @@ final class ReplicaTest extends ReplicaTestBase {
assertThat(logReplica.getLogTablet().isDataLakeEnabled()).isFalse();
}
+ @SuppressWarnings("unchecked")
+ private long getLakeTieringGaugeValue(Replica replica, String metricName) {
+ MetricGroup lakeTieringMetricGroup =
replica.bucketMetrics().addGroup("lakeTiering");
+ Gauge<Long> gauge =
+ (Gauge<Long>)
+ ((AbstractMetricGroup)
lakeTieringMetricGroup).getMetrics().get(metricName);
+ return gauge.getValue();
+ }
+
private void makeLogReplicaAsLeader(Replica replica) throws Exception {
makeLeaderReplica(
replica,
diff --git a/website/docs/maintenance/observability/monitor-metrics.md
b/website/docs/maintenance/observability/monitor-metrics.md
index 5383250b8..50ae66ea3 100644
--- a/website/docs/maintenance/observability/monitor-metrics.md
+++ b/website/docs/maintenance/observability/monitor-metrics.md
@@ -880,7 +880,7 @@ Some metrics might not be exposed when using other JVM
implementations (e.g. IBM
<td>Gauge</td>
</tr>
<tr>
- <td rowspan="2">table_bucket_lakeTiering</td>
+ <td rowspan="3">table_bucket_lakeTiering</td>
<td>pendingRecords</td>
<td>The number of records lag between the latest log record and the
latest tiered lake log record for this table bucket. Returns -1 if row count is
disabled (WAL mode or v0.9 old table) and no tiering has completed.</td>
<td>Gauge</td>
@@ -890,6 +890,11 @@ Some metrics might not be exposed when using other JVM
implementations (e.g. IBM
<td>The timestamp lag between the latest log record and the latest
tiered lake log record for this table bucket, in milliseconds.</td>
<td>Gauge</td>
</tr>
+ <tr>
+ <td>pendingRecordsLag</td>
+ <td>The elapsed time, in milliseconds, since the oldest committed record
in this table bucket became pending for lake tiering. Returns 0 when no
committed records are pending lake tiering.</td>
+ <td>Gauge</td>
+ </tr>
<tr>
<td rowspan="3">table_bucket_remoteLog</td>
<td>numSegments</td>
