[CDAP-21172]Added cdap-metadata-ext-spanner Module with ExtensionLoader

cdap-common/src/main/java/io/cdap/cdap/common/conf/Constants.java

@@ -2114,9 +2114,13 @@ public static final class Metadata {
     public static final String STORAGE_PROVIDER_IMPLEMENTATION = "metadata.storage.implementation";
     public static final String STORAGE_PROVIDER_NOSQL = "nosql";
     public static final String STORAGE_PROVIDER_ELASTICSEARCH = "elastic";
+    public static final String STORAGE_PROVIDER_SPANNER = "gcp-spanner";
 
     public static final String METADATA_WRITER_SUBSCRIBER = "metadata.writer";
     public static final String METADATA_CONSUMER_WRITER_SUBSCRIBER = "metadata.consumer.writer";
+
+    // Metadata configs
+    public static final String METADATA_STORAGE_EXT_DIR = "metadata.storage.extensions.dir";
   }
 
   /**

cdap-common/src/main/resources/cdap-default.xml

@@ -2845,6 +2845,11 @@
     <value>/opt/cdap/master/ext/log-publisher</value>
   </property>
 
+  <property>
+    <name>metadata.storage.extensions.dir</name>
+    <value>/opt/cdap/master/ext/metadata-storage</value>
+  </property>
+
   <!-- Metrics Configuration -->
 
   <property>

cdap-data-fabric/src/main/java/io/cdap/cdap/data/runtime/DataSetsModules.java

@@ -45,6 +45,7 @@
 import io.cdap.cdap.data2.registry.UsageWriter;
 import io.cdap.cdap.metadata.elastic.ElasticsearchMetadataStorage;
 import io.cdap.cdap.security.impersonation.OwnerStore;
+import io.cdap.cdap.spi.metadata.DelegatingMetadataStorage;
 import io.cdap.cdap.spi.metadata.MetadataStorage;
 import io.cdap.cdap.spi.metadata.dataset.DatasetMetadataStorage;
 import io.cdap.cdap.spi.metadata.noop.NoopMetadataStorage;
@@ -178,11 +179,21 @@ public MetadataStorage get() {
     if (Constants.Metadata.STORAGE_PROVIDER_NOSQL.equalsIgnoreCase(config)) {
       return injector.getInstance(DatasetMetadataStorage.class);
     }
+
+    // TODO (CDAP-21173): Load elastic search implementation using DelegatingMetadataStorage
     if (Constants.Metadata.STORAGE_PROVIDER_ELASTICSEARCH.equalsIgnoreCase(config)) {
       return injector.getInstance(ElasticsearchMetadataStorage.class);
     }
-    throw new IllegalArgumentException("Unsupported MetadataStorage '" + config + "'. Only '"
-        + Constants.Metadata.STORAGE_PROVIDER_NOSQL + "' and '"
-        + Constants.Metadata.STORAGE_PROVIDER_ELASTICSEARCH + "' are allowed.");
+    if (Constants.Metadata.STORAGE_PROVIDER_SPANNER.equalsIgnoreCase(config)) {
+      return injector.getInstance(DelegatingMetadataStorage.class);
+    }
+    String errorMessage = String.format(
+      "Unsupported MetadataStorage '%s'. Only '%s', '%s' and '%s' are allowed.",
+      config,
+      Constants.Metadata.STORAGE_PROVIDER_NOSQL,
+      Constants.Metadata.STORAGE_PROVIDER_SPANNER,
+      Constants.Metadata.STORAGE_PROVIDER_ELASTICSEARCH
+    );
+    throw new IllegalArgumentException(errorMessage);
   }
 }

cdap-data-fabric/src/main/java/io/cdap/cdap/data2/metadata/AuditMetadataStorage.java

@@ -83,6 +83,11 @@ public void setAuditPublisher(AuditPublisher auditPublisher) {
     this.auditPublisher = auditPublisher;
   }
 
+  @Override
+  public String getName() {
+    return getClass().getSimpleName();
+  }
+
   @Override
   public void createIndex() throws IOException {
     try {

cdap-data-fabric/src/main/java/io/cdap/cdap/spi/metadata/DefaultMetadataStorageContext.java

@@ -0,0 +1,46 @@
+/*
+ * Copyright © 2025 Cask Data, Inc.
+ *
+ * Licensed 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.
+ */
+
+package io.cdap.cdap.spi.metadata;
+
+import io.cdap.cdap.common.conf.CConfiguration;
+import java.util.Collections;
+import java.util.Map;
+import javax.annotation.Nullable;
+
+
+/**
+ * Default implementation of the {@link MetadataStorageContext}.
+ * TODO(CDAP-21174): Generalize the context for metadata storage
+ */
+public class DefaultMetadataStorageContext implements MetadataStorageContext {
+
+  private final Map<String, String> properties;
+
+  DefaultMetadataStorageContext(CConfiguration cConf, @Nullable String prefix) {
+    if (prefix != null) {
+      this.properties = Collections.unmodifiableMap(cConf.getPropsWithPrefix(prefix));
+    } else {
+      this.properties = Collections.emptyMap();
+    }
+  }
+
+  @Override
+  public Map<String, String> getProperties() {
+    return properties;
+  }
+}
+

