From a81ec2f089ab9aac13c08c75b362b8516a829088 Mon Sep 17 00:00:00 2001 From: fengyubiao Date: Thu, 10 Aug 2023 20:58:29 +0800 Subject: [PATCH 1/2] [fix] [admin] Fix get topic stats fail if a subscription catch up concurrently --- .../mledger/impl/ManagedLedgerImpl.java | 4 +++ .../persistent/PersistentSubscription.java | 22 ++++++++------- .../pulsar/broker/admin/AdminApi2Test.java | 27 +++++++++++++++++++ 3 files changed, 44 insertions(+), 9 deletions(-) diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java index 14f4bfed87172..c31a0c38cd395 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java @@ -1240,6 +1240,10 @@ public CompletableFuture getEarliestMessagePublishTimeOfPos(PositionImpl p } PositionImpl nextPos = getNextValidPosition(pos); + if (nextPos.compareTo(lastConfirmedEntry) > 0) { + return CompletableFuture.completedFuture(-1L); + } + asyncReadEntry(nextPos, new ReadEntryCallback() { @Override public void readEntryComplete(Entry entry, Object ctx) { diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java index 1b8b47da78fed..1d8f78385778c 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java @@ -1160,16 +1160,20 @@ public SubscriptionStatsImpl getStats(Boolean getPreciseBacklog, boolean subscri } else { subStats.backlogSize = -1; } - if (getEarliestTimeInBacklog && subStats.msgBacklog > 0) { - ManagedLedgerImpl managedLedger = ((ManagedLedgerImpl) cursor.getManagedLedger()); - PositionImpl markDeletedPosition = (PositionImpl) cursor.getMarkDeletedPosition(); - long result = 0; - try { - result = managedLedger.getEarliestMessagePublishTimeOfPos(markDeletedPosition).get(); - } catch (InterruptedException | ExecutionException e) { - result = -1; + if (getEarliestTimeInBacklog) { + if (subStats.msgBacklog > 0) { + ManagedLedgerImpl managedLedger = ((ManagedLedgerImpl) cursor.getManagedLedger()); + PositionImpl markDeletedPosition = (PositionImpl) cursor.getMarkDeletedPosition(); + long result = 0; + try { + result = managedLedger.getEarliestMessagePublishTimeOfPos(markDeletedPosition).get(); + } catch (InterruptedException | ExecutionException e) { + result = -1; + } + subStats.earliestMsgPublishTimeInBacklog = result; + } else { + subStats.earliestMsgPublishTimeInBacklog = -1; } - subStats.earliestMsgPublishTimeInBacklog = result; } subStats.msgBacklogNoDelayed = subStats.msgBacklog - subStats.msgDelayed; subStats.msgRateExpired = expiryMonitor.getMessageExpiryRate(); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java index d6176966d85d9..0adf945a55552 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java @@ -61,6 +61,7 @@ import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl; import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl; import org.apache.commons.lang3.reflect.FieldUtils; +import org.apache.pulsar.broker.BrokerTestUtil; import org.apache.pulsar.broker.PulsarService; import org.apache.pulsar.broker.ServiceConfiguration; import org.apache.pulsar.broker.admin.AdminApiTest.MockedPulsarService; @@ -3191,6 +3192,32 @@ public void testFailedUpdatePartitionedTopic() throws Exception { assertEquals(admin.topics().getPartitionedTopicMetadata(partitionedTopicName).partitions, newPartitions); } + /** + * Validate retring failed partitioned topic should succeed. + * @throws Exception + */ + @Test + public void testTopicStatsWithEarliestTimeInBacklogIfNoBacklog() throws Exception { + final String topicName = BrokerTestUtil.newUniqueName("persistent://" + defaultNamespace + "/tp_"); + final String subscriptionName = "s1"; + admin.topics().createNonPartitionedTopic(topicName); + admin.topics().createSubscription(topicName, subscriptionName, MessageId.earliest); + + // Send one message. + Producer producer = pulsarClient.newProducer(Schema.STRING).topic(topicName).enableBatching(false) + .create(); + MessageIdImpl messageId = (MessageIdImpl) producer.send("123"); + // Catch up. + admin.topics().skipAllMessages(topicName, subscriptionName); + // Get topic stats with earliestTimeInBacklog + TopicStats topicStats = admin.topics().getStats(topicName, false, false, true); + assertEquals(topicStats.getSubscriptions().get(subscriptionName).getEarliestMsgPublishTimeInBacklog(), -1L); + + // cleanup. + producer.close(); + admin.topics().delete(topicName); + } + @Test(dataProvider = "topicType") public void testPartitionedStatsAggregationByProducerName(String topicType) throws Exception { restartClusterIfReused(); From e9e6a69b92368bbc2b0dc7cc744321c1befb52bc Mon Sep 17 00:00:00 2001 From: fengyubiao Date: Mon, 14 Aug 2023 16:42:08 +0800 Subject: [PATCH 2/2] fix test --- .../java/org/apache/pulsar/broker/admin/AdminApiTest.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java index 342a409c4ae51..a4be829b977f7 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java @@ -1289,7 +1289,7 @@ public void testGetStats() throws Exception { TopicStats topicStats = admin.topics().getStats(topic, false, false, true); assertEquals(topicStats.getEarliestMsgPublishTimeInBacklogs(), 0); - assertEquals(topicStats.getSubscriptions().get(subName).getEarliestMsgPublishTimeInBacklog(), 0); + assertEquals(topicStats.getSubscriptions().get(subName).getEarliestMsgPublishTimeInBacklog(), -1); assertEquals(topicStats.getSubscriptions().get(subName).getBacklogSize(), -1); // publish several messages @@ -1309,7 +1309,7 @@ public void testGetStats() throws Exception { topicStats = admin.topics().getStats(topic, false, true, true); assertEquals(topicStats.getEarliestMsgPublishTimeInBacklogs(), 0); - assertEquals(topicStats.getSubscriptions().get(subName).getEarliestMsgPublishTimeInBacklog(), 0); + assertEquals(topicStats.getSubscriptions().get(subName).getEarliestMsgPublishTimeInBacklog(), -1); assertEquals(topicStats.getSubscriptions().get(subName).getBacklogSize(), 0); }