You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/10/12 19:18:40 UTC
(pulsar) 15/27: [fix][ml] Fix underReplicated always true in topic internal stats (#26834)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-5.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 071328943ac7b1cceba301b2ba4a5ef3f9687d83
Author: Jakub Kucharovic <ja...@kucharovic.cz>
AuthorDate: Mon Oct 12 08:20:58 2026 +0200
[fix][ml] Fix underReplicated always true in topic internal stats (#26834)
(cherry picked from commit cacbe73c222cf0678bb0a30bc0b13edbf2b359ba)
---
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 2 +-
.../bookkeeper/mledger/impl/ManagedLedgerTest.java | 36 ++++++++++++++++++++++
.../stats/client/PulsarBrokerStatsClientTest.java | 5 ++-
.../pulsar/tests/integration/admin/AdminTest.java | 2 +-
4 files changed, 40 insertions(+), 5 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 a6056aef7c0e..98394275779d 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
@@ -5370,7 +5370,7 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
.flatMap(Collection::stream)
.collect(Collectors.toSet());
if (bookies != null) {
- info.underReplicated = !bookies.contains(ensemble);
+ info.underReplicated = !bookies.containsAll(ensemble);
}
}
}
diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java
index a084640f10c2..d16ea78d8417 100644
--- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java
@@ -105,6 +105,7 @@ import org.apache.bookkeeper.client.BookKeeper;
import org.apache.bookkeeper.client.BookKeeper.DigestType;
import org.apache.bookkeeper.client.EnsemblePlacementPolicy;
import org.apache.bookkeeper.client.LedgerHandle;
+import org.apache.bookkeeper.client.LedgerMetadataBuilder;
import org.apache.bookkeeper.client.PulsarMockBookKeeper;
import org.apache.bookkeeper.client.PulsarMockLedgerHandle;
import org.apache.bookkeeper.client.PulsarMockReadHandleInterceptor;
@@ -149,6 +150,8 @@ import org.apache.bookkeeper.mledger.impl.cache.EntryCacheManager;
import org.apache.bookkeeper.mledger.proto.ManagedLedgerInfo;
import org.apache.bookkeeper.mledger.proto.ManagedLedgerInfo.LedgerInfo;
import org.apache.bookkeeper.mledger.util.Futures;
+import org.apache.bookkeeper.net.BookieId;
+import org.apache.bookkeeper.net.BookieSocketAddress;
import org.apache.bookkeeper.test.MockedBookKeeperTestCase;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.commons.lang3.mutable.MutableBoolean;
@@ -4903,6 +4906,39 @@ public class ManagedLedgerTest extends MockedBookKeeperTestCase {
Assert.assertFalse(managedLedger.ledgerCache.containsKey(lastLedger));
}
+ @Test
+ public void testInternalStatsUnderReplicated() throws Exception {
+ ManagedLedgerImpl ledger = spy((ManagedLedgerImpl) factory.open("testInternalStatsUnderReplicated",
+ initManagedLedgerConfig(defaultConfig())));
+ long ledgerId = ledger.ledgers.lastKey();
+ List<BookieId> registered = new ArrayList<>(PulsarMockBookKeeper.getMockEnsemble());
+ List<BookieId> withUnregistered = List.of(registered.get(0), registered.get(1),
+ new BookieSocketAddress("192.0.2.4", 1234).toBookieId());
+
+ mockEnsembles(ledger, ledgerId, registered, registered);
+ assertFalse(ledger.getManagedLedgerInternalStats(true).get().ledgers.get(0).underReplicated);
+
+ mockEnsembles(ledger, ledgerId, registered, withUnregistered);
+ assertTrue(ledger.getManagedLedgerInternalStats(true).get().ledgers.get(0).underReplicated);
+
+ mockEnsembles(ledger, ledgerId, withUnregistered, registered);
+ assertTrue(ledger.getManagedLedgerInternalStats(true).get().ledgers.get(0).underReplicated);
+
+ ledger.close();
+ }
+
+ private static void mockEnsembles(ManagedLedgerImpl ledger, long ledgerId, List<BookieId> olderEnsemble,
+ List<BookieId> currentEnsemble) {
+ LedgerMetadata metadata = LedgerMetadataBuilder.create()
+ .withDigestType(DigestType.CRC32C.toApiDigestType())
+ .withPassword(new byte[0])
+ .withId(ledgerId)
+ .newEnsembleEntry(0L, olderEnsemble)
+ .newEnsembleEntry(10L, currentEnsemble)
+ .build();
+ doReturn(CompletableFuture.completedFuture(metadata)).when(ledger).getLedgerMetadata(ledgerId);
+ }
+
@Test
public void testIsNoMessagesAfterPos() throws Exception {
final byte[] data = new byte[]{1, 2, 3};
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/stats/client/PulsarBrokerStatsClientTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/stats/client/PulsarBrokerStatsClientTest.java
index 9376983c93a6..16c618f5d3d7 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/stats/client/PulsarBrokerStatsClientTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/stats/client/PulsarBrokerStatsClientTest.java
@@ -126,9 +126,8 @@ public class PulsarBrokerStatsClientTest extends SharedPulsarBaseTest {
PersistentTopicInternalStats internalStats = topic.getInternalStats(true).get();
assertNotNull(internalStats.ledgers.get(0).metadata);
- // For the mock test, the default ensembles is ["192.0.2.1:1234","192.0.2.2:1234","192.0.2.3:1234"]
- // The registered bookie ID is 192.168.1.1:5000
- assertTrue(internalStats.ledgers.get(0).underReplicated);
+ // Every bookie in the ensemble is registered and available
+ assertFalse(internalStats.ledgers.get(0).underReplicated);
for (ManagedLedgerInternalStats.LedgerInfo ledgerInfo : internalStats.ledgers) {
if (ledgerInfo.ledgerId == firstLedgerId) {
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/admin/AdminTest.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/admin/AdminTest.java
index c22c09cb3ea8..606ef517d073 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/admin/AdminTest.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/admin/AdminTest.java
@@ -65,7 +65,7 @@ public class AdminTest extends MessagingBase {
}
log.info().attr("to", topicName).log("Successfully to publish 10 messages to");
- PersistentTopicInternalStats stats = admin.topics().getInternalStats(topicName);
+ PersistentTopicInternalStats stats = admin.topics().getInternalStats(topicName, true);
Assert.assertTrue(stats.ledgers.size() > 0);
for (PersistentTopicInternalStats.LedgerInfo ledger : stats.ledgers) {
Assert.assertFalse(ledger.underReplicated);