Skip to content

[Metadata] Fix zk callback thread #13809

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
nodece wants to merge 1 commit into from
Closed

[Metadata] Fix zk callback thread #13809

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
24 changes: 6 additions & 18 deletions pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/AbstractMetadataStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,9 +31,9 @@
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.stream.Collectors;
Expand All @@ -60,7 +60,7 @@ public abstract class AbstractMetadataStore implements MetadataStoreExtended, Co

private final CopyOnWriteArrayList<Consumer<Notification>> listeners = new CopyOnWriteArrayList<>();
private final CopyOnWriteArrayList<Consumer<SessionEvent>> sessionListeners = new CopyOnWriteArrayList<>();
protected final ScheduledExecutorService executor;
private final ExecutorService listenerNotificationExecutor;
private final AsyncLoadingCache<String, List<String>> childrenCache;
private final AsyncLoadingCache<String, Boolean> existsCache;
private final CopyOnWriteArrayList<MetadataCacheImpl<?>> metadataCaches = new CopyOnWriteArrayList<>();
Expand All @@ -75,8 +75,8 @@ public abstract class AbstractMetadataStore implements MetadataStoreExtended, Co
protected abstract CompletableFuture<Boolean> existsFromStore(String path);

protected AbstractMetadataStore() {
this.executor = Executors
.newSingleThreadScheduledExecutor(new DefaultThreadFactory("metadata-store"));
this.listenerNotificationExecutor = Executors
.newSingleThreadExecutor(new DefaultThreadFactory("metadata-store-listener-notification"));
registerListener(this);

this.childrenCache = Caffeine.newBuilder()
Expand Down Expand Up @@ -190,7 +190,7 @@ protected CompletableFuture<Void> receivedNotification(Notification notification
});