cdap-data-fabric/src/main/java/io/cdap/cdap/spi/metadata/DelegatingMetadataStorage.java

@@ -0,0 +1,96 @@
+/*
+ * Copyright © 2025 Cask Data, Inc.
+ *
+ * Licensed 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.
+ */
+
+package io.cdap.cdap.spi.metadata;
+
+import com.google.inject.Inject;
+import io.cdap.cdap.common.conf.CConfiguration;
+import io.cdap.cdap.common.conf.Constants;
+import java.io.IOException;
+import java.util.List;
+
+/**
+ * Delegates {@link MetadataStorage} based on configured extension.
+ */
+public class DelegatingMetadataStorage implements MetadataStorage {
+  private final CConfiguration cConf;
+  private final MetadataStorage delegate;
+  private static String prefix;
+  private static final String SPANNER_METADATA_STORAGE = "gcp-spanner";
+
+  @Inject
+  DelegatingMetadataStorage(CConfiguration cConf, MetadataStorageExtensionLoader extensionLoader) throws Exception {
+    this.cConf = cConf;
+    this.delegate = extensionLoader.get(getName());
+
+    if (this.delegate == null) {
+      throw new IllegalArgumentException("Unsupported MetadataProvider type: " + getName());
+    }
+    // TODO(CDAP-21174): Generalize the context for metadata storage
+    if (getName().equals(SPANNER_METADATA_STORAGE)) {
+      this.prefix = String.format("%s%s.", Constants.Dataset.STORAGE_EXTENSION_PROPERTY_PREFIX,
+                                  SPANNER_METADATA_STORAGE);
+    } else {
+      this.prefix = null;
+    }
+
+    this.delegate.initialize(new DefaultMetadataStorageContext(cConf, prefix));
+  }
+
+  @Override
+  public void createIndex() throws IOException {
+    delegate.createIndex();
+  }
+
+  @Override
+  public void close() {
+    if (delegate != null) {
+      delegate.close();
+    }
+  }
+
+  @Override
+  public String getName() {
+    return cConf.get(Constants.Metadata.STORAGE_PROVIDER_IMPLEMENTATION);
+  }
+
+  @Override
+  public void dropIndex() throws IOException {
+    delegate.dropIndex();
+  }
+
+  @Override
+  public MetadataChange apply(MetadataMutation mutation, MutationOptions options) throws IOException {
+    return delegate.apply(mutation, options);
+  }
+
+  @Override
+  public List<MetadataChange> batch(List<? extends MetadataMutation> mutations, MutationOptions options)
+    throws IOException {
+    return delegate.batch(mutations, options);
+  }
+
+  @Override
+  public Metadata read(Read read) throws IOException {
+    return delegate.read(read);
+  }
+
+  @Override
+  public SearchResponse search(SearchRequest request) throws IOException {
+    return delegate.search(request);
+  }
+}
+

cdap-data-fabric/src/main/java/io/cdap/cdap/spi/metadata/MetadataStorageExtensionLoader.java

@@ -0,0 +1,70 @@
+/*
+ * Copyright © 2025 Cask Data, Inc.
+ *
+ * Licensed 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.
+ */
+
+package io.cdap.cdap.spi.metadata;
+
+import com.google.inject.Inject;
+import io.cdap.cdap.common.conf.CConfiguration;
+import io.cdap.cdap.common.conf.Constants;
+import io.cdap.cdap.common.lang.ClassPathResources;
+import io.cdap.cdap.common.lang.FilterClassLoader;
+import io.cdap.cdap.extension.AbstractExtensionLoader;
+import java.io.IOException;
+import java.util.Collections;
+import java.util.Set;
+
+/**
+ * Extension loader for {@link MetadataStorage} implementations.
+ */
+public class MetadataStorageExtensionLoader extends AbstractExtensionLoader<String, MetadataStorage> {
+
+  private static final Set<String> ALLOWED_RESOURCES = createAllowedResources();
+  private static final Set<String> ALLOWED_PACKAGES = createPackageSets(ALLOWED_RESOURCES);
+
+  @Inject
+  public MetadataStorageExtensionLoader(CConfiguration cConf) {
+    super(cConf.get(Constants.Metadata.METADATA_STORAGE_EXT_DIR));
+  }
+
+  private static Set<String> createAllowedResources() {
+    try {
+      return ClassPathResources.getResourcesWithDependencies(MetadataStorage.class.getClassLoader(),
+                                                             MetadataStorage.class);
+    } catch (IOException e) {
+      throw new RuntimeException("Failed to trace dependencies for MetadataStorage extension.", e);
+    }
+  }
+
+  @Override
+  protected Set<String> getSupportedTypesForProvider(MetadataStorage metadataStorage) {
+    return Collections.singleton(metadataStorage.getName());
+  }
+
+  @Override
+  protected FilterClassLoader.Filter getExtensionParentClassLoaderFilter() {
+    return new FilterClassLoader.Filter() {
+      @Override
+      public boolean acceptResource(String resource) {
+        return ALLOWED_RESOURCES.contains(resource);
+      }
+
+      @Override
+      public boolean acceptPackage(String packageName) {
+        return ALLOWED_PACKAGES.contains(packageName);
+      }
+    };
+  }
+}

