[improve][broker] Add config fsyncEnable for RocksdbMetadataStore

conf/standalone.conf

@@ -29,7 +29,6 @@ metadataStoreUrl=
 # Configuration file path for metadata store. It's supported by RocksdbMetadataStore and EtcdMetadataStore for now
 metadataStoreConfigPath=
 
-
 # The metadata store URL for the configuration data. If empty, we fall back to use metadataStoreUrl
 configurationMetadataStoreUrl=
 

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java

@@ -512,7 +512,6 @@ The delayed message index bucket time step(in seconds) in per bucket snapshot se
     )
     private String metadataStoreConfigPath = null;
 
-
     @FieldContext(
             dynamic = true,
             category = CATEGORY_SERVER,

pulsar-metadata/src/main/java/org/apache/pulsar/metadata/api/MetadataStoreConfig.java

@@ -81,6 +81,12 @@ public class MetadataStoreConfig {
     @Builder.Default
     private final String metadataStoreName = "";
 
+    /**
+     * Whether we should enable fsync for local metadata store, It's supported by RocksdbMetadataStore for now.
+     */
+    @Builder.Default
+    private final boolean fsyncEnable = true;
+
     /**
      * Pluggable MetadataEventSynchronizer to sync metadata events across the
      * separate clusters.

pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/RocksdbMetadataStore.java

@@ -87,8 +87,7 @@ public class RocksdbMetadataStore extends AbstractMetadataStore {
     private final ReentrantReadWriteLock dbStateLock;
     private volatile State state;
 
-    private final WriteOptions optionSync;
-    private final WriteOptions optionDontSync;
+    private final WriteOptions writeOptions;
     private final ReadOptions optionCache;
     private final ReadOptions optionDontCache;
     private MetadataEventSynchronizer synchronizer;
@@ -235,8 +234,7 @@ private RocksdbMetadataStore(String metadataURL, MetadataStoreConfig metadataSto
 
         db = openDB(dataPath.toString(), metadataStoreConfig.getConfigFilePath());
 
-        this.optionSync = new WriteOptions().setSync(true);
-        this.optionDontSync = new WriteOptions().setSync(false);
+        this.writeOptions = new WriteOptions().setSync(metadataStoreConfig.isFsyncEnable());
         this.optionCache = new ReadOptions().setFillCache(true);
         this.optionDontCache = new ReadOptions().setFillCache(false);
 
@@ -261,7 +259,7 @@ private long loadInstanceId() throws RocksDBException {
         } else {
             instanceId = 0;
         }
-        db.put(optionSync, INSTANCE_ID_KEY, toBytes(instanceId));
+        db.put(writeOptions, INSTANCE_ID_KEY, toBytes(instanceId));
         return instanceId;
     }
 
@@ -271,7 +269,7 @@ private AtomicLong loadSequentialIdGenerator() throws RocksDBException {
         if (value != null) {
             generator.set(toLong(value));
         } else {
-            db.put(optionSync, INSTANCE_ID_KEY, toBytes(generator.get()));
+            db.put(writeOptions, INSTANCE_ID_KEY, toBytes(generator.get()));
         }
         return generator;
     }
@@ -369,8 +367,7 @@ public synchronized void close() throws MetadataStoreException {
             state = State.CLOSED;
             log.info("close.instanceId={}", instanceId);
             db.close();
-            optionSync.close();
-            optionDontSync.close();
+            writeOptions.close();
             optionCache.close();
             optionDontCache.close();
             super.close();
@@ -496,7 +493,7 @@ protected CompletableFuture<Void> storeDelete(String path, Optional<Long> expect
             if (state == State.CLOSED) {
                 throw new MetadataStoreException.AlreadyClosedException("");
             }
-            try (Transaction transaction = db.beginTransaction(optionSync)) {
+            try (Transaction transaction = db.beginTransaction(writeOptions)) {
                 byte[] pathBytes = toBytes(path);
                 byte[] oldValueData = transaction.getForUpdate(optionDontCache, pathBytes, true);
                 MetaValue metaValue = MetaValue.parse(oldValueData);
@@ -535,7 +532,7 @@ protected CompletableFuture<Stat> storePut(String path, byte[] data, Optional<Lo
             if (state == State.CLOSED) {
                 throw new MetadataStoreException.AlreadyClosedException("");
             }
-            try (Transaction transaction = db.beginTransaction(optionSync)) {
+            try (Transaction transaction = db.beginTransaction(writeOptions)) {
                 byte[] pathBytes = toBytes(path);
                 byte[] oldValueData = transaction.getForUpdate(optionDontCache, pathBytes, true);
                 MetaValue metaValue = MetaValue.parse(oldValueData);

pulsar-metadata/src/test/java/org/apache/pulsar/metadata/CounterTest.java

@@ -25,7 +25,6 @@
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertNotEquals;
 import static org.testng.Assert.fail;
-
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -45,7 +44,7 @@ public class CounterTest extends BaseMetadataStoreTest {
     public void basicTest(String provider, Supplier<String> urlSupplier) throws Exception {
         @Cleanup
         MetadataStoreExtended store = MetadataStoreExtended.create(urlSupplier.get(),
-                MetadataStoreConfig.builder().build());
+                MetadataStoreConfig.builder().fsyncEnable(false).build());
 
         @Cleanup
         CoordinationService cs1 = new CoordinationServiceImpl(store);
@@ -102,7 +101,7 @@ public void testCounterDoesNotAutoReset(String provider, Supplier<String> urlSup
     public void testGetNextCounterRetry(String provider, Supplier<String> urlSupplier) throws Exception {
         @Cleanup
         MetadataStoreExtended store = MetadataStoreExtended.create(urlSupplier.get(),
-                MetadataStoreConfig.builder().build());
+                MetadataStoreConfig.builder().fsyncEnable(false).build());
 
         MetadataStoreExtended spy = spy(store);
 

pulsar-metadata/src/test/java/org/apache/pulsar/metadata/LeaderElectionTest.java

@@ -43,7 +43,7 @@ public class LeaderElectionTest extends BaseMetadataStoreTest {
     public void basicTest(String provider, Supplier<String> urlSupplier) throws Exception {
         @Cleanup
         MetadataStoreExtended store = MetadataStoreExtended.create(urlSupplier.get(),
-                MetadataStoreConfig.builder().build());
+                MetadataStoreConfig.builder().fsyncEnable(false).build());
 
         @Cleanup
         CoordinationService coordinationService = new CoordinationServiceImpl(store);
@@ -133,7 +133,7 @@ public void multipleMembers(String provider, Supplier<String> urlSupplier) throw
     public void leaderNodeIsDeletedExternally(String provider, Supplier<String> urlSupplier) throws Exception {
         @Cleanup
         MetadataStoreExtended store = MetadataStoreExtended.create(urlSupplier.get(),
-                MetadataStoreConfig.builder().build());
+                MetadataStoreConfig.builder().fsyncEnable(false).build());
 
         @Cleanup
         CoordinationService coordinationService = new CoordinationServiceImpl(store);
@@ -161,7 +161,7 @@ public void leaderNodeIsDeletedExternally(String provider, Supplier<String> urlS
     public void closeAll(String provider, Supplier<String> urlSupplier) throws Exception {
         @Cleanup
         MetadataStoreExtended store = MetadataStoreExtended.create(urlSupplier.get(),
-                MetadataStoreConfig.builder().build());
+                MetadataStoreConfig.builder().fsyncEnable(false).build());
         MetadataCache<String> cache = store.getMetadataCache(String.class);
 
         CoordinationService cs = new CoordinationServiceImpl(store);
@@ -191,7 +191,7 @@ public void closeAll(String provider, Supplier<String> urlSupplier) throws Excep
     public void revalidateLeaderWithinSameSession(String provider, Supplier<String> urlSupplier) throws Exception {
         @Cleanup
         MetadataStoreExtended store = MetadataStoreExtended.create(urlSupplier.get(),
-                MetadataStoreConfig.builder().build());
+                MetadataStoreConfig.builder().fsyncEnable(false).build());
 
         String path = newKey();
 

pulsar-metadata/src/test/java/org/apache/pulsar/metadata/LockManagerTest.java

@@ -53,7 +53,7 @@ public class LockManagerTest extends BaseMetadataStoreTest {
     public void acquireLocks(String provider, Supplier<String> urlSupplier) throws Exception {
         @Cleanup
         MetadataStoreExtended store = MetadataStoreExtended.create(urlSupplier.get(),
-                MetadataStoreConfig.builder().build());
+                MetadataStoreConfig.builder().fsyncEnable(false).build());
 
         @Cleanup
         CoordinationService coordinationService = new CoordinationServiceImpl(store);
@@ -104,7 +104,7 @@ public void acquireLocks(String provider, Supplier<String> urlSupplier) throws E
     public void cleanupOnClose(String provider, Supplier<String> urlSupplier) throws Exception {
         @Cleanup
         MetadataStoreExtended store = MetadataStoreExtended.create(urlSupplier.get(),
-                MetadataStoreConfig.builder().build());
+                MetadataStoreConfig.builder().fsyncEnable(false).build());
 
         @Cleanup
         CoordinationService coordinationService = new CoordinationServiceImpl(store);
@@ -135,7 +135,7 @@ public void cleanupOnClose(String provider, Supplier<String> urlSupplier) throws
     public void updateValue(String provider, Supplier<String> urlSupplier) throws Exception {
         @Cleanup
         MetadataStoreExtended store = MetadataStoreExtended.create(urlSupplier.get(),
-                MetadataStoreConfig.builder().build());
+                MetadataStoreConfig.builder().fsyncEnable(false).build());
 
         MetadataCache<String> cache = store.getMetadataCache(String.class);
 
@@ -159,7 +159,7 @@ public void updateValue(String provider, Supplier<String> urlSupplier) throws Ex
     public void updateValueWhenVersionIsOutOfSync(String provider, Supplier<String> urlSupplier) throws Exception {
         @Cleanup
         MetadataStoreExtended store = MetadataStoreExtended.create(urlSupplier.get(),
-                MetadataStoreConfig.builder().build());
+                MetadataStoreConfig.builder().fsyncEnable(false).build());
 
         MetadataCache<String> cache = store.getMetadataCache(String.class);
 
@@ -187,7 +187,7 @@ public void updateValueWhenVersionIsOutOfSync(String provider, Supplier<String>
     public void updateValueWhenKeyDisappears(String provider, Supplier<String> urlSupplier) throws Exception {
         @Cleanup
         MetadataStoreExtended store = MetadataStoreExtended.create(urlSupplier.get(),
-                MetadataStoreConfig.builder().build());
+                MetadataStoreConfig.builder().fsyncEnable(false).build());
 
         MetadataCache<String> cache = store.getMetadataCache(String.class);
 
@@ -213,7 +213,7 @@ public void updateValueWhenKeyDisappears(String provider, Supplier<String> urlSu
     public void revalidateLockWithinSameSession(String provider, Supplier<String> urlSupplier) throws Exception {
         @Cleanup
         MetadataStoreExtended store = MetadataStoreExtended.create(urlSupplier.get(),
-                MetadataStoreConfig.builder().build());
+                MetadataStoreConfig.builder().fsyncEnable(false).build());
 
         @Cleanup
         CoordinationService cs2 = new CoordinationServiceImpl(store);

pulsar-metadata/src/test/java/org/apache/pulsar/metadata/MetadataCacheTest.java

@@ -245,7 +245,8 @@ public void insertionDeletionWitGenericType(String provider, Supplier<String> ur
     @Test(dataProvider = "impl")
     public void insertionDeletion(String provider, Supplier<String> urlSupplier) throws Exception {
         @Cleanup
-        MetadataStore store = MetadataStoreFactory.create(urlSupplier.get(), MetadataStoreConfig.builder().build());
+        MetadataStore store = MetadataStoreFactory.create(urlSupplier.get(),
+                MetadataStoreConfig.builder().fsyncEnable(false).build());
         MetadataCache<MyClass> objCache = store.getMetadataCache(MyClass.class);
 
         String key1 = newKey();

pulsar-metadata/src/test/java/org/apache/pulsar/metadata/MetadataStoreTest.java

@@ -60,7 +60,8 @@ public class MetadataStoreTest extends BaseMetadataStoreTest {
     @Test(dataProvider = "impl")
     public void emptyStoreTest(String provider, Supplier<String> urlSupplier) throws Exception {
         @Cleanup
-        MetadataStore store = MetadataStoreFactory.create(urlSupplier.get(), MetadataStoreConfig.builder().build());
+        MetadataStore store = MetadataStoreFactory.create(urlSupplier.get(),
+                MetadataStoreConfig.builder().fsyncEnable(false).build());
 
         assertFalse(store.exists("/non-existing-key").join());
         assertFalse(store.exists("/non-existing-key/child").join());
@@ -89,7 +90,8 @@ public void emptyStoreTest(String provider, Supplier<String> urlSupplier) throws
     @Test(dataProvider = "impl")
     public void concurrentPutTest(String provider, Supplier<String> urlSupplier) throws Exception {
         @Cleanup
-        MetadataStore store = MetadataStoreFactory.create(urlSupplier.get(), MetadataStoreConfig.builder().build());
+        MetadataStore store = MetadataStoreFactory.create(urlSupplier.get(),
+                MetadataStoreConfig.builder().fsyncEnable(false).build());
 
         String data = "data";
         String path = "/non-existing-key";
@@ -109,7 +111,8 @@ public void concurrentPutTest(String provider, Supplier<String> urlSupplier) thr
     @Test(dataProvider = "impl")
     public void insertionTestWithExpectedVersion(String provider, Supplier<String> urlSupplier) throws Exception {
         @Cleanup
-        MetadataStore store = MetadataStoreFactory.create(urlSupplier.get(), MetadataStoreConfig.builder().build());
+        MetadataStore store = MetadataStoreFactory.create(urlSupplier.get(),
+                MetadataStoreConfig.builder().fsyncEnable(false).build());
 
         String key1 = newKey();
 
@@ -167,7 +170,8 @@ public void insertionTestWithExpectedVersion(String provider, Supplier<String> u
     @Test(dataProvider = "impl")
     public void getChildrenTest(String provider, Supplier<String> urlSupplier) throws Exception {
         @Cleanup
-        MetadataStore store = MetadataStoreFactory.create(urlSupplier.get(), MetadataStoreConfig.builder().build());
+        MetadataStore store = MetadataStoreFactory.create(urlSupplier.get(),
+                MetadataStoreConfig.builder().fsyncEnable(false).build());
 
         String key = newKey();
         int n = 10;
@@ -218,7 +222,8 @@ public void navigateChildrenTest(String provider, Supplier<String> urlSupplier)
     @Test(dataProvider = "impl")
     public void deletionTest(String provider, Supplier<String> urlSupplier) throws Exception {
         @Cleanup
-        MetadataStore store = MetadataStoreFactory.create(urlSupplier.get(), MetadataStoreConfig.builder().build());
+        MetadataStore store = MetadataStoreFactory.create(urlSupplier.get(),
+                MetadataStoreConfig.builder().fsyncEnable(false).build());
 
         String key = newKey();
         int n = 10;
@@ -252,7 +257,8 @@ public void deletionTest(String provider, Supplier<String> urlSupplier) throws E
     @Test(dataProvider = "impl")
     public void emptyKeyTest(String provider, Supplier<String> urlSupplier) throws Exception {
         @Cleanup
-        MetadataStore store = MetadataStoreFactory.create(urlSupplier.get(), MetadataStoreConfig.builder().build());
+        MetadataStore store = MetadataStoreFactory.create(urlSupplier.get(),
+                MetadataStoreConfig.builder().fsyncEnable(false).build());
 
         try {
             store.delete("", Optional.empty()).join();
@@ -293,7 +299,8 @@ public void emptyKeyTest(String provider, Supplier<String> urlSupplier) throws E
     @Test(dataProvider = "impl")
     public void notificationListeners(String provider, Supplier<String> urlSupplier) throws Exception {
         @Cleanup
-        MetadataStore store = MetadataStoreFactory.create(urlSupplier.get(), MetadataStoreConfig.builder().build());
+        MetadataStore store = MetadataStoreFactory.create(urlSupplier.get(),
+                MetadataStoreConfig.builder().fsyncEnable(false).build());
 
         BlockingQueue<Notification> notifications = new LinkedBlockingDeque<>();
         store.registerListener(n -> {
@@ -359,7 +366,8 @@ public void notificationListeners(String provider, Supplier<String> urlSupplier)
     @Test(dataProvider = "impl")
     public void testDeleteRecursive(String provider, Supplier<String> urlSupplier) throws Exception {
         @Cleanup
-        MetadataStore store = MetadataStoreFactory.create(urlSupplier.get(), MetadataStoreConfig.builder().build());
+        MetadataStore store = MetadataStoreFactory.create(urlSupplier.get(),
+                MetadataStoreConfig.builder().fsyncEnable(false).build());
 
         String prefix = newKey();
 
@@ -384,7 +392,8 @@ public void testDeleteRecursive(String provider, Supplier<String> urlSupplier) t
     @Test(dataProvider = "impl")
     public void testDeleteUnusedDirectories(String provider, Supplier<String> urlSupplier) throws Exception {
         @Cleanup
-        MetadataStore store = MetadataStoreFactory.create(urlSupplier.get(), MetadataStoreConfig.builder().build());
+        MetadataStore store = MetadataStoreFactory.create(urlSupplier.get(),
+                MetadataStoreConfig.builder().fsyncEnable(false).build());
 
         String prefix = newKey();
 
@@ -413,7 +422,8 @@ public void testDeleteUnusedDirectories(String provider, Supplier<String> urlSup
     @Test(dataProvider = "impl")
     public void testPersistent(String provider, Supplier<String> urlSupplier) throws Exception {
         String metadataUrl = urlSupplier.get();
-        MetadataStore store = MetadataStoreFactory.create(metadataUrl, MetadataStoreConfig.builder().build());
+        MetadataStore store =
+                MetadataStoreFactory.create(metadataUrl, MetadataStoreConfig.builder().fsyncEnable(false).build());
         byte[] data = "testPersistent".getBytes(StandardCharsets.UTF_8);
 
         String key = newKey() + "/a/b/c";
@@ -429,7 +439,8 @@ public void testPersistent(String provider, Supplier<String> urlSupplier) throws
 
     @Test(dataProvider = "impl")
     public void testConcurrentPutGetOneKey(String provider, Supplier<String> urlSupplier) throws Exception {
-        MetadataStore store = MetadataStoreFactory.create(urlSupplier.get(), MetadataStoreConfig.builder().build());
+        MetadataStore store = MetadataStoreFactory.create(urlSupplier.get(),
+                MetadataStoreConfig.builder().fsyncEnable(false).build());
         byte[] data = new byte[]{0};
         String path = newKey();
         int maxValue = 100;
@@ -474,7 +485,8 @@ public void run() {
     @Test(dataProvider = "impl")
     public void testConcurrentPut(String provider, Supplier<String> urlSupplier) throws Exception {
         @Cleanup
-        MetadataStore store = MetadataStoreFactory.create(urlSupplier.get(), MetadataStoreConfig.builder().build());
+        MetadataStore store = MetadataStoreFactory.create(urlSupplier.get(),
+                MetadataStoreConfig.builder().fsyncEnable(false).build());
 
         String k = newKey();
         CompletableFuture<Void> f1 =
@@ -489,7 +501,8 @@ public void testConcurrentPut(String provider, Supplier<String> urlSupplier) thr
     @Test(dataProvider = "impl")
     public void testConcurrentDelete(String provider, Supplier<String> urlSupplier) throws Exception {
         @Cleanup
-        MetadataStore store = MetadataStoreFactory.create(urlSupplier.get(), MetadataStoreConfig.builder().build());
+        MetadataStore store = MetadataStoreFactory.create(urlSupplier.get(),
+                MetadataStoreConfig.builder().fsyncEnable(false).build());
 
         String k = newKey();
         store.put(k, new byte[0], Optional.of(-1L)).join();
@@ -505,7 +518,8 @@ public void testConcurrentDelete(String provider, Supplier<String> urlSupplier)
     @Test(dataProvider = "impl")
     public void testGetChildren(String provider, Supplier<String> urlSupplier) throws Exception {
         @Cleanup
-        MetadataStore store = MetadataStoreFactory.create(urlSupplier.get(), MetadataStoreConfig.builder().build());
+        MetadataStore store = MetadataStoreFactory.create(urlSupplier.get(),
+                MetadataStoreConfig.builder().fsyncEnable(false).build());
 
         store.put("/a/a-1", "value1".getBytes(StandardCharsets.UTF_8), Optional.empty()).join();
         store.put("/a/a-2", "value1".getBytes(StandardCharsets.UTF_8), Optional.empty()).join();