luoyuxia commented on code in PR #3469:
URL: https://github.com/apache/fluss/pull/3469#discussion_r3506251823
##########
fluss-server/src/main/java/org/apache/fluss/server/log/LogTablet.java:
##########
@@ -599,15 +631,63 @@ public void updateLakeLogStartOffset(long
lakeHouseLogStartOffset) {
public void updateLakeLogEndOffset(long lakeLogEndOffset) {
if (lakeLogEndOffset > this.lakeLogEndOffset) {
this.lakeLogEndOffset = lakeLogEndOffset;
+ if (!hasPendingLakeBacklog()) {
+ clearPendingStartTime();
+ }
}
}
public void updateLakeMaxTimestamp(long lakeMaxTimestamp) {
if (lakeMaxTimestamp > this.lakeMaxTimestamp) {
this.lakeMaxTimestamp = lakeMaxTimestamp;
+ onLakeProgressUpdated();
}
}
+ private void onHighWatermarkAdvanced(boolean hadPendingRecords) {
Review Comment:
actually, I do spend some time to fingure out it. I'm wondering whether the
following is better:
```
private void onHighWatermarkUpdated(long previousHighWatermark, long
currentHighWatermark) {
long firstPendingOffset = Math.max(lakeLogEndOffset,
localLogStartOffset());
boolean hadPendingBeforeUpdate = firstPendingOffset <
previousHighWatermark;
boolean hasPendingAfterUpdate = firstPendingOffset <
currentHighWatermark;
if (!hasPendingAfterUpdate) {
clearPendingStartTime();
} else if (!hadPendingBeforeUpdate) {
markPendingStartTimeNow();
}
}
```
##########
fluss-server/src/main/java/org/apache/fluss/server/log/LogTablet.java:
##########
@@ -599,15 +631,63 @@ public void updateLakeLogStartOffset(long
lakeHouseLogStartOffset) {
public void updateLakeLogEndOffset(long lakeLogEndOffset) {
if (lakeLogEndOffset > this.lakeLogEndOffset) {
this.lakeLogEndOffset = lakeLogEndOffset;
+ if (!hasPendingLakeBacklog()) {
+ clearPendingStartTime();
+ }
}
}
public void updateLakeMaxTimestamp(long lakeMaxTimestamp) {
if (lakeMaxTimestamp > this.lakeMaxTimestamp) {
this.lakeMaxTimestamp = lakeMaxTimestamp;
+ onLakeProgressUpdated();
}
}
+ private void onHighWatermarkAdvanced(boolean hadPendingRecords) {
Review Comment:
is there any diffeernce between `hadPendingRecords` and `hasPendingRecords`
##########
fluss-server/src/main/java/org/apache/fluss/server/log/LogTablet.java:
##########
@@ -599,15 +631,63 @@ public void updateLakeLogStartOffset(long
lakeHouseLogStartOffset) {
public void updateLakeLogEndOffset(long lakeLogEndOffset) {
if (lakeLogEndOffset > this.lakeLogEndOffset) {
this.lakeLogEndOffset = lakeLogEndOffset;
+ if (!hasPendingLakeBacklog()) {
+ clearPendingStartTime();
+ }
}
}
public void updateLakeMaxTimestamp(long lakeMaxTimestamp) {
if (lakeMaxTimestamp > this.lakeMaxTimestamp) {
this.lakeMaxTimestamp = lakeMaxTimestamp;
+ onLakeProgressUpdated();
}
}
+ private void onHighWatermarkAdvanced(boolean hadPendingRecords) {
+ boolean hasPendingRecords = hasPendingLakeBacklog();
+ if (!hasPendingRecords) {
+ clearPendingStartTime();
+ } else if (!hadPendingRecords) {
+ markPendingStartTimeNow();
+ }
+ }
+
+ private void onLakeProgressUpdated() {
+ if (!hasPendingLakeBacklog()) {
+ clearPendingStartTime();
+ return;
+ }
+ long timestampLag = getTimestampLag();
+ if (timestampLag < 0L) {
+ return;
+ }
+ long candidatePendingStartTimeMs =
+ Math.max(0L, clock.milliseconds() - Math.max(0L,
timestampLag));
+ advancePendingStartTime(candidatePendingStartTimeMs);
+ }
+
+ private boolean hasPendingLakeBacklog() {
Review Comment:
hasPendingLakeBacklog -> hasPendingLakeTieringRecords
##########
fluss-server/src/main/java/org/apache/fluss/server/log/LogTablet.java:
##########
@@ -599,15 +631,63 @@ public void updateLakeLogStartOffset(long
lakeHouseLogStartOffset) {
public void updateLakeLogEndOffset(long lakeLogEndOffset) {
if (lakeLogEndOffset > this.lakeLogEndOffset) {
this.lakeLogEndOffset = lakeLogEndOffset;
+ if (!hasPendingLakeBacklog()) {
+ clearPendingStartTime();
+ }
}
}
public void updateLakeMaxTimestamp(long lakeMaxTimestamp) {
if (lakeMaxTimestamp > this.lakeMaxTimestamp) {
this.lakeMaxTimestamp = lakeMaxTimestamp;
+ onLakeProgressUpdated();
}
}
+ private void onHighWatermarkAdvanced(boolean hadPendingRecords) {
+ boolean hasPendingRecords = hasPendingLakeBacklog();
+ if (!hasPendingRecords) {
+ clearPendingStartTime();
+ } else if (!hadPendingRecords) {
+ markPendingStartTimeNow();
+ }
+ }
+
+ private void onLakeProgressUpdated() {
+ if (!hasPendingLakeBacklog()) {
+ clearPendingStartTime();
+ return;
+ }
+ long timestampLag = getTimestampLag();
+ if (timestampLag < 0L) {
+ return;
+ }
+ long candidatePendingStartTimeMs =
Review Comment:
nit: Result of 'max' is the same as the second argument making the call
meaningless
##########
website/docs/maintenance/observability/monitor-metrics.md:
##########
@@ -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 backlog-aware lag for committed records pending lake tiering for
this table bucket, in milliseconds. It is 0 when there are no committed records
pending lake tiering.</td>
Review Comment:
actually, for me, the backlog-aware is not intuitive.. Some thing like this:
```
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.
```
?
##########
fluss-server/src/main/java/org/apache/fluss/server/log/LogTablet.java:
##########
@@ -599,15 +631,63 @@ public void updateLakeLogStartOffset(long
lakeHouseLogStartOffset) {
public void updateLakeLogEndOffset(long lakeLogEndOffset) {
if (lakeLogEndOffset > this.lakeLogEndOffset) {
this.lakeLogEndOffset = lakeLogEndOffset;
+ if (!hasPendingLakeBacklog()) {
+ clearPendingStartTime();
+ }
}
}
public void updateLakeMaxTimestamp(long lakeMaxTimestamp) {
if (lakeMaxTimestamp > this.lakeMaxTimestamp) {
this.lakeMaxTimestamp = lakeMaxTimestamp;
+ onLakeProgressUpdated();
}
}
+ private void onHighWatermarkAdvanced(boolean hadPendingRecords) {
+ boolean hasPendingRecords = hasPendingLakeBacklog();
Review Comment:
nit: we can skip handle pendingRecords for non-tablelake table
##########
fluss-server/src/main/java/org/apache/fluss/server/log/LogTablet.java:
##########
@@ -599,15 +631,63 @@ public void updateLakeLogStartOffset(long
lakeHouseLogStartOffset) {
public void updateLakeLogEndOffset(long lakeLogEndOffset) {
if (lakeLogEndOffset > this.lakeLogEndOffset) {
this.lakeLogEndOffset = lakeLogEndOffset;
+ if (!hasPendingLakeBacklog()) {
+ clearPendingStartTime();
+ }
}
}
public void updateLakeMaxTimestamp(long lakeMaxTimestamp) {
if (lakeMaxTimestamp > this.lakeMaxTimestamp) {
this.lakeMaxTimestamp = lakeMaxTimestamp;
+ onLakeProgressUpdated();
}
}
+ private void onHighWatermarkAdvanced(boolean hadPendingRecords) {
+ boolean hasPendingRecords = hasPendingLakeBacklog();
+ if (!hasPendingRecords) {
+ clearPendingStartTime();
+ } else if (!hadPendingRecords) {
+ markPendingStartTimeNow();
+ }
+ }
+
+ private void onLakeProgressUpdated() {
+ if (!hasPendingLakeBacklog()) {
+ clearPendingStartTime();
+ return;
+ }
+ long timestampLag = getTimestampLag();
+ if (timestampLag < 0L) {
+ return;
+ }
+ long candidatePendingStartTimeMs =
+ Math.max(0L, clock.milliseconds() - Math.max(0L,
timestampLag));
+ advancePendingStartTime(candidatePendingStartTimeMs);
+ }
+
+ private boolean hasPendingLakeBacklog() {
+ long firstPendingOffset =
Review Comment:
can be simplified to
```
Math.max(lakeLogEndOffset, localLogStartOffset()) < getHighWatermark();
```
?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
