You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2021/10/28 16:59:16 UTC
[GitHub] [pulsar] LeBW opened a new pull request #12523: add a metric to get earliest time in the backlog
LeBW opened a new pull request #12523:
URL: https://github.com/apache/pulsar/pull/12523
<!--
### Contribution Checklist
- Name the pull request in the form "[Issue XYZ][component] Title of the pull request", where *XYZ* should be replaced by the actual issue number.
Skip *Issue XYZ* if there is no associated github issue for this pull request.
Skip *component* if you are unsure about which is the best component. E.g. `[docs] Fix typo in produce method`.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Each pull request should address only one issue, not mix up code from multiple issues.
- Each commit in the pull request has a meaningful commit message
- Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
**(The sections below can be removed for hotfixes of typos)**
-->
### Motivation
Currently, there is no metric to show the publish time of the earliest message in the backlog.
### Modifications
Add a metric to show the timestamp from `the publish time of the earliest message in the backlog` to `the current time`.
Usage:
```bash
pulsar-admin topics stats <topic-name> -etb
# The command will only compute the metric when the flag `-etb` is `true`.
```
### Verifying this change
This change added tests and can be verified as follows:
- Add unit test `AdminApiTest#testGetStats`
### Does this pull request potentially affect one of the following parts:
*If `yes` was chosen, please highlight the changes*
- Dependencies (does it add or upgrade a dependency): (no)
- The public API: (no)
- The schema: (no)
- The default values of configurations: ( no)
- The wire protocol: ( no)
- The rest endpoints: (no)
- The admin cli options: (yes)
- Anything that affects deployment: (no)
### Documentation
Check the box below and label this PR (if you have committer privilege).
Need to update docs?
- [ ] `doc-required`
(If you need help on updating docs, create a doc issue)
- [ ] `no-need-doc`
(Please explain why)
- [ ] `doc`
(If this PR contains doc changes)
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] LeBW commented on a change in pull request #12523: add a metric to get earliest time in the backlog
Posted by GitBox <gi...@apache.org>.
LeBW commented on a change in pull request #12523:
URL: https://github.com/apache/pulsar/pull/12523#discussion_r743612100
##########
File path: pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Topics.java
##########
@@ -856,8 +858,13 @@ default void delete(String topic, boolean force) throws PulsarAdminException {
* @throws PulsarAdminException
* Unexpected error
*/
- TopicStats getStats(String topic, boolean getPreciseBacklog,
- boolean subscriptionBacklogSize) throws PulsarAdminException;
+ TopicStats getStats(String topic, boolean getPreciseBacklog, boolean subscriptionBacklogSize,
Review comment:
Maybe the method below (Line 864 - 866) can resolve the compatibility issue because it's the original API with the `getEarliestTimeInBacklog` default to be `false`?
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on pull request #12523: add a metric to get earliest time in the backlog
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on pull request #12523:
URL: https://github.com/apache/pulsar/pull/12523#issuecomment-979957696
@Jason918 @hangc0276 Please help review this PR again.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] LeBW commented on a change in pull request #12523: add a metric to get earliest time in the backlog
Posted by GitBox <gi...@apache.org>.
LeBW commented on a change in pull request #12523:
URL: https://github.com/apache/pulsar/pull/12523#discussion_r748965635
##########
File path: managed-ledger/pom.xml
##########
@@ -66,6 +66,12 @@
<version>${project.version}</version>
</dependency>
+ <dependency>
Review comment:
Thanks for your suggestion. I've move the dependency part from pulsar-client-original to pulsar-common
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Jason918 commented on a change in pull request #12523: add a metric to get earliest time in the backlog
Posted by GitBox <gi...@apache.org>.
Jason918 commented on a change in pull request #12523:
URL: https://github.com/apache/pulsar/pull/12523#discussion_r758061664
##########
File path: pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/TopicStatsImpl.java
##########
@@ -76,6 +76,9 @@
/** Get estimated total unconsumed or backlog size in bytes. */
public long backlogSize;
+ /** Get the time from the oldest message in the backlog to current time. */
+ public long timeBacklogInMills;
Review comment:
Same here.
##########
File path: pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/SubscriptionStatsImpl.java
##########
@@ -55,6 +55,9 @@
/** Size of backlog in byte. **/
public long backlogSize;
+ /** Get the time from the oldest message in the backlog to current time. */
+ public long timeBacklogInMills;
Review comment:
From description, this timeBacklogInMills should be a time period as "current_timestamp"-"min_message_timestamp".
But it seems that it just set as min_message_timestamp?
And IMHO, name it as "backlogTime" would be more consistent with "backlogSize"
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] LeBW commented on a change in pull request #12523: add a metric to get earliest time in the backlog
Posted by GitBox <gi...@apache.org>.
LeBW commented on a change in pull request #12523:
URL: https://github.com/apache/pulsar/pull/12523#discussion_r741726849
##########
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:
Fixed
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] LeBW commented on pull request #12523: add a metric to get earliest time in the backlog
Posted by GitBox <gi...@apache.org>.
LeBW commented on pull request #12523:
URL: https://github.com/apache/pulsar/pull/12523#issuecomment-961588405
/pulsarbot run-failure-checks
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] gaoran10 commented on a change in pull request #12523: add a metric to get earliest time in the backlog
Posted by GitBox <gi...@apache.org>.
gaoran10 commented on a change in pull request #12523:
URL: https://github.com/apache/pulsar/pull/12523#discussion_r744285199
##########
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:
Maybe you could use the method `getEntryTimestamp()` of the class `MessageImpl`.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] gaoran10 commented on a change in pull request #12523: add a metric to get earliest time in the backlog
Posted by GitBox <gi...@apache.org>.
gaoran10 commented on a change in pull request #12523:
URL: https://github.com/apache/pulsar/pull/12523#discussion_r762787362
##########
File path: pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/SubscriptionStatsImpl.java
##########
@@ -55,6 +55,9 @@
/** Size of backlog in byte. **/
public long backlogSize;
+ /** Get the time from the oldest message in the backlog to current time. */
+ public long timeBacklogInMills;
Review comment:
fix
##########
File path: pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/TopicStatsImpl.java
##########
@@ -76,6 +76,9 @@
/** Get estimated total unconsumed or backlog size in bytes. */
public long backlogSize;
+ /** Get the time from the oldest message in the backlog to current time. */
+ public long timeBacklogInMills;
Review comment:
fix
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] github-actions[bot] commented on pull request #12523: add a metric to get earliest time in the backlog
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #12523:
URL: https://github.com/apache/pulsar/pull/12523#issuecomment-954030892
@LeBW:Thanks for your contribution. For this PR, do we need to update docs?
(The [PR template contains info about doc](https://github.com/apache/pulsar/blob/master/.github/PULL_REQUEST_TEMPLATE.md#documentation), which helps others know more about the changes. Can you provide doc-related info in this and future PR descriptions? Thanks)
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] hangc0276 commented on a change in pull request #12523: add a metric to get earliest time in the backlog
Posted by GitBox <gi...@apache.org>.
hangc0276 commented on a change in pull request #12523:
URL: https://github.com/apache/pulsar/pull/12523#discussion_r759339065
##########
File path: pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/SubscriptionStats.java
##########
@@ -49,6 +49,9 @@
/** Size of backlog in byte. **/
long getBacklogSize();
+ /** Get the time from the oldest message in the backlog to current time. */
+ long getTimeBacklogInMills();
Review comment:
Change to `getBacklogTime` is more better, and use second instead of mill seconds is more friendly for users.
##########
File path: pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/SubscriptionStatsImpl.java
##########
@@ -55,6 +55,9 @@
/** Size of backlog in byte. **/
public long backlogSize;
+ /** Get the time from the oldest message in the backlog to current time. */
+ public long timeBacklogInMills;
Review comment:
+1
##########
File path: pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java
##########
@@ -440,6 +440,17 @@ public static void skipMessageMetadata(ByteBuf buffer) {
buffer.skipBytes(metadataSize);
}
+ public static long getEntryPublishTimestamp(ByteBuf headersAndPayloadWithBrokerEntryMetadata) throws IOException {
Review comment:
change the method name to `getEntryTimestamp` is better.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] LeBW commented on pull request #12523: add a metric to get earliest time in the backlog
Posted by GitBox <gi...@apache.org>.
LeBW commented on pull request #12523:
URL: https://github.com/apache/pulsar/pull/12523#issuecomment-955934040
> Each subscription has a different mark delete position, which indicates that they have a different backlog, so the earliest publish time is different for each subscription, we need to compute the earliest publish time for every subscription.
ok, I'll add the earliest publish time for every subscription.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Jason918 commented on a change in pull request #12523: add a metric to get earliest time in the backlog
Posted by GitBox <gi...@apache.org>.
Jason918 commented on a change in pull request #12523:
URL: https://github.com/apache/pulsar/pull/12523#discussion_r746202265
##########
File path: managed-ledger/pom.xml
##########
@@ -66,6 +66,12 @@
<version>${project.version}</version>
</dependency>
+ <dependency>
Review comment:
It's a little confusing that add the pulsar-client-original dependency to managed-ledger.
Is there anyway we can move the dependent part to pulsar-common? Since we did not add client to managed-ledger right?
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on a change in pull request #12523: add a metric to get earliest time in the backlog
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on a change in pull request #12523:
URL: https://github.com/apache/pulsar/pull/12523#discussion_r742460181
##########
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:
@hangc0276 yes, it looks like a time-based lag.
##########
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:
@hangc0276 yes, it looks like a time-based lag.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] LeBW commented on a change in pull request #12523: add a metric to get earliest time in the backlog
Posted by GitBox <gi...@apache.org>.
LeBW commented on a change in pull request #12523:
URL: https://github.com/apache/pulsar/pull/12523#discussion_r741726291
##########
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:
I remember that the requirement is to compute the time from the earliest message timestamp to the current time. @codelipenghui
##########
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:
Fixed
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] LeBW commented on a change in pull request #12523: add a metric to get earliest time in the backlog
Posted by GitBox <gi...@apache.org>.
LeBW commented on a change in pull request #12523:
URL: https://github.com/apache/pulsar/pull/12523#discussion_r743604270
##########
File path: managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
##########
@@ -1141,6 +1144,53 @@ 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) {
+ ByteBuf metadataAndPayload = entry.getDataBuffer();
+ BrokerEntryMetadata brokerEntryMetadata = Commands.parseBrokerEntryMetadataIfExist(metadataAndPayload);
+ if (brokerEntryMetadata != null && brokerEntryMetadata.hasBrokerTimestamp()) {
+ future.complete(brokerEntryMetadata.getBrokerTimestamp());
+ } else {
+ MessageMetadata messageMetadata = Commands.parseMessageMetadata(metadataAndPayload);
+ if (messageMetadata.hasPublishTime()) {
+ future.complete(messageMetadata.getPublishTime());
+ } else {
+ future.complete(0L);
+ }
+ }
+ }
+
+ @Override
+ public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
+ future.completeExceptionally(exception);
+ }
+ }, null);
+
+ long result;
+ try {
+ result = future.get();
Review comment:
Even if we return a `Future` here, maybe we still need to wait the entry read somewhere else (which will also block the request thread) ?
##########
File path: pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java
##########
@@ -563,10 +563,14 @@ void run() throws PulsarAdminException {
+ ", locking required.")
private boolean subscriptionBacklogSize = false;
+ @Parameter(names = { "-etb",
+ "--get-earliest-time-in-backlog" }, description = "Set true to get earliest time in backlog")
Review comment:
Can `--get-backlog-in-mills` reflect that it will get a timestamp? Maybe it's a little confusing as it looks like getting the backlog itself. (It's just my question and I can still change it to `--get-backlog-in-mills`)
##########
File path: pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Topics.java
##########
@@ -856,8 +858,13 @@ default void delete(String topic, boolean force) throws PulsarAdminException {
* @throws PulsarAdminException
* Unexpected error
*/
- TopicStats getStats(String topic, boolean getPreciseBacklog,
- boolean subscriptionBacklogSize) throws PulsarAdminException;
+ TopicStats getStats(String topic, boolean getPreciseBacklog, boolean subscriptionBacklogSize,
Review comment:
Maybe the method below (Line 864 - 866) can resolve the compatibility issue because it's the original API with the `getEarliestTimeInBacklog` default to be `false`?
##########
File path: managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
##########
@@ -1141,6 +1144,53 @@ 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) {
+ ByteBuf metadataAndPayload = entry.getDataBuffer();
+ BrokerEntryMetadata brokerEntryMetadata = Commands.parseBrokerEntryMetadataIfExist(metadataAndPayload);
+ if (brokerEntryMetadata != null && brokerEntryMetadata.hasBrokerTimestamp()) {
+ future.complete(brokerEntryMetadata.getBrokerTimestamp());
+ } else {
+ MessageMetadata messageMetadata = Commands.parseMessageMetadata(metadataAndPayload);
+ if (messageMetadata.hasPublishTime()) {
+ future.complete(messageMetadata.getPublishTime());
+ } else {
+ future.complete(0L);
+ }
+ }
+ }
+
+ @Override
+ public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
+ future.completeExceptionally(exception);
+ }
+ }, null);
+
+ long result;
+ try {
+ result = future.get();
Review comment:
Even if we return a `Future` here, maybe we still need to wait the entry read somewhere else (which will also block the request thread) ?
##########
File path: pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java
##########
@@ -563,10 +563,14 @@ void run() throws PulsarAdminException {
+ ", locking required.")
private boolean subscriptionBacklogSize = false;
+ @Parameter(names = { "-etb",
+ "--get-earliest-time-in-backlog" }, description = "Set true to get earliest time in backlog")
Review comment:
Can `--get-backlog-in-mills` reflect that it will get a timestamp? Maybe it's a little confusing as it looks like getting the backlog itself. (It's just my question and I can still change it to `--get-backlog-in-mills`)
##########
File path: pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Topics.java
##########
@@ -856,8 +858,13 @@ default void delete(String topic, boolean force) throws PulsarAdminException {
* @throws PulsarAdminException
* Unexpected error
*/
- TopicStats getStats(String topic, boolean getPreciseBacklog,
- boolean subscriptionBacklogSize) throws PulsarAdminException;
+ TopicStats getStats(String topic, boolean getPreciseBacklog, boolean subscriptionBacklogSize,
Review comment:
Maybe the method below (Line 864 - 866) can resolve the compatibility issue because it's the original API with the `getEarliestTimeInBacklog` default to be `false`?
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on a change in pull request #12523: add a metric to get earliest time in the backlog
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on a change in pull request #12523:
URL: https://github.com/apache/pulsar/pull/12523#discussion_r743373563
##########
File path: managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
##########
@@ -1141,6 +1144,53 @@ 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) {
+ ByteBuf metadataAndPayload = entry.getDataBuffer();
+ BrokerEntryMetadata brokerEntryMetadata = Commands.parseBrokerEntryMetadataIfExist(metadataAndPayload);
+ if (brokerEntryMetadata != null && brokerEntryMetadata.hasBrokerTimestamp()) {
+ future.complete(brokerEntryMetadata.getBrokerTimestamp());
+ } else {
+ MessageMetadata messageMetadata = Commands.parseMessageMetadata(metadataAndPayload);
+ if (messageMetadata.hasPublishTime()) {
+ future.complete(messageMetadata.getPublishTime());
+ } else {
+ future.complete(0L);
+ }
+ }
+ }
+
+ @Override
+ public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
+ future.completeExceptionally(exception);
+ }
+ }, null);
+
+ long result;
+ try {
+ result = future.get();
Review comment:
Any reason to call future.get() here? We can return a Future for this method, this will block the request thread for waiting the entry read.
##########
File path: pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Topics.java
##########
@@ -856,8 +858,13 @@ default void delete(String topic, boolean force) throws PulsarAdminException {
* @throws PulsarAdminException
* Unexpected error
*/
- TopicStats getStats(String topic, boolean getPreciseBacklog,
- boolean subscriptionBacklogSize) throws PulsarAdminException;
+ TopicStats getStats(String topic, boolean getPreciseBacklog, boolean subscriptionBacklogSize,
Review comment:
We can introduce a new API TopicStats getStats(String topic, GetStatsOptions options);
So can we can introduce more params without breaking the API.
##########
File path: pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java
##########
@@ -563,10 +563,14 @@ void run() throws PulsarAdminException {
+ ", locking required.")
private boolean subscriptionBacklogSize = false;
+ @Parameter(names = { "-etb",
+ "--get-earliest-time-in-backlog" }, description = "Set true to get earliest time in backlog")
Review comment:
```suggestion
"--get-backlog-in-mills" }, description = "Set true to get earliest time in backlog")
```
##########
File path: pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java
##########
@@ -563,10 +563,14 @@ void run() throws PulsarAdminException {
+ ", locking required.")
private boolean subscriptionBacklogSize = false;
+ @Parameter(names = { "-etb",
+ "--get-earliest-time-in-backlog" }, description = "Set true to get earliest time in backlog")
+ private boolean getEarliestTimeInBacklog = false;
Review comment:
Better to use a consistent name.
##########
File path: pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Topics.java
##########
@@ -856,8 +858,13 @@ default void delete(String topic, boolean force) throws PulsarAdminException {
* @throws PulsarAdminException
* Unexpected error
*/
- TopicStats getStats(String topic, boolean getPreciseBacklog,
- boolean subscriptionBacklogSize) throws PulsarAdminException;
+ TopicStats getStats(String topic, boolean getPreciseBacklog, boolean subscriptionBacklogSize,
Review comment:
We can't change the API directly, this will introduce a compatibility issue while users upgrade to the new version.
##########
File path: pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Topics.java
##########
@@ -856,8 +858,13 @@ default void delete(String topic, boolean force) throws PulsarAdminException {
* @throws PulsarAdminException
* Unexpected error
*/
- TopicStats getStats(String topic, boolean getPreciseBacklog,
- boolean subscriptionBacklogSize) throws PulsarAdminException;
+ TopicStats getStats(String topic, boolean getPreciseBacklog, boolean subscriptionBacklogSize,
Review comment:
No, all of the getStats methods is a public API, users might be used in their application, if we change one of them, users will encounter compatibility issues.
```java
TopicStats getStats(String topic, boolean getPreciseBacklog,
boolean subscriptionBacklogSize) throws PulsarAdminException;
default TopicStats getStats(String topic, boolean getPreciseBacklog) throws PulsarAdminException {
return getStats(topic, getPreciseBacklog, false);
}
default TopicStats getStats(String topic) throws PulsarAdminException {
return getStats(topic, false, false);
}
```
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] gaoran10 commented on a change in pull request #12523: add a metric to get earliest time in the backlog
Posted by GitBox <gi...@apache.org>.
gaoran10 commented on a change in pull request #12523:
URL: https://github.com/apache/pulsar/pull/12523#discussion_r744280486
##########
File path: pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Topics.java
##########
@@ -856,8 +858,13 @@ default void delete(String topic, boolean force) throws PulsarAdminException {
* @throws PulsarAdminException
* Unexpected error
*/
- TopicStats getStats(String topic, boolean getPreciseBacklog,
- boolean subscriptionBacklogSize) throws PulsarAdminException;
+ TopicStats getStats(String topic, boolean getPreciseBacklog, boolean subscriptionBacklogSize,
Review comment:
Maybe we could return a completable future object by the new API?
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] LeBW commented on pull request #12523: add a metric to get earliest time in the backlog
Posted by GitBox <gi...@apache.org>.
LeBW commented on pull request #12523:
URL: https://github.com/apache/pulsar/pull/12523#issuecomment-961588405
/pulsarbot run-failure-checks
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on a change in pull request #12523: add a metric to get earliest time in the backlog
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on a change in pull request #12523:
URL: https://github.com/apache/pulsar/pull/12523#discussion_r743373563
##########
File path: managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
##########
@@ -1141,6 +1144,53 @@ 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) {
+ ByteBuf metadataAndPayload = entry.getDataBuffer();
+ BrokerEntryMetadata brokerEntryMetadata = Commands.parseBrokerEntryMetadataIfExist(metadataAndPayload);
+ if (brokerEntryMetadata != null && brokerEntryMetadata.hasBrokerTimestamp()) {
+ future.complete(brokerEntryMetadata.getBrokerTimestamp());
+ } else {
+ MessageMetadata messageMetadata = Commands.parseMessageMetadata(metadataAndPayload);
+ if (messageMetadata.hasPublishTime()) {
+ future.complete(messageMetadata.getPublishTime());
+ } else {
+ future.complete(0L);
+ }
+ }
+ }
+
+ @Override
+ public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
+ future.completeExceptionally(exception);
+ }
+ }, null);
+
+ long result;
+ try {
+ result = future.get();
Review comment:
Any reason to call future.get() here? We can return a Future for this method, this will block the request thread for waiting the entry read.
##########
File path: pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Topics.java
##########
@@ -856,8 +858,13 @@ default void delete(String topic, boolean force) throws PulsarAdminException {
* @throws PulsarAdminException
* Unexpected error
*/
- TopicStats getStats(String topic, boolean getPreciseBacklog,
- boolean subscriptionBacklogSize) throws PulsarAdminException;
+ TopicStats getStats(String topic, boolean getPreciseBacklog, boolean subscriptionBacklogSize,
Review comment:
We can introduce a new API TopicStats getStats(String topic, GetStatsOptions options);
So can we can introduce more params without breaking the API.
##########
File path: pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java
##########
@@ -563,10 +563,14 @@ void run() throws PulsarAdminException {
+ ", locking required.")
private boolean subscriptionBacklogSize = false;
+ @Parameter(names = { "-etb",
+ "--get-earliest-time-in-backlog" }, description = "Set true to get earliest time in backlog")
Review comment:
```suggestion
"--get-backlog-in-mills" }, description = "Set true to get earliest time in backlog")
```
##########
File path: pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java
##########
@@ -563,10 +563,14 @@ void run() throws PulsarAdminException {
+ ", locking required.")
private boolean subscriptionBacklogSize = false;
+ @Parameter(names = { "-etb",
+ "--get-earliest-time-in-backlog" }, description = "Set true to get earliest time in backlog")
+ private boolean getEarliestTimeInBacklog = false;
Review comment:
Better to use a consistent name.
##########
File path: pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Topics.java
##########
@@ -856,8 +858,13 @@ default void delete(String topic, boolean force) throws PulsarAdminException {
* @throws PulsarAdminException
* Unexpected error
*/
- TopicStats getStats(String topic, boolean getPreciseBacklog,
- boolean subscriptionBacklogSize) throws PulsarAdminException;
+ TopicStats getStats(String topic, boolean getPreciseBacklog, boolean subscriptionBacklogSize,
Review comment:
We can't change the API directly, this will introduce a compatibility issue while users upgrade to the new version.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] gaoran10 commented on a change in pull request #12523: add a metric to get earliest time in the backlog
Posted by GitBox <gi...@apache.org>.
gaoran10 commented on a change in pull request #12523:
URL: https://github.com/apache/pulsar/pull/12523#discussion_r744277301
##########
File path: managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
##########
@@ -1141,6 +1144,53 @@ 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) {
+ ByteBuf metadataAndPayload = entry.getDataBuffer();
Review comment:
It seems that after getting the timestamp we could release the ByteBuf data.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] LeBW commented on a change in pull request #12523: add a metric to get earliest time in the backlog
Posted by GitBox <gi...@apache.org>.
LeBW commented on a change in pull request #12523:
URL: https://github.com/apache/pulsar/pull/12523#discussion_r746191084
##########
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:
> Maybe you could use the method `getEntryTimestamp()` of the class `MessageImpl`.
Fixed
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] gaoran10 commented on a change in pull request #12523: add a metric to get earliest time in the backlog
Posted by GitBox <gi...@apache.org>.
gaoran10 commented on a change in pull request #12523:
URL: https://github.com/apache/pulsar/pull/12523#discussion_r762787274
##########
File path: pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java
##########
@@ -440,6 +440,17 @@ public static void skipMessageMetadata(ByteBuf buffer) {
buffer.skipBytes(metadataSize);
}
+ public static long getEntryPublishTimestamp(ByteBuf headersAndPayloadWithBrokerEntryMetadata) throws IOException {
Review comment:
fix
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on a change in pull request #12523: add a metric to get earliest time in the backlog
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on a change in pull request #12523:
URL: https://github.com/apache/pulsar/pull/12523#discussion_r748952066
##########
File path: managed-ledger/pom.xml
##########
@@ -66,6 +66,12 @@
<version>${project.version}</version>
</dependency>
+ <dependency>
Review comment:
Yes, we can add this dependency to managed ledger
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] LeBW commented on a change in pull request #12523: add a metric to get earliest time in the backlog
Posted by GitBox <gi...@apache.org>.
LeBW commented on a change in pull request #12523:
URL: https://github.com/apache/pulsar/pull/12523#discussion_r741726291
##########
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:
I remember that the requirement is to compute the time from the earliest message timestamp to the current time. @codelipenghui
##########
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:
Fixed
##########
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:
I remember that the requirement is to compute the time from the earliest message timestamp to the current time. @codelipenghui
##########
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:
Fixed
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] LeBW commented on a change in pull request #12523: add a metric to get earliest time in the backlog
Posted by GitBox <gi...@apache.org>.
LeBW commented on a change in pull request #12523:
URL: https://github.com/apache/pulsar/pull/12523#discussion_r741726291
##########
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:
I remember that the requirement is to compute the time from the earliest message timestamp to the current time. @codelipenghui
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on a change in pull request #12523: add a metric to get earliest time in the backlog
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on a change in pull request #12523:
URL: https://github.com/apache/pulsar/pull/12523#discussion_r743373563
##########
File path: managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
##########
@@ -1141,6 +1144,53 @@ 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) {
+ ByteBuf metadataAndPayload = entry.getDataBuffer();
+ BrokerEntryMetadata brokerEntryMetadata = Commands.parseBrokerEntryMetadataIfExist(metadataAndPayload);
+ if (brokerEntryMetadata != null && brokerEntryMetadata.hasBrokerTimestamp()) {
+ future.complete(brokerEntryMetadata.getBrokerTimestamp());
+ } else {
+ MessageMetadata messageMetadata = Commands.parseMessageMetadata(metadataAndPayload);
+ if (messageMetadata.hasPublishTime()) {
+ future.complete(messageMetadata.getPublishTime());
+ } else {
+ future.complete(0L);
+ }
+ }
+ }
+
+ @Override
+ public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
+ future.completeExceptionally(exception);
+ }
+ }, null);
+
+ long result;
+ try {
+ result = future.get();
Review comment:
Any reason to call future.get() here? We can return a Future for this method, this will block the request thread for waiting the entry read.
##########
File path: pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Topics.java
##########
@@ -856,8 +858,13 @@ default void delete(String topic, boolean force) throws PulsarAdminException {
* @throws PulsarAdminException
* Unexpected error
*/
- TopicStats getStats(String topic, boolean getPreciseBacklog,
- boolean subscriptionBacklogSize) throws PulsarAdminException;
+ TopicStats getStats(String topic, boolean getPreciseBacklog, boolean subscriptionBacklogSize,
Review comment:
We can introduce a new API TopicStats getStats(String topic, GetStatsOptions options);
So can we can introduce more params without breaking the API.
##########
File path: pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java
##########
@@ -563,10 +563,14 @@ void run() throws PulsarAdminException {
+ ", locking required.")
private boolean subscriptionBacklogSize = false;
+ @Parameter(names = { "-etb",
+ "--get-earliest-time-in-backlog" }, description = "Set true to get earliest time in backlog")
Review comment:
```suggestion
"--get-backlog-in-mills" }, description = "Set true to get earliest time in backlog")
```
##########
File path: pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java
##########
@@ -563,10 +563,14 @@ void run() throws PulsarAdminException {
+ ", locking required.")
private boolean subscriptionBacklogSize = false;
+ @Parameter(names = { "-etb",
+ "--get-earliest-time-in-backlog" }, description = "Set true to get earliest time in backlog")
+ private boolean getEarliestTimeInBacklog = false;
Review comment:
Better to use a consistent name.
##########
File path: pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Topics.java
##########
@@ -856,8 +858,13 @@ default void delete(String topic, boolean force) throws PulsarAdminException {
* @throws PulsarAdminException
* Unexpected error
*/
- TopicStats getStats(String topic, boolean getPreciseBacklog,
- boolean subscriptionBacklogSize) throws PulsarAdminException;
+ TopicStats getStats(String topic, boolean getPreciseBacklog, boolean subscriptionBacklogSize,
Review comment:
We can't change the API directly, this will introduce a compatibility issue while users upgrade to the new version.
##########
File path: pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Topics.java
##########
@@ -856,8 +858,13 @@ default void delete(String topic, boolean force) throws PulsarAdminException {
* @throws PulsarAdminException
* Unexpected error
*/
- TopicStats getStats(String topic, boolean getPreciseBacklog,
- boolean subscriptionBacklogSize) throws PulsarAdminException;
+ TopicStats getStats(String topic, boolean getPreciseBacklog, boolean subscriptionBacklogSize,
Review comment:
No, all of the getStats methods is a public API, users might be used in their application, if we change one of them, users will encounter compatibility issues.
```java
TopicStats getStats(String topic, boolean getPreciseBacklog,
boolean subscriptionBacklogSize) throws PulsarAdminException;
default TopicStats getStats(String topic, boolean getPreciseBacklog) throws PulsarAdminException {
return getStats(topic, getPreciseBacklog, false);
}
default TopicStats getStats(String topic) throws PulsarAdminException {
return getStats(topic, false, false);
}
```
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] gaoran10 commented on a change in pull request #12523: add a metric to get earliest time in the backlog
Posted by GitBox <gi...@apache.org>.
gaoran10 commented on a change in pull request #12523:
URL: https://github.com/apache/pulsar/pull/12523#discussion_r744277998
##########
File path: managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
##########
@@ -1141,6 +1144,53 @@ 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) {
+ ByteBuf metadataAndPayload = entry.getDataBuffer();
+ BrokerEntryMetadata brokerEntryMetadata = Commands.parseBrokerEntryMetadataIfExist(metadataAndPayload);
+ if (brokerEntryMetadata != null && brokerEntryMetadata.hasBrokerTimestamp()) {
+ future.complete(brokerEntryMetadata.getBrokerTimestamp());
+ } else {
+ MessageMetadata messageMetadata = Commands.parseMessageMetadata(metadataAndPayload);
+ if (messageMetadata.hasPublishTime()) {
+ future.complete(messageMetadata.getPublishTime());
+ } else {
+ future.complete(0L);
+ }
+ }
+ }
+
+ @Override
+ public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
+ future.completeExceptionally(exception);
+ }
+ }, null);
+
+ long result;
+ try {
+ result = future.get();
Review comment:
Maybe we could refer to the interface `CompletableFuture<PersistentTopicInternalStats> getInternalStats(boolean includeLedgerMetadata)` of the class `org.apache.pulsar.broker.service.Topic.java`.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] gaoran10 commented on a change in pull request #12523: add a metric to get earliest time in the backlog
Posted by GitBox <gi...@apache.org>.
gaoran10 commented on a change in pull request #12523:
URL: https://github.com/apache/pulsar/pull/12523#discussion_r762788824
##########
File path: pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/SubscriptionStats.java
##########
@@ -49,6 +49,9 @@
/** Size of backlog in byte. **/
long getBacklogSize();
+ /** Get the time from the oldest message in the backlog to current time. */
+ long getTimeBacklogInMills();
Review comment:
Maybe we could use mill seconds here to make it consistent with other time-related stats.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] LeBW commented on a change in pull request #12523: add a metric to get earliest time in the backlog
Posted by GitBox <gi...@apache.org>.
LeBW commented on a change in pull request #12523:
URL: https://github.com/apache/pulsar/pull/12523#discussion_r743604270
##########
File path: managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
##########
@@ -1141,6 +1144,53 @@ 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) {
+ ByteBuf metadataAndPayload = entry.getDataBuffer();
+ BrokerEntryMetadata brokerEntryMetadata = Commands.parseBrokerEntryMetadataIfExist(metadataAndPayload);
+ if (brokerEntryMetadata != null && brokerEntryMetadata.hasBrokerTimestamp()) {
+ future.complete(brokerEntryMetadata.getBrokerTimestamp());
+ } else {
+ MessageMetadata messageMetadata = Commands.parseMessageMetadata(metadataAndPayload);
+ if (messageMetadata.hasPublishTime()) {
+ future.complete(messageMetadata.getPublishTime());
+ } else {
+ future.complete(0L);
+ }
+ }
+ }
+
+ @Override
+ public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
+ future.completeExceptionally(exception);
+ }
+ }, null);
+
+ long result;
+ try {
+ result = future.get();
Review comment:
Even if we return a `Future` here, maybe we still need to wait the entry read somewhere else (which will also block the request thread) ?
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] hangc0276 commented on a change in pull request #12523: add a metric to get earliest time in the backlog
Posted by GitBox <gi...@apache.org>.
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] LeBW commented on a change in pull request #12523: add a metric to get earliest time in the backlog
Posted by GitBox <gi...@apache.org>.
LeBW commented on a change in pull request #12523:
URL: https://github.com/apache/pulsar/pull/12523#discussion_r746190903
##########
File path: managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
##########
@@ -1141,6 +1144,53 @@ 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) {
+ ByteBuf metadataAndPayload = entry.getDataBuffer();
+ BrokerEntryMetadata brokerEntryMetadata = Commands.parseBrokerEntryMetadataIfExist(metadataAndPayload);
+ if (brokerEntryMetadata != null && brokerEntryMetadata.hasBrokerTimestamp()) {
+ future.complete(brokerEntryMetadata.getBrokerTimestamp());
+ } else {
+ MessageMetadata messageMetadata = Commands.parseMessageMetadata(metadataAndPayload);
+ if (messageMetadata.hasPublishTime()) {
+ future.complete(messageMetadata.getPublishTime());
+ } else {
+ future.complete(0L);
+ }
+ }
+ }
+
+ @Override
+ public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
+ future.completeExceptionally(exception);
+ }
+ }, null);
+
+ long result;
+ try {
+ result = future.get();
Review comment:
Fixed by using `CompletableFuture<TopicStatsImpl>`
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on a change in pull request #12523: add a metric to get earliest time in the backlog
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on a change in pull request #12523:
URL: https://github.com/apache/pulsar/pull/12523#discussion_r744069637
##########
File path: managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
##########
@@ -1141,6 +1144,53 @@ 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) {
+ ByteBuf metadataAndPayload = entry.getDataBuffer();
+ BrokerEntryMetadata brokerEntryMetadata = Commands.parseBrokerEntryMetadataIfExist(metadataAndPayload);
+ if (brokerEntryMetadata != null && brokerEntryMetadata.hasBrokerTimestamp()) {
+ future.complete(brokerEntryMetadata.getBrokerTimestamp());
+ } else {
+ MessageMetadata messageMetadata = Commands.parseMessageMetadata(metadataAndPayload);
+ if (messageMetadata.hasPublishTime()) {
+ future.complete(messageMetadata.getPublishTime());
+ } else {
+ future.complete(0L);
+ }
+ }
+ }
+
+ @Override
+ public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
+ future.completeExceptionally(exception);
+ }
+ }, null);
+
+ long result;
+ try {
+ result = future.get();
Review comment:
Previously, the broker only aggregated the data cached in memory, but after this change, we will get the data from the bookkeeper which introduced IO operations, so it's better to change the PersistentTopics.getStats() to an async way to make sure we don't block the jetty thread.
Here is an example: https://github.com/apache/pulsar/blob/04ddc9b0324e6d09b6926bbfe5402f006ad633c3/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java#L1131
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] LeBW commented on a change in pull request #12523: add a metric to get earliest time in the backlog
Posted by GitBox <gi...@apache.org>.
LeBW commented on a change in pull request #12523:
URL: https://github.com/apache/pulsar/pull/12523#discussion_r749957876
##########
File path: pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java
##########
@@ -563,10 +563,14 @@ void run() throws PulsarAdminException {
+ ", locking required.")
private boolean subscriptionBacklogSize = false;
+ @Parameter(names = { "-etb",
+ "--get-earliest-time-in-backlog" }, description = "Set true to get earliest time in backlog")
Review comment:
> Ok, to avoid confusion, I think we can only return the timestamp of the earliest unacked message?
OK, I'll do it.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on a change in pull request #12523: add a metric to get earliest time in the backlog
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on a change in pull request #12523:
URL: https://github.com/apache/pulsar/pull/12523#discussion_r749856334
##########
File path: pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java
##########
@@ -563,10 +563,14 @@ void run() throws PulsarAdminException {
+ ", locking required.")
private boolean subscriptionBacklogSize = false;
+ @Parameter(names = { "-etb",
+ "--get-earliest-time-in-backlog" }, description = "Set true to get earliest time in backlog")
Review comment:
Ok, to avoid confusion, I think we can only return the timestamp of the earliest unacked message?
##########
File path: pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java
##########
@@ -440,6 +440,17 @@ public static void skipMessageMetadata(ByteBuf buffer) {
buffer.skipBytes(metadataSize);
}
+ public static long getEntryTimestamp(ByteBuf headersAndPayloadWithBrokerEntryMetadata) throws IOException {
Review comment:
```suggestion
public static long getEntryPublishTimestamp(ByteBuf headersAndPayloadWithBrokerEntryMetadata) throws IOException {
```
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui merged pull request #12523: add a metric to get earliest time in the backlog
Posted by GitBox <gi...@apache.org>.
codelipenghui merged pull request #12523:
URL: https://github.com/apache/pulsar/pull/12523
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] github-actions[bot] commented on pull request #12523: add a metric to get earliest time in the backlog
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #12523:
URL: https://github.com/apache/pulsar/pull/12523#issuecomment-955863200
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on a change in pull request #12523: add a metric to get earliest time in the backlog
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on a change in pull request #12523:
URL: https://github.com/apache/pulsar/pull/12523#discussion_r743622687
##########
File path: pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Topics.java
##########
@@ -856,8 +858,13 @@ default void delete(String topic, boolean force) throws PulsarAdminException {
* @throws PulsarAdminException
* Unexpected error
*/
- TopicStats getStats(String topic, boolean getPreciseBacklog,
- boolean subscriptionBacklogSize) throws PulsarAdminException;
+ TopicStats getStats(String topic, boolean getPreciseBacklog, boolean subscriptionBacklogSize,
Review comment:
No, all of the getStats methods is a public API, users might be used in their application, if we change one of them, users will encounter compatibility issues.
```java
TopicStats getStats(String topic, boolean getPreciseBacklog,
boolean subscriptionBacklogSize) throws PulsarAdminException;
default TopicStats getStats(String topic, boolean getPreciseBacklog) throws PulsarAdminException {
return getStats(topic, getPreciseBacklog, false);
}
default TopicStats getStats(String topic) throws PulsarAdminException {
return getStats(topic, false, false);
}
```
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] gaoran10 commented on a change in pull request #12523: add a metric to get earliest time in the backlog
Posted by GitBox <gi...@apache.org>.
gaoran10 commented on a change in pull request #12523:
URL: https://github.com/apache/pulsar/pull/12523#discussion_r744277998
##########
File path: managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
##########
@@ -1141,6 +1144,53 @@ 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) {
+ ByteBuf metadataAndPayload = entry.getDataBuffer();
+ BrokerEntryMetadata brokerEntryMetadata = Commands.parseBrokerEntryMetadataIfExist(metadataAndPayload);
+ if (brokerEntryMetadata != null && brokerEntryMetadata.hasBrokerTimestamp()) {
+ future.complete(brokerEntryMetadata.getBrokerTimestamp());
+ } else {
+ MessageMetadata messageMetadata = Commands.parseMessageMetadata(metadataAndPayload);
+ if (messageMetadata.hasPublishTime()) {
+ future.complete(messageMetadata.getPublishTime());
+ } else {
+ future.complete(0L);
+ }
+ }
+ }
+
+ @Override
+ public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
+ future.completeExceptionally(exception);
+ }
+ }, null);
+
+ long result;
+ try {
+ result = future.get();
Review comment:
Maybe we could refer to the interface `CompletableFuture<PersistentTopicInternalStats> getInternalStats(boolean includeLedgerMetadata)` of the class `org.apache.pulsar.broker.service.Topic`.
If it's failed to get the timestamp, the return value is 0, this may cause confusion because if there is no backlog, the return value is also 0, maye it's better to return the error response.
##########
File path: managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
##########
@@ -1141,6 +1144,53 @@ 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) {
+ ByteBuf metadataAndPayload = entry.getDataBuffer();
+ BrokerEntryMetadata brokerEntryMetadata = Commands.parseBrokerEntryMetadataIfExist(metadataAndPayload);
+ if (brokerEntryMetadata != null && brokerEntryMetadata.hasBrokerTimestamp()) {
+ future.complete(brokerEntryMetadata.getBrokerTimestamp());
+ } else {
+ MessageMetadata messageMetadata = Commands.parseMessageMetadata(metadataAndPayload);
+ if (messageMetadata.hasPublishTime()) {
+ future.complete(messageMetadata.getPublishTime());
+ } else {
+ future.complete(0L);
+ }
+ }
+ }
+
+ @Override
+ public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
+ future.completeExceptionally(exception);
+ }
+ }, null);
+
+ long result;
+ try {
+ result = future.get();
Review comment:
Maybe we could refer to the interface `CompletableFuture<PersistentTopicInternalStats> getInternalStats(boolean includeLedgerMetadata)` of the class `org.apache.pulsar.broker.service.Topic`.
If it's failed to get the timestamp, the return value is 0, this may cause confusion because if there is no backlog, the return value is also 0, maybe it's better to return the error response.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] gaoran10 commented on a change in pull request #12523: add a metric to get earliest time in the backlog
Posted by GitBox <gi...@apache.org>.
gaoran10 commented on a change in pull request #12523:
URL: https://github.com/apache/pulsar/pull/12523#discussion_r744277998
##########
File path: managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
##########
@@ -1141,6 +1144,53 @@ 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) {
+ ByteBuf metadataAndPayload = entry.getDataBuffer();
+ BrokerEntryMetadata brokerEntryMetadata = Commands.parseBrokerEntryMetadataIfExist(metadataAndPayload);
+ if (brokerEntryMetadata != null && brokerEntryMetadata.hasBrokerTimestamp()) {
+ future.complete(brokerEntryMetadata.getBrokerTimestamp());
+ } else {
+ MessageMetadata messageMetadata = Commands.parseMessageMetadata(metadataAndPayload);
+ if (messageMetadata.hasPublishTime()) {
+ future.complete(messageMetadata.getPublishTime());
+ } else {
+ future.complete(0L);
+ }
+ }
+ }
+
+ @Override
+ public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
+ future.completeExceptionally(exception);
+ }
+ }, null);
+
+ long result;
+ try {
+ result = future.get();
Review comment:
Maybe we could refer to the interface `CompletableFuture<PersistentTopicInternalStats> getInternalStats(boolean includeLedgerMetadata)` of the class `org.apache.pulsar.broker.service.Topic`.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on a change in pull request #12523: add a metric to get earliest time in the backlog
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on a change in pull request #12523:
URL: https://github.com/apache/pulsar/pull/12523#discussion_r743374976
##########
File path: pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Topics.java
##########
@@ -856,8 +858,13 @@ default void delete(String topic, boolean force) throws PulsarAdminException {
* @throws PulsarAdminException
* Unexpected error
*/
- TopicStats getStats(String topic, boolean getPreciseBacklog,
- boolean subscriptionBacklogSize) throws PulsarAdminException;
+ TopicStats getStats(String topic, boolean getPreciseBacklog, boolean subscriptionBacklogSize,
Review comment:
We can introduce a new API TopicStats getStats(GetStatsOptions options);
So can we can introduce more params without breaking the API.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] LeBW commented on pull request #12523: add a metric to get earliest time in the backlog
Posted by GitBox <gi...@apache.org>.
LeBW commented on pull request #12523:
URL: https://github.com/apache/pulsar/pull/12523#issuecomment-961588405
/pulsarbot run-failure-checks
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] hangc0276 commented on a change in pull request #12523: add a metric to get earliest time in the backlog
Posted by GitBox <gi...@apache.org>.
hangc0276 commented on a change in pull request #12523:
URL: https://github.com/apache/pulsar/pull/12523#discussion_r742469504
##########
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:
Whether the `result` < 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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] LeBW commented on a change in pull request #12523: add a metric to get earliest time in the backlog
Posted by GitBox <gi...@apache.org>.
LeBW commented on a change in pull request #12523:
URL: https://github.com/apache/pulsar/pull/12523#discussion_r742484742
##########
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:
The `result` is either 0 or a timestamp. So maybe it's impossible to `< 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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] hangc0276 commented on a change in pull request #12523: add a metric to get earliest time in the backlog
Posted by GitBox <gi...@apache.org>.
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] LeBW commented on a change in pull request #12523: add a metric to get earliest time in the backlog
Posted by GitBox <gi...@apache.org>.
LeBW commented on a change in pull request #12523:
URL: https://github.com/apache/pulsar/pull/12523#discussion_r743609208
##########
File path: pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java
##########
@@ -563,10 +563,14 @@ void run() throws PulsarAdminException {
+ ", locking required.")
private boolean subscriptionBacklogSize = false;
+ @Parameter(names = { "-etb",
+ "--get-earliest-time-in-backlog" }, description = "Set true to get earliest time in backlog")
Review comment:
Can `--get-backlog-in-mills` reflect that it will get a timestamp? Maybe it's a little confusing as it looks like getting the backlog itself. (It's just my question and I can still change it to `--get-backlog-in-mills`)
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] hangc0276 commented on a change in pull request #12523: add a metric to get earliest time in the backlog
Posted by GitBox <gi...@apache.org>.
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] LeBW commented on a change in pull request #12523: add a metric to get earliest time in the backlog
Posted by GitBox <gi...@apache.org>.
LeBW commented on a change in pull request #12523:
URL: https://github.com/apache/pulsar/pull/12523#discussion_r746190464
##########
File path: pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Topics.java
##########
@@ -856,8 +858,13 @@ default void delete(String topic, boolean force) throws PulsarAdminException {
* @throws PulsarAdminException
* Unexpected error
*/
- TopicStats getStats(String topic, boolean getPreciseBacklog,
- boolean subscriptionBacklogSize) throws PulsarAdminException;
+ TopicStats getStats(String topic, boolean getPreciseBacklog, boolean subscriptionBacklogSize,
Review comment:
Fixed by using a new API `TopicStats getStats(GetStatsOptions options)`
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] LeBW commented on a change in pull request #12523: add a metric to get earliest time in the backlog
Posted by GitBox <gi...@apache.org>.
LeBW commented on a change in pull request #12523:
URL: https://github.com/apache/pulsar/pull/12523#discussion_r750193635
##########
File path: pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java
##########
@@ -563,10 +563,14 @@ void run() throws PulsarAdminException {
+ ", locking required.")
private boolean subscriptionBacklogSize = false;
+ @Parameter(names = { "-etb",
+ "--get-earliest-time-in-backlog" }, description = "Set true to get earliest time in backlog")
Review comment:
> Ok, to avoid confusion, I think we can only return the timestamp of the earliest unacked message?
Done
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org