Merge pull request #5380 from atlanhq/mlh-1240-custom-metadata-consistency

Mlh 1240 custom metadata consistency

Author: sriram-atlan

.github/workflows/maven.yml

@@ -31,6 +31,7 @@ on:
       - tagscanarymerge
       - fixlabels
       - interceptapis
+      - mlh-1240-custom-metadata-consistency
 
 jobs:
   build:

common/src/main/java/org/apache/atlas/repository/Constants.java

@@ -538,6 +538,13 @@ public enum SupportedFileExtensions { XLSX, XLS, CSV }
         add(STAKEHOLDER_TITLE_ENTITY_TYPE);
     }};
 
+    public static final String TYPEDEF_ENUM_CACHE_LATEST_VERSION = "typdef.enum.cache.version";
+    public static final String TYPEDEF_BUSINESS_METADATA_CACHE_LATEST_VERSION = "typdef.bm.cache.version";
+    public static final String TYPEDEF_CLASSIFICATION_METADATA_CACHE_LATEST_VERSION = "typdef.cls.cache.version";
+    public static final String TYPEDEF_STRUCT_CACHE_LATEST_VERSION = "typdef.struct.cache.version";
+    public static final String TYPEDEF_ENTITY_CACHE_LATEST_VERSION = "typdef.entity.cache.version";
+    public static final String TYPEDEF_RELATIONSHIP_CACHE_LATEST_VERSION = "typdef.relationship.cache.version";
+
     private Constants() {
     }
 

common/src/main/java/org/apache/atlas/service/redis/AbstractRedisService.java

@@ -5,6 +5,7 @@
 import org.apache.atlas.service.metrics.MetricUtils;
 import org.apache.commons.configuration.Configuration;
 import org.apache.commons.lang.ArrayUtils;
+import org.apache.commons.lang.StringUtils;
 import org.redisson.api.RLock;
 import org.redisson.api.RedissonClient;
 import org.redisson.config.Config;
@@ -191,6 +192,22 @@ public String getValue(String key) {
         }
     }
 
