add leaseTime in redis locks

Author: aarshi0301

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

@@ -27,15 +27,18 @@ public abstract class AbstractRedisService implements RedisService {
     private static final String ATLAS_REDIS_MASTER_NAME = "atlas.redis.master_name";
     private static final String ATLAS_REDIS_LOCK_WAIT_TIME_MS = "atlas.redis.lock.wait_time.ms";
     private static final String ATLAS_REDIS_LOCK_WATCHDOG_TIMEOUT_MS = "atlas.redis.lock.watchdog_timeout.ms";
+    private static final String ATLAS_REDIS_LEASE_TIME_MS = "atlas.redis.lease_time.ms";
     private static final int DEFAULT_REDIS_WAIT_TIME_MS = 15_000;
     private static final int DEFAULT_REDIS_LOCK_WATCHDOG_TIMEOUT_MS = 600_000;
+    private static final int DEFAULT_REDIS_LEASE_TIME_MS = 60_000;
     private static final String ATLAS_METASTORE_SERVICE = "atlas-metastore-service";
 
     RedissonClient redisClient;
     RedissonClient redisCacheClient;
     Map<String, RLock> keyLockMap;
     Configuration atlasConfig;
     long waitTimeInMS;
+    long leaseTimeInMS;
     long watchdogTimeoutInMS;
 
     @Override
@@ -44,7 +47,7 @@ public boolean acquireDistributedLock(String key) throws Exception {
         boolean isLockAcquired;
         try {
             RLock lock = redisClient.getFairLock(key);
-            isLockAcquired = lock.tryLock(waitTimeInMS, TimeUnit.MILLISECONDS);
+            isLockAcquired = lock.tryLock(waitTimeInMS, leaseTimeInMS, TimeUnit.MILLISECONDS);
             if (isLockAcquired) {
                 keyLockMap.put(key, lock);
             } else {
@@ -57,16 +60,18 @@ public boolean acquireDistributedLock(String key) throws Exception {
         return isLockAcquired;
     }
 
+
     @Override
     public void releaseDistributedLock(String key) {
-        if (!keyLockMap.containsKey(key)) {
-            return;
-        }
+            if (!keyLockMap.containsKey(key)) {
+                return;
+            }
         try {
             RLock lock = keyLockMap.get(key);
             if (lock.isHeldByCurrentThread()) {
                 lock.unlock();
             }
+
         } catch (Exception e) {
             getLogger().error("Failed to release distributed lock for {}", key, e);
         }
@@ -106,6 +111,7 @@ private Config initAtlasConfig() throws AtlasException {
         keyLockMap = new ConcurrentHashMap<>();
         atlasConfig = ApplicationProperties.get();
         waitTimeInMS = atlasConfig.getLong(ATLAS_REDIS_LOCK_WAIT_TIME_MS, DEFAULT_REDIS_WAIT_TIME_MS);
+        leaseTimeInMS = atlasConfig.getLong(ATLAS_REDIS_LEASE_TIME_MS, DEFAULT_REDIS_LEASE_TIME_MS);
         watchdogTimeoutInMS = atlasConfig.getLong(ATLAS_REDIS_LOCK_WATCHDOG_TIMEOUT_MS, DEFAULT_REDIS_LOCK_WATCHDOG_TIMEOUT_MS);
         Config redisConfig = new Config();
         redisConfig.setLockWatchdogTimeout(watchdogTimeoutInMS);

webapp/src/main/java/org/apache/atlas/web/rest/TypesREST.java

@@ -513,7 +513,6 @@ public AtlasTypesDef updateAtlasTypeDefs(final AtlasTypesDef typesDef, @QueryPar
             throw new AtlasBaseException("Error while updating a type definition");
         } finally {
             RequestContext.clear();
-            redisService.releaseDistributedLock(ATLAS_TYPEDEF_LOCK);
             AtlasPerfTracer.log(perf);
         }
     }