undo reflection

Signed-off-by: Sean Kao <[email protected]>

flint-core/src/main/scala/org/opensearch/flint/core/FlintClientBuilder.java

@@ -5,30 +5,15 @@
 
 package org.opensearch.flint.core;
 
-import java.lang.reflect.Constructor;
 import org.opensearch.flint.core.storage.FlintOpenSearchClient;
-import org.opensearch.flint.core.metadata.log.FlintMetadataLogService;
+import org.opensearch.flint.core.storage.FlintOpenSearchMetadataLogService;
 
 /**
  * {@link FlintClient} builder.
  */
 public class FlintClientBuilder {
 
   public static FlintClient build(FlintOptions options) {
-    return new FlintOpenSearchClient(options,
-        instantiateFlintMetadataLogService(options.getCustomFlintMetadataLogServiceClass(), options));
-  }
-
-  /**
-   * Attempts to instantiate Flint metadata log service using reflection.
-   */
-  private static FlintMetadataLogService instantiateFlintMetadataLogService(String className, FlintOptions options) {
-    try {
-      Class<?> flintMetadataLogServiceClass = Class.forName(className);
-      Constructor<?> constructor = flintMetadataLogServiceClass.getConstructor(FlintOptions.class);
-      return (FlintMetadataLogService) constructor.newInstance(options);
-    } catch (Exception e) {
-      throw new RuntimeException("Failed to instantiate FlintMetadataLogService: " + className, e);
-    }
+    return new FlintOpenSearchClient(options, new FlintOpenSearchMetadataLogService(options));
   }
 }

flint-core/src/main/scala/org/opensearch/flint/core/FlintOptions.java

@@ -90,10 +90,6 @@ public class FlintOptions implements Serializable {
 
   public static final String DEFAULT_BATCH_BYTES = "1mb";
 
-  public static final String CUSTOM_FLINT_METADATA_LOG_SERVICE_CLASS = "customFlintMetadataLogServiceClass";
-
-  public static final String DEFAULT_FLINT_METADATA_LOG_SERVICE_CLASS  = "org.opensearch.flint.core.storage.FlintOpenSearchMetadataLogService";
-
   public FlintOptions(Map<String, String> options) {
     this.options = options;
     this.retryOptions = new FlintRetryOptions(options);
@@ -166,8 +162,4 @@ public int getBatchBytes() {
     return (int) org.apache.spark.network.util.JavaUtils
         .byteStringAs(options.getOrDefault(BATCH_BYTES, DEFAULT_BATCH_BYTES), ByteUnit.BYTE);
   }
-
-  public String getCustomFlintMetadataLogServiceClass() {
-    return options.getOrDefault(CUSTOM_FLINT_METADATA_LOG_SERVICE_CLASS, DEFAULT_FLINT_METADATA_LOG_SERVICE_CLASS);
-  }
 }