cdap-data-fabric/src/main/java/io/cdap/cdap/spi/metadata/dataset/DatasetMetadataStorage.java

@@ -74,6 +74,11 @@ public class DatasetMetadataStorage extends SearchHelper implements MetadataStor
     super(txClient, tableDefinition);
   }
 
+  @Override
+  public String getName() {
+    return getClass().getSimpleName();
+  }
+
   @Override
   public void createIndex() throws IOException {
     createDatasets();

cdap-data-fabric/src/main/java/io/cdap/cdap/spi/metadata/noop/NoopMetadataStorage.java

@@ -34,6 +34,11 @@
  */
 public class NoopMetadataStorage implements MetadataStorage {
 
+  @Override
+  public String getName() {
+    return getClass().getSimpleName();
+  }
+
   @Override
   public void createIndex() throws IOException {
     // no-op

cdap-elastic/src/main/java/io/cdap/cdap/metadata/elastic/ElasticsearchMetadataStorage.java

@@ -217,6 +217,11 @@ public ElasticsearchMetadataStorage(CConfiguration cConf, SConfiguration sConf)
         RetryStrategies.fixDelay(retrySleepMs, TimeUnit.MILLISECONDS));
   }
 
+  @Override
+  public String getName() {
+    return getClass().getSimpleName();
+  }
+
   @Override
   public void close() {
     Closeables.closeQuietly(client);

cdap-master/pom.xml

@@ -113,6 +113,11 @@
       <artifactId>cdap-messaging-ext-spanner</artifactId>
       <version>${project.version}</version>
     </dependency>
+    <dependency>
+      <groupId>io.cdap.cdap</groupId>
+      <artifactId>cdap-metadata-ext-spanner</artifactId>
+      <version>${project.version}</version>
+    </dependency>
     <dependency>
       <groupId>org.apache.tephra</groupId>
       <artifactId>tephra-api</artifactId>
@@ -251,6 +256,7 @@
         <stage.authenticators.ext.dir>${stage.opt.dir}/ext/authenticators</stage.authenticators.ext.dir>
         <stage.credential.provider.extensions.dir>${stage.opt.dir}/ext/credentialproviders</stage.credential.provider.extensions.dir>
         <stage.messaging.ext.dir>${stage.opt.dir}/ext/messagingproviders</stage.messaging.ext.dir>
+        <stage.metadata.ext.dir>${stage.opt.dir}/ext/metadatastorage</stage.metadata.ext.dir>
         <stage.metricswriters.ext.dir>${stage.opt.dir}/ext/metricswriters/google_cloud_monitoring_writer
         </stage.metricswriters.ext.dir>
         <stage.eventwriters.ext.dir>${stage.opt.dir}/ext/eventwriters/google_cloud_pubsub_writer
@@ -733,6 +739,28 @@
                 </configuration>
               </execution>
 
+              <!-- Copy metadata extensions -->
+              <execution>
+                <id>copy-metadata-ext-spanner</id>
+                <phase>process-resources</phase>
+                <goals>
+                  <goal>copy-resources</goal>
+                </goals>
+                <configuration combine.self="override">
+                  <outputDirectory>${stage.metadata.ext.dir}/spanner</outputDirectory>
+                  <resources>
+                    <resource>
+                      <directory>
+                        ${project.parent.basedir}/cdap-metadata-ext-spanner/target/libexec/
+                      </directory>
+                      <includes>
+                        <include>*.jar</include>
+                      </includes>
+                    </resource>
+                  </resources>
+                </configuration>
+              </execution>
+
               <!-- Copy remote authenticator extensions -->
               <execution>
                 <id>copy-authenticator-ext-gcp</id>