Support to specify different config for Configuration and Local Metad…

…ata Store in oxia impl

pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/oxia/OxiaMetadataStore.java

@@ -41,6 +41,7 @@
 import java.util.concurrent.CompletionStage;
 import lombok.NonNull;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.pulsar.metadata.api.GetResult;
 import org.apache.pulsar.metadata.api.MetadataEventSynchronizer;
 import org.apache.pulsar.metadata.api.MetadataStoreConfig;
@@ -80,15 +81,17 @@ public OxiaMetadataStore(
         }
         synchronizer = Optional.ofNullable(metadataStoreConfig.getSynchronizer());
         identity = UUID.randomUUID().toString();
-        client =
-                OxiaClientBuilder.create(serviceAddress)
-                        .clientIdentifier(identity)
-                        .namespace(namespace)
-                        .sessionTimeout(Duration.ofMillis(metadataStoreConfig.getSessionTimeoutMillis()))
-                        .batchLinger(Duration.ofMillis(linger))
-                        .maxRequestsPerBatch(metadataStoreConfig.getBatchingMaxOperations())
-                        .asyncClient()
-                        .get();
+        OxiaClientBuilder oxiaClientBuilder = OxiaClientBuilder
+                .create(serviceAddress)
+                .clientIdentifier(identity)
+                .namespace(namespace)
+                .sessionTimeout(Duration.ofMillis(metadataStoreConfig.getSessionTimeoutMillis()))
+                .batchLinger(Duration.ofMillis(linger))
+                .maxRequestsPerBatch(metadataStoreConfig.getBatchingMaxOperations());
+        if (StringUtils.isNotBlank(metadataStoreConfig.getConfigFilePath())) {
+            oxiaClientBuilder.loadConfig(metadataStoreConfig.getConfigFilePath());
+        }
+        client = oxiaClientBuilder.asyncClient().get();
         init();
     }
 

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

@@ -100,7 +100,7 @@ public Object[][] distributedImplementations() {
         };
     }
 
-    private synchronized String getOxiaServerConnectString() {
+    protected synchronized String getOxiaServerConnectString() {
         if (oxiaServer == null) {
             oxiaServer = new OxiaContainer(OxiaContainer.DEFAULT_IMAGE_NAME);
             oxiaServer.start();

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

@@ -24,9 +24,11 @@
 import static org.testng.Assert.assertTrue;
 import static org.testng.Assert.fail;
 import java.nio.charset.StandardCharsets;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
 import java.util.UUID;
@@ -38,10 +40,14 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Supplier;
+
+import io.streamnative.oxia.client.ClientConfig;
+import io.streamnative.oxia.client.api.AsyncOxiaClient;
+import io.streamnative.oxia.client.session.SessionFactory;
+import io.streamnative.oxia.client.session.SessionManager;
 import lombok.Cleanup;
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.lang3.reflect.FieldUtils;
 import org.apache.pulsar.common.util.FutureUtil;
 import org.apache.pulsar.metadata.api.GetResult;
 import org.apache.pulsar.metadata.api.MetadataStore;
@@ -55,11 +61,13 @@
 import org.apache.pulsar.metadata.api.Stat;
 import org.apache.pulsar.metadata.impl.PulsarZooKeeperClient;
 import org.apache.pulsar.metadata.impl.ZKMetadataStore;
+import org.apache.pulsar.metadata.impl.oxia.OxiaMetadataStore;
 import org.apache.zookeeper.WatchedEvent;
 import org.apache.zookeeper.Watcher;
 import org.apache.zookeeper.ZooKeeper;
 import org.assertj.core.util.Lists;
 import org.awaitility.Awaitility;
+import org.awaitility.reflect.WhiteboxImpl;
 import org.testng.annotations.DataProvider;
 import org.testng.annotations.Test;
 
@@ -529,6 +537,27 @@ public void testZkLoadConfigFromFile() throws Exception {
         assertFalse(zooKeeper.getClientConfig().isSaslClientEnabled());
     }
 
+    @Test
+    public void testOxiaLoadConfigFromFile() throws Exception {
+        final String metadataStoreName = UUID.randomUUID().toString().replaceAll("-", "");
+        String oxia = "oxia://" + getOxiaServerConnectString();
+        MetadataStoreConfig.MetadataStoreConfigBuilder builder =
+                MetadataStoreConfig.builder().metadataStoreName(metadataStoreName);
+        builder.fsyncEnable(false);
+        builder.batchingEnabled(true);
+        builder.sessionTimeoutMillis(30000);
+        builder.configFilePath("src/test/resources/oxia_client.conf");
+        MetadataStoreConfig config = builder.build();
+
+        OxiaMetadataStore store = (OxiaMetadataStore) MetadataStoreFactory.create(oxia, config);
+        var client = (AsyncOxiaClient) WhiteboxImpl.getInternalState(store, "client");
+        var sessionManager = (SessionManager) WhiteboxImpl.getInternalState(client, "sessionManager");
+        var sessionFactory = (SessionFactory) WhiteboxImpl.getInternalState(sessionManager, "factory");
+        var clientConfig = (ClientConfig) WhiteboxImpl.getInternalState(sessionFactory, "config");
+        var sessionTimeout = clientConfig.sessionTimeout();
+        assertEquals(sessionTimeout, Duration.ofSeconds(60));
+    }
+
     @Test(dataProvider = "impl")
     public void testPersistent(String provider, Supplier<String> urlSupplier) throws Exception {
         String metadataUrl = urlSupplier.get();

pulsar-metadata/src/test/resources/oxia_client.conf

@@ -0,0 +1,20 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+sessionTimeout=60000