Skip to content

[fix][broker] PulsarLedgerManager to pass correct error code to BK client #16664

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
dlg99 wants to merge 3 commits into from
Closed

[fix][broker] PulsarLedgerManager to pass correct error code to BK client #16664

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
64859d5
PulsarLedgerManager to pass correct error code to BK client; handle d…
dlg99 Jul 19, 2022
8850e45
To not break code that relies on specific chain of exceptions, like e…
dlg99 Jul 19, 2022
40df0c4
CR feedback
dlg99 Jul 19, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 27 additions & 2 deletions ...ged-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -893,7 +893,19 @@ private void deleteManagedLedgerData(BookKeeper bkc, String managedLedgerName, M
DeleteLedgerCallback callback, Object ctx) {
Futures.waitForAll(info.ledgers.stream()
.filter(li -> !li.isOffloaded)
.map(li -> bkc.newDeleteLedgerOp().withLedgerId(li.ledgerId).execute())
.map(li -> bkc.newDeleteLedgerOp().withLedgerId(li.ledgerId).execute()
.handle((result, ex) -> {
if (ex != null) {
int rc = BKException.getExceptionCode(ex);
if (rc == BKException.Code.NoSuchLedgerExistsOnMetadataServerException
|| rc == BKException.Code.NoSuchLedgerExistsException) {
log.info("Ledger {} does not exist, ignoring", li.ledgerId);
return null;
Comment thread
dlg99 marked this conversation as resolved.
}
throw new CompletionException(ex);
}
return result;
}))
.collect(Collectors.toList()))
.thenRun(() -> {
// Delete the metadata
Expand Down Expand Up @@ -921,7 +933,20 @@ private CompletableFuture<Void> deleteCursor(BookKeeper bkc, String managedLedge

// Delete the cursor ledger if present
if (cursor.cursorsLedgerId != -1) {
cursorLedgerDeleteFuture = bkc.newDeleteLedgerOp().withLedgerId(cursor.cursorsLedgerId).execute();
cursorLedgerDeleteFuture = bkc.newDeleteLedgerOp().withLedgerId(cursor.cursorsLedgerId)
.execute()
.handle((result, ex) -> {
if (ex != null) {

@eolivelli eolivelli Jul 20, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

what about looking for NotFoundException in the exception chain ?
we should have some utility to traverse the chain and look for a specific class

@dlg99 dlg99 Jul 20, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This does not work, in this case PulsarLedgerManager returns NotFoundException to Bookkeeper's code which remaps it into UnexpectedConditionException for the pulsar callback.

int rc = BKException.getExceptionCode(ex);
if (rc == BKException.Code.NoSuchLedgerExistsOnMetadataServerException
|| rc == BKException.Code.NoSuchLedgerExistsException) {
log.info("Ledger {} does not exist, ignoring", cursor.cursorsLedgerId);
return null;
}
throw new CompletionException(ex);
}
return result;
});
} else {
cursorLedgerDeleteFuture = CompletableFuture.completedFuture(null);
}
Expand Down
150 changes: 149 additions & 1 deletion pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerBkEnsemblesTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@

import static org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest.retryStrategically;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNotEquals;
import static org.testng.Assert.fail;

import java.lang.reflect.Field;
Expand All @@ -31,9 +32,13 @@
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

import com.google.common.collect.Sets;
import lombok.Cleanup;

import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.client.BKException;
import org.apache.bookkeeper.client.BookKeeper;
import org.apache.bookkeeper.client.api.LedgerMetadata;
import org.apache.bookkeeper.mledger.ManagedLedgerConfig;
import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
import org.apache.bookkeeper.mledger.impl.ManagedLedgerFactoryImpl;
Expand All @@ -47,12 +52,14 @@
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.common.policies.data.TenantInfoImpl;
import org.apache.zookeeper.ZooKeeper;
import org.awaitility.Awaitility;
import org.testng.Assert;
import org.testng.annotations.Test;

@Test(groups = "broker")
@Slf4j
public class BrokerBkEnsemblesTests extends BkEnsemblesTestBase {

public BrokerBkEnsemblesTests() {
Expand Down Expand Up @@ -238,7 +245,7 @@ public void testSkipCorruptDataLedger() throws Exception {
try {
bookKeeper.deleteLedger(entry.getKey());
} catch (Exception e) {
e.printStackTrace();
log.warn("failed to delete ledger {}", entry.getKey(), e);
}
}
});
Expand Down Expand Up @@ -279,6 +286,147 @@ public void testSkipCorruptDataLedger() throws Exception {
consumer.close();
}

@Test
public void testTruncateCorruptDataLedger() throws Exception {
// Ensure intended state for autoSkipNonRecoverableData
admin.brokers().updateDynamicConfiguration("autoSkipNonRecoverableData", "false");

@Cleanup
PulsarClient client = PulsarClient.builder()
.serviceUrl(pulsar.getWebServiceAddress())
.statsInterval(0, TimeUnit.SECONDS)
.build();

final int totalMessages = 100;
final int totalDataLedgers = 5;
final int entriesPerLedger = totalMessages / totalDataLedgers;

final String tenant = "prop";
try {
admin.tenants().createTenant(tenant, new TenantInfoImpl(Sets.newHashSet("role1", "role2"),
Sets.newHashSet(config.getClusterName())));
} catch (Exception e) {
Comment thread
dlg99 marked this conversation as resolved.

}
final String ns1 = tenant + "/crash-broker";
try {
admin.namespaces().createNamespace(ns1, Sets.newHashSet(config.getClusterName()));
} catch (Exception e) {

}

final String topic1 = "persistent://" + ns1 + "/my-topic-" + System.currentTimeMillis();

// Create subscription
Consumer<byte[]> consumer = client.newConsumer().topic(topic1).subscriptionName("my-subscriber-name")
.receiverQueueSize(5).subscribe();

PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getOrCreateTopic(topic1).get();
ManagedLedgerImpl ml = (ManagedLedgerImpl) topic.getManagedLedger();
ManagedCursorImpl cursor = (ManagedCursorImpl) ml.getCursors().iterator().next();
Field configField = ManagedCursorImpl.class.getDeclaredField("config");
configField.setAccessible(true);
// Create multiple data-ledger
ManagedLedgerConfig config = (ManagedLedgerConfig) configField.get(cursor);
config.setMaxEntriesPerLedger(entriesPerLedger);
config.setMinimumRolloverTime(1, TimeUnit.MILLISECONDS);
// bookkeeper client
Field bookKeeperField = ManagedLedgerImpl.class.getDeclaredField("bookKeeper");
bookKeeperField.setAccessible(true);
// Create multiple data-ledger
BookKeeper bookKeeper = (BookKeeper) bookKeeperField.get(ml);

// (1) publish messages in 10 data-ledgers each with 20 entries under managed-ledger
Producer<byte[]> producer = client.newProducer().topic(topic1).create();
for (int i = 0; i < totalMessages; i++) {
String message = "my-message-" + i;
producer.send(message.getBytes());
}

// validate: consumer is able to consume msg and close consumer after reading 1 entry
Assert.assertNotNull(consumer.receive(1, TimeUnit.SECONDS));
consumer.close();

NavigableMap<Long, LedgerInfo> ledgerInfo = ml.getLedgersInfo();
Assert.assertEquals(ledgerInfo.size(), totalDataLedgers);
Entry<Long, LedgerInfo> lastLedger = ledgerInfo.lastEntry();
long firstLedgerToDelete = lastLedger.getKey();

// (2) delete first 4 data-ledgers
ledgerInfo.entrySet().forEach(entry -> {
if (!entry.equals(lastLedger)) {
assertEquals(entry.getValue().getEntries(), entriesPerLedger);
try {
bookKeeper.deleteLedger(entry.getKey());
} catch (Exception e) {
log.warn("failed to delete ledger {}", entry.getKey(), e);
}
}
});

// create 5 more ledgers
for (int i = 0; i < totalMessages; i++) {
String message = "my-message2-" + i;
producer.send(message.getBytes());
}

ml.delete();

// Admin should be able to truncate the topic
admin.topics().truncate(topic1);

ledgerInfo.entrySet().forEach(entry -> {
log.warn("found ledger: {}", entry.getKey());
assertNotEquals(firstLedgerToDelete, entry.getKey());
});

// Currently, ledger deletion is async and failed deletion
// does not actually fail truncation but logs an exception
// and creates scheduled task to retry
Awaitility.await().atMost(30, TimeUnit.SECONDS).untilAsserted(() -> {
LedgerMetadata meta = bookKeeper
.getLedgerMetadata(firstLedgerToDelete)
.exceptionally(e -> null)
.get();
assertEquals(null, meta, "ledger should be deleted " + firstLedgerToDelete);
});

// Should not throw, deleting absent ledger must be a noop
// unless PulsarManager returned a wrong error which
// got translated to BKUnexpectedConditionException
try {
bookKeeper.deleteLedger(firstLedgerToDelete);
} catch (BKException.BKNoSuchLedgerExistsOnMetadataServerException bke) {
// pass
}

producer.close();
consumer.close();
}

@Test
public void testDeleteLedgerFactoryCorruptLedger() throws Exception {
ManagedLedgerFactoryImpl factory = (ManagedLedgerFactoryImpl) pulsar.getManagedLedgerFactory();
ManagedLedgerImpl ml = (ManagedLedgerImpl) factory.open("test");

// bookkeeper client
Field bookKeeperField = ManagedLedgerImpl.class.getDeclaredField("bookKeeper");
bookKeeperField.setAccessible(true);
// Create multiple data-ledger
BookKeeper bookKeeper = (BookKeeper) bookKeeperField.get(ml);

ml.addEntry("dummy-entry-1".getBytes());

NavigableMap<Long, LedgerInfo> ledgerInfo = ml.getLedgersInfo();
long lastLedger = ledgerInfo.lastEntry().getKey();

ml.close();
bookKeeper.deleteLedger(lastLedger);

// BK ledger is deleted, factory should not throw on delete
factory.delete("test");
}

@Test(timeOut = 20000)
public void testTopicWithWildCardChar() throws Exception {
@Cleanup
Expand Down
50 changes: 41 additions & 9 deletions pulsar-metadata/src/main/java/org/apache/pulsar/metadata/api/MetadataStoreException.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,12 +21,35 @@
import java.io.IOException;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ExecutionException;
import org.apache.bookkeeper.client.BKException;

/**
* Generic metadata store exception.
*/
public class MetadataStoreException extends IOException {

private static Throwable makeBkFriendlyException(BKException bkException, Throwable cause) {
if (cause == null) {
return bkException;
}

Throwable lastCause = cause;
while (true) {
if (lastCause instanceof BKException) {
// BKException.getExceptionCode() traverses chain of causes
// no need to create another exception
return cause;
}
if (lastCause.getCause() == null) {
break;
}
lastCause = lastCause.getCause();
}

lastCause.initCause(bkException);
return cause;
}

public MetadataStoreException(Throwable t) {
super(t);
}
Expand Down Expand Up @@ -60,42 +83,49 @@ public InvalidImplementationException(String msg) {
* Key not found in store.
*/
public static class NotFoundException extends MetadataStoreException {
private static final BKException bkEx =
BKException.create(BKException.Code.NoSuchLedgerExistsOnMetadataServerException);

public NotFoundException() {
super((Throwable) null);
super(makeBkFriendlyException(bkEx, null));
}

public NotFoundException(Throwable t) {
super(t);
super(makeBkFriendlyException(bkEx, t));
}

public NotFoundException(String msg) {
super(msg);
super(msg, makeBkFriendlyException(bkEx, null));
}
}

/**
* Key was already in store.
*/
public static class AlreadyExistsException extends MetadataStoreException {
private static final BKException bkEx = BKException.create(BKException.Code.LedgerExistException);

public AlreadyExistsException(Throwable t) {
super(t);
super(makeBkFriendlyException(bkEx, t));
}

public AlreadyExistsException(String msg) {
super(msg);
super(msg, makeBkFriendlyException(bkEx, null));
}
}

/**
* Unsuccessful update due to mismatched expected version.
*/
public static class BadVersionException extends MetadataStoreException {
private static final BKException bkEx = BKException.create(BKException.Code.MetadataVersionException);

public BadVersionException(Throwable t) {
super(t);
super(makeBkFriendlyException(bkEx, t));
}

public BadVersionException(String msg) {
super(msg);
super(msg, makeBkFriendlyException(bkEx, null));
}
}

Expand Down Expand Up @@ -137,12 +167,14 @@ public LockBusyException(String msg) {
* The store was already closed.
*/
public static class AlreadyClosedException extends MetadataStoreException {
private static final BKException bkEx = BKException.create(BKException.Code.LedgerClosedException);

public AlreadyClosedException(Throwable t) {
super(t);
super(makeBkFriendlyException(bkEx, t));
}

public AlreadyClosedException(String msg) {
super(msg);
super(msg, makeBkFriendlyException(bkEx, null));
}
}

Expand Down