hangc0276 commented on a change in pull request #12523:
URL: https://github.com/apache/pulsar/pull/12523#discussion_r741572579
##########
File path:
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
##########
@@ -1065,6 +1066,12 @@ public SubscriptionStatsImpl getStats(Boolean
getPreciseBacklog, boolean subscri
subStats.backlogSize = ((ManagedLedgerImpl)
topic.getManagedLedger())
.getEstimatedBacklogSize((PositionImpl)
cursor.getMarkDeletedPosition());
}
+ if (getEarliestTimeInBacklog && subStats.msgBacklog > 0) {
+ ManagedLedgerImpl managedLedger = ((ManagedLedgerImpl)
cursor.getManagedLedger());
+ PositionImpl markDeletedPosition = (PositionImpl)
cursor.getMarkDeletedPosition();
+ long result =
managedLedger.getEarliestMessagePublishTimeOfPos(markDeletedPosition);
+ subStats.timeBacklogInMills = result == 0 ? 0 :
System.currentTimeMillis() - result;
Review comment:
Why not just use the publish timestamp, otherwise user should calculate
the earliest message timestamp.
##########
File path:
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
##########
@@ -1141,6 +1143,48 @@ public long getEstimatedBacklogSize() {
}
}
+ @Override
+ public long getEarliestMessagePublishTimeInBacklog() {
+ PositionImpl pos = getMarkDeletePositionOfSlowestConsumer();
+
+ return getEarliestMessagePublishTimeOfPos(pos);
+ }
+
+ public long getEarliestMessagePublishTimeOfPos(PositionImpl pos) {
+ if (pos == null) {
+ return 0L;
+ }
+ PositionImpl nextPos = getNextValidPosition(pos);
+
+ CompletableFuture<Long> future = new CompletableFuture<>();
+ asyncReadEntry(nextPos, new ReadEntryCallback() {
+ @Override
+ public void readEntryComplete(Entry entry, Object ctx) {
Review comment:
We'd better check whether `broker_timestamp` has been set first, the
priority of the timestamp is `broker_timestamp` > `publish_time` > `0`
--
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]
