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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
