Skip to content

Commit 5d557e8

Browse files
committed
fix not update LastDataMessagePublishedTimestamp.
1 parent ac263c0 commit 5d557e8

2 files changed

Lines changed: 6 additions & 0 deletions

File tree

‎pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -54,6 +54,7 @@
5454
import java.util.stream.Collectors;
5555
import javax.annotation.Nonnull;
5656
import lombok.Getter;
57+
import lombok.Setter;
5758
import lombok.Value;
5859
import org.apache.bookkeeper.client.api.LedgerMetadata;
5960
import org.apache.bookkeeper.mledger.AsyncCallbacks;
@@ -269,6 +270,7 @@ protected TopicStatsHelper initialValue() {
269270

270271
// Record the last time a data message (ie: not an internal Pulsar marker) is published on the topic
271272
@Getter
273+
@Setter
272274
private volatile long lastDataMessagePublishedTimestamp = 0;
273275
@Getter
274276
private final ExecutorService orderedExecutor;

‎pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,8 @@
2323
import io.netty.util.Timeout;
2424
import io.netty.util.Timer;
2525
import io.netty.util.TimerTask;
26+
27+
import java.time.Clock;
2628
import java.util.List;
2729
import java.util.concurrent.CompletableFuture;
2830
import java.util.concurrent.ConcurrentHashMap;
@@ -321,6 +323,8 @@ public void addComplete(Position position, ByteBuf entryData, Object ctx) {
321323
}
322324
txnCommittedCounter.increment();
323325
completableFuture.complete(null);
326+
// need to update the last data message published timestamp
327+
topic.setLastDataMessagePublishedTimestamp(Clock.systemUTC().millis());
324328
}
325329

326330
@Override

0 commit comments

Comments
 (0)