Skip to content

Commit 257bddf

Browse files
author
Komal Yadav
committed
Add SpannerMetadataModule with ExtensionLoader
Format Changes changes Changes name-changes formatting-changes elastic-changes changes Added copyright section Updated Updated Added Javadocs Added Javadocs
1 parent 60e58d4 commit 257bddf

25 files changed

Lines changed: 688 additions & 27 deletions

File tree

‎cdap-app-fabric/src/main/java/io/cdap/cdap/metadata/MetadataHttpHandler.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -31,8 +31,8 @@
3131
import io.cdap.cdap.common.security.AuditDetail;
3232
import io.cdap.cdap.common.security.AuditPolicy;
3333
import io.cdap.cdap.data2.metadata.MetadataCompatibility;
34-
import io.cdap.cdap.metadata.elastic.ScopedNameOfKindTypeAdapter;
35-
import io.cdap.cdap.metadata.elastic.ScopedNameTypeAdapter;
34+
import io.cdap.cdap.spi.metadata.ScopedNameTypeAdapter;
35+
import io.cdap.cdap.spi.metadata.ScopedNameOfKindTypeAdapter;
3636
import io.cdap.cdap.proto.EntityScope;
3737
import io.cdap.cdap.proto.ProgramType;
3838
import io.cdap.cdap.proto.codec.NamespacedEntityIdCodec;

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

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2114,9 +2114,13 @@ public static final class Metadata {
21142114
public static final String STORAGE_PROVIDER_IMPLEMENTATION = "metadata.storage.implementation";
21152115
public static final String STORAGE_PROVIDER_NOSQL = "nosql";
21162116
public static final String STORAGE_PROVIDER_ELASTICSEARCH = "elastic";
2117+
public static final String STORAGE_PROVIDER_SPANNER = "spanner";
21172118

21182119
public static final String METADATA_WRITER_SUBSCRIBER = "metadata.writer";
21192120
public static final String METADATA_CONSUMER_WRITER_SUBSCRIBER = "metadata.consumer.writer";
2121+
2122+
// Metadata configs
2123+
public static final String METADATA_STORAGE_EXT_DIR = "metadata.storage.extensions.dir";
21202124
}
21212125

