From 85803743a1e3574c812d39d5211b94ddedc0de30 Mon Sep 17 00:00:00 2001 From: Zixuan Liu Date: Fri, 21 Jan 2022 16:20:13 +0800 Subject: [PATCH] [Metadata] Fix zk callback thread Signed-off-by: Zixuan Liu --- .../metadata/impl/AbstractMetadataStore.java | 24 +--- .../pulsar/metadata/impl/ZKMetadataStore.java | 123 ++++++++++-------- .../AbstractBatchedMetadataStore.java | 17 ++- .../pulsar/metadata/MetadataStoreTest.java | 27 ++++ 4 files changed, 119 insertions(+), 72 deletions(-) diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/AbstractMetadataStore.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/AbstractMetadataStore.java index e8230e0113ffe..8dbf2e5a75dde 100644 --- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/AbstractMetadataStore.java +++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/AbstractMetadataStore.java @@ -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; @@ -60,7 +60,7 @@ public abstract class AbstractMetadataStore implements MetadataStoreExtended, Co private final CopyOnWriteArrayList> listeners = new CopyOnWriteArrayList<>(); private final CopyOnWriteArrayList> sessionListeners = new CopyOnWriteArrayList<>(); - protected final ScheduledExecutorService executor; + private final ExecutorService listenerNotificationExecutor; private final AsyncLoadingCache> childrenCache; private final AsyncLoadingCache existsCache; private final CopyOnWriteArrayList> metadataCaches = new CopyOnWriteArrayList<>(); @@ -75,8 +75,8 @@ public abstract class AbstractMetadataStore implements MetadataStoreExtended, Co protected abstract CompletableFuture 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() @@ -190,7 +190,7 @@ protected CompletableFuture receivedNotification(Notification notification }); return null; - }, executor); + }, listenerNotificationExecutor); } catch (RejectedExecutionException e) { return FutureUtil.failedFuture(e); } @@ -301,8 +301,7 @@ protected void receivedSessionEvent(SessionEvent event) { @Override public void close() throws Exception { - executor.shutdownNow(); - executor.awaitTermination(10, TimeUnit.SECONDS); + // noop } @VisibleForTesting @@ -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) { diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/ZKMetadataStore.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/ZKMetadataStore.java index cf0b7c3d049ed..0bb2a8e01020e 100644 --- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/ZKMetadataStore.java +++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/ZKMetadataStore.java @@ -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; @@ -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); @@ -146,48 +150,50 @@ protected void receivedSessionEvent(SessionEvent event) { protected void batchOperation(List 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)))); } } @@ -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); + } + } + } } diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/batching/AbstractBatchedMetadataStore.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/batching/AbstractBatchedMetadataStore.java index 5d78282abcac3..6e94e22eec18e 100644 --- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/batching/AbstractBatchedMetadataStore.java +++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/batching/AbstractBatchedMetadataStore.java @@ -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; @@ -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(); @@ -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; @@ -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(); } @@ -151,7 +162,7 @@ private void enqueue(MessagePassingQueue 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)); diff --git a/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/MetadataStoreTest.java b/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/MetadataStoreTest.java index 672322bd44925..d0daed9c8f56f 100644 --- a/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/MetadataStoreTest.java +++ b/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/MetadataStoreTest.java @@ -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; @@ -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 urlSupplier) throws Exception { + testMultipleCallback(provider, urlSupplier, true); + testMultipleCallback(provider, urlSupplier, false); + } + + private void testMultipleCallback(String provider, Supplier 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"); + } + } }