You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by xi...@apache.org on 2023/07/07 12:32:56 UTC
[pulsar] branch branch-2.10 updated: [fix] [meta]Switch to the metadata store thread after zk operation (#20303)
This is an automated email from the ASF dual-hosted git repository.
xiangying pushed a commit to branch branch-2.10
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-2.10 by this push:
new 32e1463c1dc [fix] [meta]Switch to the metadata store thread after zk operation (#20303)
32e1463c1dc is described below
commit 32e1463c1dc90406ce626bc79a916d2bed540636
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Tue May 30 18:07:31 2023 +0800
[fix] [meta]Switch to the metadata store thread after zk operation (#20303)
(cherry picked from commit 35d961242089a0448cb425da637a2afbdaca0005)
---
.../metadata/impl/AbstractMetadataStore.java | 13 ++++
.../pulsar/metadata/impl/ZKMetadataStore.java | 46 +++++++-------
.../apache/pulsar/metadata/MetadataStoreTest.java | 70 ++++++++++++++++++++++
3 files changed, 106 insertions(+), 23 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 521b8bea546..c1845ba5f4f 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
@@ -36,6 +36,7 @@ import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
+import java.util.function.Supplier;
import java.util.stream.Collectors;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
@@ -328,6 +329,18 @@ public abstract class AbstractMetadataStore implements MetadataStoreExtended, Co
}
}
+ /**
+ * Run the task in the executor thread and fail the future if the executor is shutting down.
+ */
+ @VisibleForTesting
+ public void execute(Runnable task, Supplier<List<CompletableFuture<?>>> futures) {
+ try {
+ executor.execute(task);
+ } catch (final Throwable t) {
+ futures.get().forEach(f -> f.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 4676843b972..6f980d3e794 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
@@ -173,29 +173,29 @@ public class ZKMetadataStore extends AbstractBatchedMetadataStore
}
// 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()));
- }
- }
+ execute(() -> {
+ 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()));
+ }
+ }
+ }, () -> ops.stream().map(MetadataOp::getFuture).collect(Collectors.toList()));
}, null);
} catch (Throwable t) {
ops.forEach(o -> o.getFuture().completeExceptionally(new MetadataStoreException(t)));
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 a06b8e813e8..d5231284577 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
@@ -30,6 +30,7 @@ import java.util.Collections;
import java.util.List;
import java.util.Optional;
import java.util.Set;
+import java.util.UUID;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
@@ -50,8 +51,10 @@ import org.apache.pulsar.metadata.api.MetadataStoreFactory;
import org.apache.pulsar.metadata.api.Notification;
import org.apache.pulsar.metadata.api.NotificationType;
import org.apache.pulsar.metadata.api.Stat;
+import org.apache.pulsar.metadata.impl.ZKMetadataStore;
import org.assertj.core.util.Lists;
import org.awaitility.Awaitility;
+import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
@Slf4j
@@ -390,6 +393,73 @@ public class MetadataStoreTest extends BaseMetadataStoreTest {
assertFalse(store.exists(prefix).join());
}
+ @DataProvider(name = "conditionOfSwitchThread")
+ public Object[][] conditionOfSwitchThread(){
+ return new Object[][]{
+ {false, false},
+ {false, true},
+ {true, false},
+ {true, true}
+ };
+ }
+
+ @Test(dataProvider = "conditionOfSwitchThread")
+ public void testThreadSwitchOfZkMetadataStore(boolean hasSynchronizer, boolean enabledBatch) throws Exception {
+ final String prefix = newKey();
+ final String metadataStoreName = UUID.randomUUID().toString().replaceAll("-", "");
+ MetadataStoreConfig.MetadataStoreConfigBuilder builder =
+ MetadataStoreConfig.builder().metadataStoreName(metadataStoreName);
+ builder.fsyncEnable(false);
+ builder.batchingEnabled(enabledBatch);
+ if (!hasSynchronizer) {
+ builder.synchronizer(null);
+ }
+ MetadataStoreConfig config = builder.build();
+ @Cleanup
+ ZKMetadataStore store = (ZKMetadataStore) MetadataStoreFactory.create(zks.getConnectionString(), config);
+
+ final Runnable verify = () -> {
+ String currentThreadName = Thread.currentThread().getName();
+ String errorMessage = String.format("Expect to switch to thread %s, but currently it is thread %s",
+ metadataStoreName, currentThreadName);
+ assertTrue(Thread.currentThread().getName().startsWith(metadataStoreName), errorMessage);
+ };
+
+ // put with node which has parent(but the parent node is not exists).
+ store.put(prefix + "/a1/b1/c1", "value".getBytes(), Optional.of(-1L)).thenApply((ignore) -> {
+ verify.run();
+ return null;
+ }).join();
+ // put.
+ store.put(prefix + "/b1", "value".getBytes(), Optional.of(-1L)).thenApply((ignore) -> {
+ verify.run();
+ return null;
+ }).join();
+ // get.
+ store.get(prefix + "/b1").thenApply((ignore) -> {
+ verify.run();
+ return null;
+ }).join();
+ // get the node which is not exists.
+ store.get(prefix + "/non").thenApply((ignore) -> {
+ verify.run();
+ return null;
+ }).join();
+ // delete.
+ store.delete(prefix + "/b1", Optional.empty()).thenApply((ignore) -> {
+ verify.run();
+ return null;
+ }).join();
+ // delete the node which is not exists.
+ store.delete(prefix + "/non", Optional.empty()).thenApply((ignore) -> {
+ verify.run();
+ return null;
+ }).exceptionally(ex -> {
+ verify.run();
+ return null;
+ }).join();
+ }
+
@Test(dataProvider = "impl")
public void testPersistent(String provider, Supplier<String> urlSupplier) throws Exception {
String metadataUrl = urlSupplier.get();