21222126
/**

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

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2845,6 +2845,11 @@
28452845
<value>/opt/cdap/master/ext/log-publisher</value>
28462846
</property>
28472847

2848+
<property>
2849+
<name>metadata.storage.extensions.dir</name>
2850+
<value>/opt/cdap/master/ext/metadata-storage</value>
2851+
</property>
2852+
28482853
<!-- Metrics Configuration -->
28492854

28502855
<property>

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

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -45,6 +45,7 @@
4545
import io.cdap.cdap.data2.registry.UsageWriter;
4646
import io.cdap.cdap.metadata.elastic.ElasticsearchMetadataStorage;
4747
import io.cdap.cdap.security.impersonation.OwnerStore;
48+
import io.cdap.cdap.spi.metadata.DelegatingMetadataStorage;
4849
import io.cdap.cdap.spi.metadata.MetadataStorage;
4950
import io.cdap.cdap.spi.metadata.dataset.DatasetMetadataStorage;
5051
import io.cdap.cdap.spi.metadata.noop.NoopMetadataStorage;
@@ -181,8 +182,12 @@ public MetadataStorage get() {
181182
if (Constants.Metadata.STORAGE_PROVIDER_ELASTICSEARCH.equalsIgnoreCase(config)) {
182183
return injector.getInstance(ElasticsearchMetadataStorage.class);
183184
}
185+
if (Constants.Metadata.STORAGE_PROVIDER_SPANNER.equalsIgnoreCase(config)) {
186+
return injector.getInstance(DelegatingMetadataStorage.class);
187+
}
184188
throw new IllegalArgumentException("Unsupported MetadataStorage '" + config + "'. Only '"
185-
+ Constants.Metadata.STORAGE_PROVIDER_NOSQL + "' and '"
189+
+ Constants.Metadata.STORAGE_PROVIDER_NOSQL + "','"
190+
+ Constants.Metadata.STORAGE_PROVIDER_SPANNER + "' and '"
186191
+ Constants.Metadata.STORAGE_PROVIDER_ELASTICSEARCH + "' are allowed.");
187192
}
188193
}

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

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -83,6 +83,11 @@ public void setAuditPublisher(AuditPublisher auditPublisher) {
8383
this.auditPublisher = auditPublisher;
8484
}
8585

86+
@Override
87+
public String getName() {
88+
return getClass().getSimpleName();
89+
}
90+
8691
@Override
8792
public void createIndex() throws IOException {
8893
try {

‎cdap-data-fabric/src/main/java/io/cdap/cdap/data2/metadata/writer/DefaultMetadataServiceClient.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -22,8 +22,8 @@
2222
import io.cdap.cdap.common.http.DefaultHttpRequestConfig;
2323
import io.cdap.cdap.common.internal.remote.RemoteClient;
2424
import io.cdap.cdap.common.internal.remote.RemoteClientFactory;
25-
import io.cdap.cdap.metadata.elastic.ScopedNameOfKindTypeAdapter;
26-
import io.cdap.cdap.metadata.elastic.ScopedNameTypeAdapter;
25+
import io.cdap.cdap.spi.metadata.ScopedNameTypeAdapter;
26+
import io.cdap.cdap.spi.metadata.ScopedNameOfKindTypeAdapter;
2727
import io.cdap.cdap.proto.codec.NamespacedEntityIdCodec;
2828
import io.cdap.cdap.proto.id.NamespacedEntityId;
2929
import io.cdap.cdap.spi.metadata.Metadata;

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

Lines changed: 120 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,120 @@
1+
/*
2+
* Copyright © 2025 Cask Data, Inc.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License"); you may not
5+
* use this file except in compliance with the License. You may obtain a copy of
6+
* the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12+
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13+
* License for the specific language governing permissions and limitations under
14+
* the License.
15+
*/
16+
17+
package io.cdap.cdap.spi.metadata;
18+
19+
import com.google.inject.Inject;
20+
import io.cdap.cdap.common.conf.CConfiguration;
21+
import io.cdap.cdap.common.conf.Constants;
22+
import org.slf4j.Logger;
23+
import org.slf4j.LoggerFactory;
24+
import java.io.IOException;
25+
import java.util.List;
26+
27+
/**
28+
* Delegates {@link io.cdap.cdap.spi.metadata.MetadataStorage} based on configured extension
29+
*/
30+
public class DelegatingMetadataStorage implements MetadataStorage {
31+
private static final Logger LOG = LoggerFactory.getLogger(DelegatingMetadataStorage.class);
32+
33+
private final CConfiguration cConf;
34+
private final MetadataStorageExtensionLoader extensionLoader;
35+
private volatile MetadataStorage delegate;
36+
37+
@Inject
38+
DelegatingMetadataStorage(CConfiguration cConf, MetadataStorageExtensionLoader extensionLoader) {
39+
this.cConf = cConf;
40+
this.extensionLoader = extensionLoader;
41+
}
42+
43+
@Override
44+
public void createIndex() throws IOException {
45+
getDelegate().createIndex();
46+
}
47+
48+
@Override
49+
public void close() {
50+
if (delegate != null) {
51+
delegate.close();
52+
}
53+
}
54+
55+
public String getName() {
56+
return cConf.get(Constants.Metadata.STORAGE_PROVIDER_IMPLEMENTATION);
57+
}
58+
59+
@Override
60+
public void dropIndex() throws IOException {
61+
getDelegate().dropIndex();
62+
}
63+
64+
@Override
65+
public MetadataChange apply(MetadataMutation mutation, MutationOptions options)
66+
throws IOException {
67+
return getDelegate().apply(mutation, options);
68+
}
69+
70+
@Override
71+
public List<MetadataChange> batch(List<? extends MetadataMutation> mutations,
72+
MutationOptions options) throws IOException {
73+
return getDelegate().batch(mutations, options);
74+
}
75+
76+
@Override
77+
public Metadata read(Read read) throws IOException {
78+
return getDelegate().read(read);
79+
}
80+
81+
@Override
82+
public SearchResponse search(SearchRequest request)
83+
throws IOException {
84+
return getDelegate().search(request);
85+
}
86+
87+
/**
88+
* Returns the {@link MetadataStorage} to use based on configuration.
89+
*/
90+
private MetadataStorage getDelegate() {
91+
MetadataStorage metadataStorage = this.delegate;
92+
if (metadataStorage != null) {
93+
return metadataStorage;
94+
}
95+
synchronized (this) {
96+
metadataStorage = this.delegate;
97+
if (metadataStorage != null) {
98+
return metadataStorage;
99+
}
100+
metadataStorage = extensionLoader.get(getName());
101+
102+
if (metadataStorage == null) {
103+
throw new IllegalArgumentException(
104+
"Unsupported metadata storage implementation " + getName());
105+
}
106+
LOG.info("Metadata Storage {} is loaded", metadataStorage.getName());
107+
try {
108+
metadataStorage.initialize(new DelegatingMetadataStorageContext(this.cConf,
109+
metadataStorage.getName()));
110+
} catch (Exception e) {
111+
throw new RuntimeException(e);
112+
}
113+
LOG.info("Metadata storage {} is initialized.", metadataStorage.getName());
114+
115+
this.delegate = metadataStorage;
116+
return metadataStorage;
117+
}
118+
}
119+
}
120+

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