+    @Override
+    public String getValue(String key, String defaultValue) {
+        try {
+            String value = getValue(key);
+            if (StringUtils.isEmpty(value)) {
+                return defaultValue;
+            } else {
+                return value;
+            }
+        } catch (Exception e) {
+            MetricUtils.recordRedisConnectionFailure();
+            getLogger().error("Redis getValue operation failed for key: {}", key, e);
+            throw e;
+        }
+    }
+
     @Override
     public String putValue(String key, String value) {
         try {

common/src/main/java/org/apache/atlas/service/redis/RedisService.java

@@ -16,6 +16,8 @@ public interface RedisService {
 
   String getValue(String key);
 
+  String getValue(String key, String defaultValue);
+
   String putValue(String key, String value);
 
   String putValue(String key, String value, int timeout);

common/src/main/java/org/apache/atlas/service/redis/RedisServiceLocalImpl.java

@@ -27,6 +27,11 @@ public String getValue(String key) {
         return super.getValue(key);
     }
 
+    @Override
+    public String getValue(String key, String defaultValue) {
+        return null;
+    }
+
     @Override
     public String putValue(String key, String value, int timeout) {
         return super.putValue(key, value, timeout);

repository/src/main/java/org/apache/atlas/repository/graph/TypeCacheRefresher.java

@@ -50,6 +50,7 @@
 import org.apache.atlas.repository.patches.SuperTypesUpdatePatch;
 import org.apache.atlas.repository.patches.AtlasPatchManager;
 import org.apache.atlas.repository.patches.AtlasPatchRegistry;
+import org.apache.atlas.service.redis.RedisService;
 import org.apache.atlas.store.AtlasTypeDefStore;
 import org.apache.atlas.type.AtlasEntityType;
 import org.apache.atlas.type.AtlasStructType.AtlasAttribute;
@@ -101,15 +102,23 @@ public class AtlasTypeDefStoreInitializer implements ActiveStateChangeHandler {
     private final Configuration     conf;
     private final AtlasGraph        graph;
     private final AtlasPatchManager patchManager;
+    private final RedisService redisService;
+    private static long CURRENT_ENUM_TYPEDEF_INTERNAL_VERSION;
+    private static long CURRENT_BUSINESS_METADATA_TYPEDEF_INTERNAL_VERSION;
+    private static long CURRENT_CLASSIFICATION_TYPEDEF_INTERNAL_VERSION;
+    private static long CURRENT_STRUCT_TYPEDEF_INTERNAL_VERSION;
+    private static long CURRENT_ENTITY_TYPEDEF_INTERNAL_VERSION;
+    private static long CURRENT_RELATIONSHIP_TYPEDEF_INTERNAL_VERSION;
 
     @Inject
     public AtlasTypeDefStoreInitializer(AtlasTypeDefStore typeDefStore, AtlasTypeRegistry typeRegistry,
-                                        AtlasGraph graph, Configuration conf, AtlasPatchManager patchManager) throws AtlasBaseException {
+                                        AtlasGraph graph, Configuration conf, AtlasPatchManager patchManager, RedisService redisService) throws AtlasBaseException {
         this.typeDefStore  = typeDefStore;
         this.typeRegistry  = typeRegistry;
         this.conf          = conf;
         this.graph         = graph;
         this.patchManager  = patchManager;
+        this.redisService = redisService;
     }
 
     @PostConstruct
@@ -118,6 +127,12 @@ public void init() {
 
         if (!HAConfiguration.isHAEnabled(conf)) {
             startInternal();
+            CURRENT_ENUM_TYPEDEF_INTERNAL_VERSION = Long.parseLong(redisService.getValue(Constants.TYPEDEF_ENUM_CACHE_LATEST_VERSION, "1"));
+            CURRENT_BUSINESS_METADATA_TYPEDEF_INTERNAL_VERSION = Long.parseLong(redisService.getValue(Constants.TYPEDEF_BUSINESS_METADATA_CACHE_LATEST_VERSION, "1"));
+            CURRENT_CLASSIFICATION_TYPEDEF_INTERNAL_VERSION = Long.parseLong(redisService.getValue(Constants.TYPEDEF_CLASSIFICATION_METADATA_CACHE_LATEST_VERSION, "1"));
+            CURRENT_STRUCT_TYPEDEF_INTERNAL_VERSION = Long.parseLong(redisService.getValue(Constants.TYPEDEF_STRUCT_CACHE_LATEST_VERSION, "1"));
+            CURRENT_ENTITY_TYPEDEF_INTERNAL_VERSION = Long.parseLong(redisService.getValue(Constants.TYPEDEF_ENTITY_CACHE_LATEST_VERSION, "1"));
+            CURRENT_RELATIONSHIP_TYPEDEF_INTERNAL_VERSION = Long.parseLong(redisService.getValue(Constants.TYPEDEF_RELATIONSHIP_CACHE_LATEST_VERSION, "1"));
         } else {
             LOG.info("AtlasTypeDefStoreInitializer.init(): deferring type loading until instance activation");
         }
@@ -407,6 +422,54 @@ public int getHandlerOrder() {
         return HandlerOrder.TYPEDEF_STORE_INITIALIZER.getOrder();
     }
 
+    public static long getCurrentEnumTypedefInternalVersion() {
+        return CURRENT_ENUM_TYPEDEF_INTERNAL_VERSION;
+    }
+
+    public static void setCurrentEnumTypedefInternalVersion(long version) {
+        CURRENT_ENUM_TYPEDEF_INTERNAL_VERSION = version;
+    }
+
+    public static long getCurrentBMTypedefInternalVersion() {
+        return CURRENT_BUSINESS_METADATA_TYPEDEF_INTERNAL_VERSION;
+    }
+
+    public static void setCurrentBMTypedefInternalVersion(long version) {
+        CURRENT_BUSINESS_METADATA_TYPEDEF_INTERNAL_VERSION = version;
+    }
+
+    public static long getCurrentClassificationTypedefInternalVersion() {
+        return CURRENT_CLASSIFICATION_TYPEDEF_INTERNAL_VERSION;
+    }
+
+    public static void setCurrentClassificationTypedefInternalVersion(long version) {
+        CURRENT_CLASSIFICATION_TYPEDEF_INTERNAL_VERSION = version;
+    }
+
+    public static long getCurrentStructTypedefInternalVersion() {
+        return CURRENT_STRUCT_TYPEDEF_INTERNAL_VERSION;
+    }
+
+    public static void setCurrentStructTypedefInternalVersion(long version) {
+        CURRENT_STRUCT_TYPEDEF_INTERNAL_VERSION = version;
+    }
+
+    public static long getCurrentEntityTypedefInternalVersion() {
+        return CURRENT_ENTITY_TYPEDEF_INTERNAL_VERSION;
+    }
+
+    public static void setCurrentEntityTypedefInternalVersion(long version) {
+        CURRENT_ENTITY_TYPEDEF_INTERNAL_VERSION = version;
+    }
+
+    public static long getCurrentRelationshipTypedefInternalVersion() {
+        return CURRENT_RELATIONSHIP_TYPEDEF_INTERNAL_VERSION;
+    }
+
+    public static void setCurrentRelationshipTypedefInternalVersion(long version) {
+        CURRENT_RELATIONSHIP_TYPEDEF_INTERNAL_VERSION = version;
+    }
+
     private static boolean updateTypeAttributes(AtlasStructDef oldStructDef, AtlasStructDef newStructDef, boolean checkTypeVersion) {
         boolean ret = isTypeUpdateApplicable(oldStructDef, newStructDef, checkTypeVersion);