You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by yu...@apache.org on 2023/02/27 15:13:22 UTC
[pulsar] branch branch-2.11 updated (682d045b76a -> 3095b6a4bc0)
This is an automated email from the ASF dual-hosted git repository.
yubiao pushed a change to branch branch-2.11
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from 682d045b76a [fix][broker] Copy command fields and fix potential thread-safety in ServerCnx (#19517)
new 3280e7e66ac [fix] [ml] topic load fail by ledger lost (#19444)
new 3095b6a4bc0 [improve] [admin] Make the default value of param --get-subscription-backlog-size of admin API topics stats true (#19302)
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 29 ++++--
.../bookkeeper/mledger/impl/ManagedLedgerTest.java | 116 ++++++++++++++++++++-
.../pulsar/broker/admin/v2/PersistentTopics.java | 4 +-
.../service/persistent/PersistentSubscription.java | 2 +
.../apache/pulsar/broker/admin/AdminApiTest.java | 7 +-
.../pulsar/admin/cli/PulsarAdminToolTest.java | 6 +-
.../org/apache/pulsar/admin/cli/CmdTopics.java | 6 +-
.../policies/data/stats/SubscriptionStatsImpl.java | 2 +-
8 files changed, 150 insertions(+), 22 deletions(-)
[pulsar] 02/02: [improve] [admin] Make the default value of param --get-subscription-backlog-size of admin API topics stats true (#19302)
Posted by yu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yubiao pushed a commit to branch branch-2.11
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 3095b6a4bc038c6d134e9193bd05c3cb3bb81644
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Sat Jan 28 20:19:46 2023 +0800
[improve] [admin] Make the default value of param --get-subscription-backlog-size of admin API topics stats true (#19302)
(cherry picked from commit e417fe7895253f7cf4e6b2d019ab615f4574098a)
---
.../java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java | 4 ++--
.../pulsar/broker/service/persistent/PersistentSubscription.java | 2 ++
.../src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java | 7 +++++--
.../test/java/org/apache/pulsar/admin/cli/PulsarAdminToolTest.java | 6 +++---
.../src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java | 6 +++---
.../pulsar/common/policies/data/stats/SubscriptionStatsImpl.java | 2 +-
6 files changed, 16 insertions(+), 11 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java
index e60baa1c2dd..08185bb2970 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java
@@ -1136,7 +1136,7 @@ public class PersistentTopics extends PersistentTopicsBase {
@QueryParam("getPreciseBacklog") @DefaultValue("false") boolean getPreciseBacklog,
@ApiParam(value = "If return backlog size for each subscription, require locking on ledger so be careful "
+ "not to use when there's heavy traffic.")
- @QueryParam("subscriptionBacklogSize") @DefaultValue("false") boolean subscriptionBacklogSize,
+ @QueryParam("subscriptionBacklogSize") @DefaultValue("true") boolean subscriptionBacklogSize,
@ApiParam(value = "If return time of the earliest message in backlog")
@QueryParam("getEarliestTimeInBacklog") @DefaultValue("false") boolean getEarliestTimeInBacklog) {
validateTopicName(tenant, namespace, encodedTopic);
@@ -1238,7 +1238,7 @@ public class PersistentTopics extends PersistentTopicsBase {
@QueryParam("getPreciseBacklog") @DefaultValue("false") boolean getPreciseBacklog,
@ApiParam(value = "If return backlog size for each subscription, require locking on ledger so be careful "
+ "not to use when there's heavy traffic.")
- @QueryParam("subscriptionBacklogSize") @DefaultValue("false") boolean subscriptionBacklogSize,
+ @QueryParam("subscriptionBacklogSize") @DefaultValue("true") boolean subscriptionBacklogSize,
@ApiParam(value = "If return the earliest time in backlog")
@QueryParam("getEarliestTimeInBacklog") @DefaultValue("false") boolean getEarliestTimeInBacklog) {
try {
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 fe3faf35bc9..cb22c5c5879 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
@@ -1123,6 +1123,8 @@ public class PersistentSubscription extends AbstractSubscription implements Subs
if (subscriptionBacklogSize) {
subStats.backlogSize = ((ManagedLedgerImpl) topic.getManagedLedger())
.getEstimatedBacklogSize((PositionImpl) cursor.getMarkDeletedPosition());
+ } else {
+ subStats.backlogSize = -1;
}
if (getEarliestTimeInBacklog && subStats.msgBacklog > 0) {
ManagedLedgerImpl managedLedger = ((ManagedLedgerImpl) cursor.getManagedLedger());
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 8d58eae59a3..4611f670dcb 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
@@ -1195,14 +1195,16 @@ public class AdminApiTest extends MockedPulsarServiceBaseTest {
assertEquals(topicStats.getEarliestMsgPublishTimeInBacklogs(), 0);
assertEquals(topicStats.getSubscriptions().get(subName).getEarliestMsgPublishTimeInBacklog(), 0);
+ assertEquals(topicStats.getSubscriptions().get(subName).getBacklogSize(), -1);
// publish several messages
publishMessagesOnPersistentTopic(topic, 10);
Thread.sleep(1000);
- topicStats = admin.topics().getStats(topic, false, false, true);
+ topicStats = admin.topics().getStats(topic, false, true, true);
assertTrue(topicStats.getEarliestMsgPublishTimeInBacklogs() > 0);
assertTrue(topicStats.getSubscriptions().get(subName).getEarliestMsgPublishTimeInBacklog() > 0);
+ assertTrue(topicStats.getSubscriptions().get(subName).getBacklogSize() > 0);
for (int i = 0; i < 10; i++) {
Message<byte[]> message = consumer.receive();
@@ -1210,9 +1212,10 @@ public class AdminApiTest extends MockedPulsarServiceBaseTest {
}
Thread.sleep(1000);
- topicStats = admin.topics().getStats(topic, false, false, true);
+ topicStats = admin.topics().getStats(topic, false, true, true);
assertEquals(topicStats.getEarliestMsgPublishTimeInBacklogs(), 0);
assertEquals(topicStats.getSubscriptions().get(subName).getEarliestMsgPublishTimeInBacklog(), 0);
+ assertEquals(topicStats.getSubscriptions().get(subName).getBacklogSize(), 0);
}
diff --git a/pulsar-client-tools-test/src/test/java/org/apache/pulsar/admin/cli/PulsarAdminToolTest.java b/pulsar-client-tools-test/src/test/java/org/apache/pulsar/admin/cli/PulsarAdminToolTest.java
index 0e393c901b4..d77f312ff85 100644
--- a/pulsar-client-tools-test/src/test/java/org/apache/pulsar/admin/cli/PulsarAdminToolTest.java
+++ b/pulsar-client-tools-test/src/test/java/org/apache/pulsar/admin/cli/PulsarAdminToolTest.java
@@ -1404,7 +1404,7 @@ public class PulsarAdminToolTest {
verify(mockTopics).deleteSubscription("persistent://myprop/clust/ns1/ds1", "sub1", false);
cmdTopics.run(split("stats persistent://myprop/clust/ns1/ds1"));
- verify(mockTopics).getStats("persistent://myprop/clust/ns1/ds1", false, false, false);
+ verify(mockTopics).getStats("persistent://myprop/clust/ns1/ds1", false, true, false);
cmdTopics.run(split("stats-internal persistent://myprop/clust/ns1/ds1"));
verify(mockTopics).getInternalStats("persistent://myprop/clust/ns1/ds1", false);
@@ -1452,7 +1452,7 @@ public class PulsarAdminToolTest {
cmdTopics.run(split("partitioned-stats persistent://myprop/clust/ns1/ds1 --per-partition"));
verify(mockTopics).getPartitionedStats("persistent://myprop/clust/ns1/ds1",
- true, false, false, false);
+ true, false, true, false);
cmdTopics.run(split("partitioned-stats-internal persistent://myprop/clust/ns1/ds1"));
verify(mockTopics).getPartitionedInternalStats("persistent://myprop/clust/ns1/ds1");
@@ -1978,7 +1978,7 @@ public class PulsarAdminToolTest {
CmdTopics topics = new CmdTopics(() -> admin);
topics.run(split("stats non-persistent://myprop/ns1/ds1"));
- verify(mockTopics).getStats("non-persistent://myprop/ns1/ds1", false, false, false);
+ verify(mockTopics).getStats("non-persistent://myprop/ns1/ds1", false, true, false);
topics.run(split("stats-internal non-persistent://myprop/ns1/ds1"));
verify(mockTopics).getInternalStats("non-persistent://myprop/ns1/ds1", false);
diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java
index 4b73703395e..c5ac05110bb 100644
--- a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java
+++ b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java
@@ -743,8 +743,8 @@ public class CmdTopics extends CmdBase {
@Parameter(names = { "-sbs",
"--get-subscription-backlog-size" }, description = "Set true to get backlog size for each subscription"
- + ", locking required.")
- private boolean subscriptionBacklogSize = false;
+ + ", locking required. If set to false, the attribute 'backlogSize' in the response will be -1")
+ private boolean subscriptionBacklogSize = true;
@Parameter(names = { "-etb",
"--get-earliest-time-in-backlog" }, description = "Set true to get earliest time in backlog")
@@ -809,7 +809,7 @@ public class CmdTopics extends CmdBase {
@Parameter(names = { "-sbs",
"--get-subscription-backlog-size" }, description = "Set true to get backlog size for each subscription"
+ ", locking required.")
- private boolean subscriptionBacklogSize = false;
+ private boolean subscriptionBacklogSize = true;
@Parameter(names = { "-etb",
"--get-earliest-time-in-backlog" }, description = "Set true to get earliest time in backlog")
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/SubscriptionStatsImpl.java b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/SubscriptionStatsImpl.java
index 02e1e8e184b..48f2cf24ffd 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/SubscriptionStatsImpl.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/SubscriptionStatsImpl.java
@@ -58,7 +58,7 @@ public class SubscriptionStatsImpl implements SubscriptionStats {
/** Number of entries in the subscription backlog. */
public long msgBacklog;
- /** Size of backlog in byte. **/
+ /** Size of backlog in byte, -1 means that the argument "subscriptionBacklogSize" is false when calling the API. **/
public long backlogSize;
/** Get the publish time of the earliest message in the backlog. */
[pulsar] 01/02: [fix] [ml] topic load fail by ledger lost (#19444)
Posted by yu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yubiao pushed a commit to branch branch-2.11
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 3280e7e66ac8c234c4063d1e5e6553435af2aa63
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Tue Feb 21 21:47:28 2023 +0800
[fix] [ml] topic load fail by ledger lost (#19444)
Makes only ledgers removed from the meta of ledger info can be deleted from the BK.
(cherry picked from commit 3314d70231cbed89b6eefa2073ef8c048d84ec16)
---
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 29 ++++--
.../bookkeeper/mledger/impl/ManagedLedgerTest.java | 116 ++++++++++++++++++++-
2 files changed, 134 insertions(+), 11 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 7cb7763482e..54eea3432c0 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
@@ -66,6 +66,7 @@ import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
import java.util.function.Predicate;
import java.util.function.Supplier;
import java.util.stream.Collectors;
+import javax.annotation.Nullable;
import lombok.Getter;
import org.apache.bookkeeper.client.AsyncCallback;
import org.apache.bookkeeper.client.AsyncCallback.CreateCallback;
@@ -150,7 +151,7 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
protected final BookKeeper bookKeeper;
protected final String name;
private final Map<String, byte[]> ledgerMetadata;
- private final BookKeeper.DigestType digestType;
+ protected final BookKeeper.DigestType digestType;
protected ManagedLedgerConfig config;
protected Map<String, String> propertiesMap;
@@ -450,6 +451,7 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
}
// Calculate total entries and size
+ final List<Long> emptyLedgersToBeDeleted = Collections.synchronizedList(new ArrayList<>());
Iterator<LedgerInfo> iterator = ledgers.values().iterator();
while (iterator.hasNext()) {
LedgerInfo li = iterator.next();
@@ -458,9 +460,7 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
TOTAL_SIZE_UPDATER.addAndGet(this, li.getSize());
} else {
iterator.remove();
- bookKeeper.asyncDeleteLedger(li.getLedgerId(), (rc, ctx) -> {
- log.info("[{}] Deleted empty ledger ledgerId={} rc={}", name, li.getLedgerId(), rc);
- }, null);
+ emptyLedgersToBeDeleted.add(li.getLedgerId());
}
}
@@ -476,6 +476,11 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
@Override
public void operationComplete(Void v, Stat stat) {
ledgersStat = stat;
+ emptyLedgersToBeDeleted.forEach(ledgerId -> {
+ bookKeeper.asyncDeleteLedger(ledgerId, (rc, ctx) -> {
+ log.info("[{}] Deleted empty ledger ledgerId={} rc={}", name, ledgerId, rc);
+ }, null);
+ });
initializeCursors(callback);
}
@@ -1506,11 +1511,12 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
}
ledgersStat = stat;
synchronized (ManagedLedgerImpl.this) {
+ LedgerHandle originalCurrentLedger = currentLedger;
ledgers.put(lh.getId(), newLedger);
currentLedger = lh;
currentLedgerEntries = 0;
currentLedgerSize = 0;
- updateLedgersIdsComplete(ledgersStat);
+ updateLedgersIdsComplete(originalCurrentLedger);
mbean.addLedgerSwitchLatencySample(System.currentTimeMillis()
- lastLedgerCreationInitiationTimestamp, TimeUnit.MILLISECONDS);
}
@@ -1581,8 +1587,15 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
store.asyncUpdateLedgerIds(name, mlInfo, ledgersStat, callback);
}
- public synchronized void updateLedgersIdsComplete(Stat stat) {
+ public synchronized void updateLedgersIdsComplete(@Nullable LedgerHandle originalCurrentLedger) {
STATE_UPDATER.set(this, State.LedgerOpened);
+ // Delete original "currentLedger" if it has been removed from "ledgers".
+ if (originalCurrentLedger != null && !ledgers.containsKey(originalCurrentLedger.getId())){
+ bookKeeper.asyncDeleteLedger(originalCurrentLedger.getId(), (rc, ctx) -> {
+ mbean.endDataLedgerDeleteOp();
+ log.info("[{}] Delete complete for empty ledger {}. rc={}", name, originalCurrentLedger.getId(), rc);
+ }, null);
+ }
updateLastLedgerCreatedTimeAndScheduleRolloverTask();
if (log.isDebugEnabled()) {
@@ -1660,10 +1673,6 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
// The last ledger was empty, so we can discard it
ledgers.remove(lh.getId());
mbean.startDataLedgerDeleteOp();
- bookKeeper.asyncDeleteLedger(lh.getId(), (rc, ctx) -> {
- mbean.endDataLedgerDeleteOp();
- log.info("[{}] Delete complete for empty ledger {}. rc={}", name, lh.getId(), rc);
- }, null);
}
trimConsumedLedgersInBackground();
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 b5fc3d65cfd..7ad7e0f8d68 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
@@ -48,6 +48,7 @@ import java.nio.ReadOnlyBufferException;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.security.GeneralSecurityException;
+import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -75,7 +76,9 @@ import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Predicate;
import lombok.Cleanup;
+import lombok.Data;
import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.client.AsyncCallback;
import org.apache.bookkeeper.client.AsyncCallback.AddCallback;
import org.apache.bookkeeper.client.BKException;
import org.apache.bookkeeper.client.BookKeeper;
@@ -145,6 +148,117 @@ public class ManagedLedgerTest extends MockedBookKeeperTestCase {
return new Object[][] { { Boolean.TRUE }, { Boolean.FALSE } };
}
+ private void makeAddEntryTimeout(ManagedLedgerImpl ml, AtomicBoolean addEntryFinished) throws Exception {
+ LedgerHandle currentLedger = ml.currentLedger;
+ final LedgerHandle spyLedgerHandle = spy(currentLedger);
+ doAnswer(invocation -> {
+ ByteBuf bs = (ByteBuf) invocation.getArguments()[0];
+ AddCallback addCallback = (AddCallback) invocation.getArguments()[1];
+ Object originalContext = invocation.getArguments()[2];
+ currentLedger.asyncAddEntry(bs, (rc, lh, entryId, ctx) -> {
+ addEntryFinished.set(true);
+ addCallback.addComplete(BKException.Code.TimeoutException, spyLedgerHandle, -1, ctx);
+ }, originalContext);
+ return null;
+ }).when(spyLedgerHandle).asyncAddEntry(any(ByteBuf.class), any(AddCallback.class), any());
+ ml.currentLedger = spyLedgerHandle;
+ }
+
+ @Data
+ private static class DeleteLedgerInfo{
+ volatile boolean hasCalled;
+ volatile CompletableFuture<Void> future = new CompletableFuture<>();
+ }
+
+ private DeleteLedgerInfo makeDelayIfDoLedgerDelete(LedgerHandle ledger, final AtomicBoolean signal,
+ BookKeeper spyBookKeeper) {
+ DeleteLedgerInfo deleteLedgerInfo = new DeleteLedgerInfo();
+ doAnswer(invocation -> {
+ long ledgerId = (long) invocation.getArguments()[0];
+ AsyncCallback.DeleteCallback originalCb = (AsyncCallback.DeleteCallback) invocation.getArguments()[1];
+ AsyncCallback.DeleteCallback cb = (rc, ctx) -> {
+ if (deleteLedgerInfo.hasCalled) {
+ deleteLedgerInfo.future.complete(null);
+ }
+ originalCb.deleteComplete(rc, ctx);
+ };
+ Object ctx = invocation.getArguments()[2];
+ if (ledgerId != ledger.getId()){
+ bkc.asyncDeleteLedger(ledgerId, originalCb, ctx);
+ } else {
+ deleteLedgerInfo.hasCalled = true;
+ new Thread(() -> {
+ Awaitility.await().atMost(Duration.ofSeconds(60)).until(signal::get);
+ bkc.asyncDeleteLedger(ledgerId, cb, ctx);
+ }).start();
+ }
+ return null;
+ }).when(spyBookKeeper).asyncDeleteLedger(any(long.class), any(AsyncCallback.DeleteCallback.class), any());
+ return deleteLedgerInfo;
+ }
+
+ /***
+ * This test simulates the following problems that can occur when ZK connections are unstable:
+ * - add entry timeout
+ * - write ZK fail when update ledger info of ML
+ * and verifies that ledger info of ML is still correct when the above problems occur.
+ */
+ @Test
+ public void testLedgerInfoMetaCorrectIfAddEntryTimeOut() throws Exception {
+ String mlName = "testLedgerInfoMetaCorrectIfAddEntryTimeOut";
+ BookKeeper spyBookKeeper = spy(bkc);
+ ManagedLedgerFactoryImpl factory = new ManagedLedgerFactoryImpl(metadataStore, spyBookKeeper);
+ ManagedLedgerImpl ml = (ManagedLedgerImpl) factory.open(mlName);
+
+ // Make add entry timeout(The data write was actually successful).
+ AtomicBoolean addEntryFinished = new AtomicBoolean(false);
+ makeAddEntryTimeout(ml, addEntryFinished);
+
+ // Make the update operation of ledger info failure when switch ledger.
+ metadataStore.failConditional(new MetadataStoreException.BadVersionException(""), (opType, path) -> {
+ if (opType == FaultInjectionMetadataStore.OperationType.PUT && addEntryFinished.get()
+ && "/managed-ledgers/testLedgerInfoMetaCorrectIfAddEntryTimeOut".equals(path)) {
+ return true;
+ }
+ return false;
+ });
+
+ // Make delete ledger is delayed if delete is called.
+ AtomicBoolean deleteLedgerDelaySignal = new AtomicBoolean(false);
+ DeleteLedgerInfo deleteLedgerInfo =
+ makeDelayIfDoLedgerDelete(ml.currentLedger, deleteLedgerDelaySignal, spyBookKeeper);
+
+ // Add one entry.
+ // - it will fail and trigger ledger switch(we mocked the error).
+ // - ledger switch will also fail(we mocked the error).
+ try {
+ ml.addEntry("1".getBytes(Charset.defaultCharset()));
+ fail("Expected the operation of add entry will fail by timeout or ledger fenced.");
+ } catch (Exception e){
+ // expected ex.
+ }
+
+ // Reopen ML.
+ try {
+ ml.close();
+ fail("Expected the operation of ml close will fail by fenced state.");
+ } catch (Exception e){
+ // expected ex.
+ }
+ ManagedLedgerImpl mlReopened = (ManagedLedgerImpl) factory.open(mlName);
+ deleteLedgerDelaySignal.set(true);
+ if (deleteLedgerInfo.hasCalled){
+ deleteLedgerInfo.future.join();
+ }
+ mlReopened.close();
+
+ // verify: all ledgers in ledger info is worked.
+ for (long ledgerId : mlReopened.getLedgersInfo().keySet()){
+ LedgerHandle lh = bkc.openLedger(ledgerId, ml.digestType, ml.getConfig().getPassword());
+ lh.close();
+ }
+ }
+
@Test
public void managedLedgerApi() throws Exception {
ManagedLedger ledger = factory.open("my_test_ledger");
@@ -3125,7 +3239,7 @@ public class ManagedLedgerTest extends MockedBookKeeperTestCase {
ledger.pendingAddEntries.add(op);
}
- ledger.updateLedgersIdsComplete(mock(Stat.class));
+ ledger.updateLedgersIdsComplete(null);
for (int i = 0; i < 10; i++) {
OpAddEntry oldOp = oldOps.get(i);
if (i > 4) {