Lines changed: 43 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,43 @@
1+
/*
2+
* Copyright © 2025 Cask Data, Inc.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License"); you may not
5+
* use this file except in compliance with the License. You may obtain a copy of
6+
* the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12+
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13+
* License for the specific language governing permissions and limitations under
14+
* the License.
15+
*/
16+
17+
package io.cdap.cdap.spi.metadata;
18+
19+
import io.cdap.cdap.common.conf.CConfiguration;
20+
import io.cdap.cdap.common.conf.Constants;
21+
import java.util.Collections;
22+
import java.util.Map;
23+
24+
/**
25+
* Default implementation of the {@link MetadataStorageContext}.
26+
*/
27+
public class DelegatingMetadataStorageContext implements MetadataStorageContext {
28+
29+
private static final String storageImpl = "gcp-spanner";
30+
private final Map<String, String> properties;
31+
32+
protected DelegatingMetadataStorageContext(CConfiguration cConf, String storageName) {
33+
String propertiesPrefix =
34+
Constants.Dataset.STORAGE_EXTENSION_PROPERTY_PREFIX + storageImpl + ".";
35+
this.properties = Collections.unmodifiableMap(cConf.getPropsWithPrefix(propertiesPrefix));
36+
}
37+
38+
@Override
39+
public Map<String, String> getProperties() {
40+
return properties;
41+
}
42+
}
43+

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

Lines changed: 75 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,75 @@
1+
/*
2+
* Copyright © 2025 Cask Data, Inc.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License"); you may not
5+
* use this file except in compliance with the License. You may obtain a copy of
6+
* the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12+
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13+
* License for the specific language governing permissions and limitations under
14+
* the License.
15+
*/
16+
17+
package io.cdap.cdap.spi.metadata;
18+
19+
import com.google.inject.Inject;
20+
import io.cdap.cdap.common.conf.CConfiguration;
21+
import io.cdap.cdap.common.conf.Constants;
22+
import io.cdap.cdap.common.lang.ClassPathResources;
23+
import io.cdap.cdap.common.lang.FilterClassLoader;
24+
import io.cdap.cdap.extension.AbstractExtensionLoader;
25+
import org.slf4j.Logger;
26+
import org.slf4j.LoggerFactory;
27+
import java.io.IOException;
28+
import java.util.Collections;
29+
import java.util.Set;
30+
31+
/**
32+
* Extension loader for {@link MetadataStorage} implementations.
33+
*/
34+
public class MetadataStorageExtensionLoader extends AbstractExtensionLoader<String, MetadataStorage> {
35+
36+
private static final Logger LOG = LoggerFactory.getLogger(MetadataStorageExtensionLoader.class);
37+
private static final Set<String> ALLOWED_RESOURCES = createAllowedResources();
38+
private static final Set<String> ALLOWED_PACKAGES = createPackageSets(ALLOWED_RESOURCES);
39+
40+
@Inject
41+
public MetadataStorageExtensionLoader(CConfiguration cConf) {
42+
super(cConf.get(Constants.Metadata.METADATA_STORAGE_EXT_DIR));
43+
LOG.debug("Metadata Storage extensions directory: {}",
44+
cConf.get(Constants.Metadata.METADATA_STORAGE_EXT_DIR));
45+
}
46+
47+
private static Set<String> createAllowedResources() {
48+
try {
49+
return ClassPathResources.getResourcesWithDependencies(MetadataStorage.class.getClassLoader(),
50+
MetadataStorage.class);
51+
} catch (IOException e) {
52+
throw new RuntimeException("Failed to trace dependencies for MetadataStorage extension.", e);
53+
}
54+
}
55+
56+
@Override
57+
protected Set<String> getSupportedTypesForProvider(MetadataStorage metadataStorage) {
58+
return Collections.singleton(metadataStorage.getName());
59+
}
60+
61+
@Override
62+
protected FilterClassLoader.Filter getExtensionParentClassLoaderFilter() {
63+
return new FilterClassLoader.Filter() {
64+
@Override
65+
public boolean acceptResource(String resource) {
66+
return ALLOWED_RESOURCES.contains(resource);
67+
}
68+
69+
@Override
70+
public boolean acceptPackage(String packageName) {
71+
return ALLOWED_PACKAGES.contains(packageName);
72+
}
73+
};
74+
}
75+
}

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

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -74,6 +74,11 @@ public class DatasetMetadataStorage extends SearchHelper implements MetadataStor
7474
super(txClient, tableDefinition);
7575
}
7676

77+
@Override
78+
public String getName(){
79+
return getClass().getSimpleName();
80+
}
81+
7782
@Override
7883
public void createIndex() throws IOException {
7984
createDatasets();

0 commit comments

Comments
 (0)