return null;
}, executor);
}, listenerNotificationExecutor);
} catch (RejectedExecutionException e) {
return FutureUtil.failedFuture(e);
}
Expand Down Expand Up @@ -301,8 +301,7 @@ protected void receivedSessionEvent(SessionEvent event) {

@Override
public void close() throws Exception {
executor.shutdownNow();
executor.awaitTermination(10, TimeUnit.SECONDS);
// noop
}

@VisibleForTesting
Expand All @@ -311,17 +310,6 @@ public void invalidateAll() {
existsCache.synchronous().invalidateAll();
}

/**
* Run the task in the executor thread and fail the future if the executor is shutting down.
*/
protected void execute(Runnable task, CompletableFuture<?> future) {
try {
executor.execute(task);
} catch (Throwable t) {
future.completeExceptionally(t);
}
}

protected static String parent(String path) {
int idx = path.lastIndexOf('/');
if (idx <= 0) {
Expand Down
123 changes: 72 additions & 51 deletions pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/ZKMetadataStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
Expand Down Expand Up @@ -122,20 +123,23 @@ protected void receivedSessionEvent(SessionEvent event) {
// Recreate the persistent watch on the new session
zkc.addWatch("/", this::handleWatchEvent, AddWatchMode.PERSISTENT_RECURSIVE,
(rc, path, ctx) -> {
if (rc == Code.OK.intValue()) {
super.receivedSessionEvent(event);
} else {
log.error("Failed to recreate persistent watch on ZooKeeper: {}", Code.get(rc));
sessionWatcher.ifPresent(ZKSessionWatcher::setSessionInvalid);
// On the reconnectable client, mark the session as expired to trigger a new reconnect and
// we will have the chance to set the watch again.
if (zkc instanceof PulsarZooKeeperClient) {
((PulsarZooKeeperClient) zkc).process(
new WatchedEvent(Watcher.Event.EventType.None,
Watcher.Event.KeeperState.Expired,
null));
}
}
execute(() -> {
if (rc == Code.OK.intValue()) {
super.receivedSessionEvent(event);
} else {
log.error("Failed to recreate persistent watch on ZooKeeper: {}", Code.get(rc));
sessionWatcher.ifPresent(ZKSessionWatcher::setSessionInvalid);
// On the reconnectable client, mark the session as expired to trigger a new
// reconnect and
// we will have the chance to set the watch again.
if (zkc instanceof PulsarZooKeeperClient) {
((PulsarZooKeeperClient) zkc).process(
new WatchedEvent(Watcher.Event.EventType.None,
Watcher.Event.KeeperState.Expired,
null));
}
}
});
}, null);
} else {
super.receivedSessionEvent(event);
Expand All @@ -146,48 +150,50 @@ protected void receivedSessionEvent(SessionEvent event) {
protected void batchOperation(List<MetadataOp> ops) {
try {
zkc.multi(ops.stream().map(this::convertOp).collect(Collectors.toList()), (rc, path, ctx, results) -> {
if (results == null) {
Code code = Code.get(rc);
if (code == Code.CONNECTIONLOSS) {
// There is the chance that we caused a connection reset by sending or requesting a batch
// that passed the max ZK limit. Retry with the individual operations
executor.schedule(() -> {
ops.forEach(o -> batchOperation(Collections.singletonList(o)));
}, 100, TimeUnit.MILLISECONDS);
} else {
MetadataStoreException e = getException(code, path);
ops.forEach(o -> o.getFuture().completeExceptionally(e));
execute(() -> {
if (results == null) {
Code code = Code.get(rc);
if (code == Code.CONNECTIONLOSS) {
// There is the chance that we caused a connection reset by sending or requesting a batch
// that passed the max ZK limit. Retry with the individual operations
batchScheduleExecutor.schedule(() -> {
ops.forEach(o -> batchOperation(Collections.singletonList(o)));
}, 100, TimeUnit.MILLISECONDS);
} else {
MetadataStoreException e = getException(code, path);
ops.forEach(o -> o.getFuture().completeExceptionally(e));
}
return;
}
return;
}

// Trigger all the futures in the batch
for (int i = 0; i < ops.size(); i++) {
OpResult opr = results.get(i);
MetadataOp op = ops.get(i);

switch (op.getType()) {
case PUT:
handlePutResult(op.asPut(), opr);
break;
case DELETE:
handleDeleteResult(op.asDelete(), opr);
break;
case GET:
handleGetResult(op.asGet(), opr);
break;
case GET_CHILDREN:
handleGetChildrenResult(op.asGetChildren(), opr);
break;

default:
op.getFuture().completeExceptionally(new MetadataStoreException(
"Operation type not supported in multi: " + op.getType()));
// Trigger all the futures in the batch
for (int i = 0; i < ops.size(); i++) {
OpResult opr = results.get(i);
MetadataOp op = ops.get(i);

switch (op.getType()) {
case PUT:
handlePutResult(op.asPut(), opr);
break;
case DELETE:
handleDeleteResult(op.asDelete(), opr);
break;
case GET:
handleGetResult(op.asGet(), opr);
break;
case GET_CHILDREN:
handleGetChildrenResult(op.asGetChildren(), opr);
break;

default:
op.getFuture().completeExceptionally(new MetadataStoreException(
"Operation type not supported in multi: " + op.getType()));
}
}
}
});
}, null);
} catch (Throwable t) {
ops.forEach(o -> o.getFuture().completeExceptionally(new MetadataStoreException(t)));
execute(() -> ops.forEach(o -> o.getFuture().completeExceptionally(new MetadataStoreException(t))));
}
}

Expand Down Expand Up @@ -562,4 +568,19 @@ private static void createFullPathOptimistic(ZooKeeper zkc, String path, byte[]
throw KeeperException.create(Code.get(rc.get()));
}
}

private void execute(Runnable command) {
execute(command, null);
}

private void execute(Runnable command, CompletableFuture<?> future) {
try{
ForkJoinPool.commonPool().execute(command);
}catch (Exception e) {
log.warn("failed to execute ZK callback", e);
if (future !=null){
future.completeExceptionally(e);
}
}
}
}
17 changes: 14 additions & 3 deletions .../src/main/java/org/apache/pulsar/metadata/impl/batching/AbstractBatchedMetadataStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,12 +18,15 @@
*/
package org.apache.pulsar.metadata.impl.batching;

import io.netty.util.concurrent.DefaultThreadFactory;
import java.util.ArrayList;
import java.util.Collections;
import java.util.EnumSet;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
Expand All @@ -50,6 +53,8 @@ public abstract class AbstractBatchedMetadataStore extends AbstractMetadataStore
private final int maxOperations;
private final int maxSize;

protected final ScheduledExecutorService batchScheduleExecutor;

protected AbstractBatchedMetadataStore(MetadataStoreConfig conf) {
super();

Expand All @@ -61,9 +66,12 @@ protected AbstractBatchedMetadataStore(MetadataStoreConfig conf) {
if (enabled) {
readOps = new MpscUnboundedArrayQueue<>(10_000);
writeOps = new MpscUnboundedArrayQueue<>(10_000);
scheduledTask =
executor.scheduleAtFixedRate(this::flush, maxDelayMillis, maxDelayMillis, TimeUnit.MILLISECONDS);
batchScheduleExecutor = Executors.newSingleThreadScheduledExecutor(
new DefaultThreadFactory("metadata-store-batch-schedule"));
scheduledTask = batchScheduleExecutor.scheduleAtFixedRate(this::flush, maxDelayMillis, maxDelayMillis,
TimeUnit.MILLISECONDS);
} else {
batchScheduleExecutor = null;
scheduledTask = null;
readOps = null;
writeOps = null;
Expand All @@ -79,6 +87,9 @@ public void close() throws Exception {
writeOps.drain(op -> op.getFuture().completeExceptionally(ex));

scheduledTask.cancel(true);

batchScheduleExecutor.shutdownNow();
batchScheduleExecutor.awaitTermination(10, TimeUnit.SECONDS);
}
super.close();
}
Expand Down Expand Up @@ -151,7 +162,7 @@ private void enqueue(MessagePassingQueue<MetadataOp> queue, MetadataOp op) {
return;
}
if (queue.size() > maxOperations && flushInProgress.compareAndSet(false, true)) {
executor.execute(this::flush);
batchScheduleExecutor.execute(this::flush);
}
} else {
batchOperation(Collections.singletonList(op));
Expand Down
27 changes: 27 additions & 0 deletions pulsar-metadata/src/test/java/org/apache/pulsar/metadata/MetadataStoreTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
Expand Down Expand Up @@ -448,4 +449,30 @@ public void run() {
assertEquals(successWrites.get(), maxValue);
assertEquals(store.get(path).get().get().getValue()[0], maxValue);
}

@Test(dataProvider = "impl")
public void testMultipleCallback(String provider, Supplier<String> urlSupplier) throws Exception {
testMultipleCallback(provider, urlSupplier, true);
testMultipleCallback(provider, urlSupplier, false);
}

private void testMultipleCallback(String provider, Supplier<String> urlSupplier, boolean batching)
throws Exception {
@Cleanup
MetadataStore store = MetadataStoreFactory.create(urlSupplier.get(),
MetadataStoreConfig.builder().batchingEnabled(batching).build());

CountDownLatch countDownLatch = new CountDownLatch(1);

store.get("/").thenAccept((unused) -> {
store.get("/").join();
store.get("/").thenAccept((unused2) -> {
countDownLatch.countDown();
});
});

if (!countDownLatch.await(5, TimeUnit.SECONDS)) {
fail("failed to test multiple callback, need to check deadlock");
}
}
}