RANGER-1980: changed Atlas version from 1.0.0-SNAPSHOT to Atlas-0.8.2 (Revert 
"RANGER-1897: TagSync should replace use of V1 Atlas APIs with V2 APIs for 
efficient tag-download from Atlas")

This reverts commit e8afb9faad81e7042877aa528635848a0043cb0c.


Project: http://git-wip-us.apache.org/repos/asf/ranger/repo
Commit: http://git-wip-us.apache.org/repos/asf/ranger/commit/1ab501c3
Tree: http://git-wip-us.apache.org/repos/asf/ranger/tree/1ab501c3
Diff: http://git-wip-us.apache.org/repos/asf/ranger/diff/1ab501c3

Branch: refs/heads/master
Commit: 1ab501c3bce92d4f6c9d4a5b9168dd34c0310289
Parents: ea30b13
Author: Madhan Neethiraj <mad...@apache.org>
Authored: Mon Feb 12 07:29:52 2018 -0800
Committer: Madhan Neethiraj <mad...@apache.org>
Committed: Mon Feb 12 13:59:54 2018 -0800

----------------------------------------------------------------------
 .../atlas/authorizer/RangerAtlasResource.java   |   4 -
 pom.xml                                         |   4 +-
 src/main/assembly/tagsync.xml                   |  30 +-
 tagsync/pom.xml                                 |  75 +++--
 .../ranger/tagsync/process/TagSynchronizer.java |   2 +-
 .../tagsync/sink/tagadmin/TagAdminRESTSink.java |  11 +-
 .../source/atlas/AtlasEntityWithTraits.java     |  98 ++++++
 .../source/atlas/AtlasHbaseResourceMapper.java  |  12 +-
 .../source/atlas/AtlasHdfsResourceMapper.java   |  13 +-
 .../source/atlas/AtlasHiveResourceMapper.java   |  11 +-
 .../source/atlas/AtlasKafkaResourceMapper.java  |  17 +-
 .../source/atlas/AtlasNotificationMapper.java   | 142 ++++----
 .../source/atlas/AtlasResourceMapper.java       |  26 +-
 .../source/atlas/AtlasResourceMapperUtil.java   |  66 ++--
 .../source/atlas/AtlasStormResourceMapper.java  |  10 +-
 .../tagsync/source/atlas/AtlasTagSource.java    |  97 +++---
 .../source/atlasrest/AtlasRESTTagSource.java    | 238 ++------------
 .../tagsync/source/atlasrest/AtlasRESTUtil.java | 325 +++++++++++++++++++
 .../source/atlasrest/RangerAtlasEntity.java     |  60 ----
 .../atlasrest/RangerAtlasEntityWithTags.java    | 118 -------
 .../process/TestHbaseResourceMapper.java        |  56 ++--
 .../tagsync/process/TestHdfsResourceMapper.java |  24 +-
 .../tagsync/process/TestHiveResourceMapper.java |  28 +-
 .../process/TestKafkaResourceMapper.java        |  16 +-
 24 files changed, 805 insertions(+), 678 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/ranger/blob/1ab501c3/plugin-atlas/src/main/java/org/apache/ranger/authorization/atlas/authorizer/RangerAtlasResource.java
----------------------------------------------------------------------
diff --git 
a/plugin-atlas/src/main/java/org/apache/ranger/authorization/atlas/authorizer/RangerAtlasResource.java
 
b/plugin-atlas/src/main/java/org/apache/ranger/authorization/atlas/authorizer/RangerAtlasResource.java
index 4367c5e..f056f3e 100644
--- 
a/plugin-atlas/src/main/java/org/apache/ranger/authorization/atlas/authorizer/RangerAtlasResource.java
+++ 
b/plugin-atlas/src/main/java/org/apache/ranger/authorization/atlas/authorizer/RangerAtlasResource.java
@@ -28,10 +28,8 @@ public class RangerAtlasResource extends 
RangerAccessResourceImpl {
     public static final String KEY_TYPE = "type";
     public static final String KEY_ENTITY = "entity";
     public static final String KEY_OPERATION = "operation";
-    /*
     public static final String KEY_TAXONOMY = "taxonomy";
     public static final String KEY_TERM = "term";
-    */
 
     private static final Logger LOG = 
LoggerFactory.getLogger(RangerAtlasResource.class);
 
@@ -46,14 +44,12 @@ public class RangerAtlasResource extends 
RangerAccessResourceImpl {
             case OPERATION:
                 setValue(KEY_OPERATION, atlasResource);
                 break;
-                /*
             case TAXONOMY:
                 setValue(KEY_TAXONOMY, atlasResource);
                 break;
             case TERM:
                 setValue(KEY_TERM, atlasResource);
                 break;
-                */
             default:
                 LOG.warn("Invalid Resource : " + atlasResource);
                 break;

http://git-wip-us.apache.org/repos/asf/ranger/blob/1ab501c3/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 69e042f..4076b67 100644
--- a/pom.xml
+++ b/pom.xml
@@ -125,10 +125,9 @@
         <apacheds.version>2.0.0-M22</apacheds.version>
         <asm.all.version>3.2</asm.all.version>
         <aspectj.version>1.8.2</aspectj.version>
-        <atlas.version>1.0.0-SNAPSHOT</atlas.version>
+        <atlas.version>0.8.2</atlas.version>
         <atlas.guava.version>14.0</atlas.guava.version>
         <atlas.gson.version>2.5</atlas.gson.version>
-        <atlas.jackson.version>2.9.2</atlas.jackson.version>
         <atlas.jettison.version>1.3.7</atlas.jettison.version>
         <atlas.commons.logging.version>1.1.3</atlas.commons.logging.version>
         <bouncycastle.version>1.55</bouncycastle.version>
@@ -210,7 +209,6 @@
         <springframework.version>3.2.10.RELEASE</springframework.version>
         <sqoop.version>1.99.7</sqoop.version>
         <storm.version>1.1.1</storm.version>
-        <sun-jersey-bundle.version>1.19</sun-jersey-bundle.version>
         <tomcat.embed.version>7.0.82</tomcat.embed.version>
         <velocity.version>1.7</velocity.version>
         <zookeeper.version>3.4.6</zookeeper.version>

http://git-wip-us.apache.org/repos/asf/ranger/blob/1ab501c3/src/main/assembly/tagsync.xml
----------------------------------------------------------------------
diff --git a/src/main/assembly/tagsync.xml b/src/main/assembly/tagsync.xml
index c929395..0b17151 100644
--- a/src/main/assembly/tagsync.xml
+++ b/src/main/assembly/tagsync.xml
@@ -37,13 +37,14 @@
                                        <include>com.101tec:zkclient</include>
                                        
<include>com.google.code.gson:gson:jar:${gson.version}</include>
                                        
<include>com.google.guava:guava:jar:${google.guava.version}</include>
+                                       
<include>com.google.inject:guice:jar:${guice.version}</include>
+                                       
<include>com.google.inject.extensions:guice-multibindings:jar:${guice.version}</include>
                                        
<include>com.sun.jersey:jersey-bundle:jar:${jersey-bundle.version}</include>
-                                       
<include>com.sun.jersey.contribs:jersey-multipart:jar:${sun-jersey-bundle.version}</include>
+                                       
<include>com.thoughtworks.paranamer:paranamer:jar:${paranamer.version}</include>
+                                       
<include>com.yammer.metrics:metrics-core</include>
                                        
<include>org.apache.atlas:atlas-notification:jar:${atlas.version}</include>
-                                       
<include>org.apache.atlas:atlas-intg:jar:${atlas.version}</include>
-                                       
<include>org.apache.atlas:atlas-client-v1:jar:${atlas.version}</include>
-                                       
<include>org.apache.atlas:atlas-client-v2:jar:${atlas.version}</include>
-                                       
<include>org.apache.atlas:atlas-client-common:jar:${atlas.version}</include>
+                                       
<include>org.apache.atlas:atlas-typesystem:jar:${atlas.version}</include>
+                                       
<include>org.apache.atlas:atlas-client:jar:${atlas.version}</include>
                                        
<include>org.apache.atlas:atlas-common:jar:${atlas.version}</include>
                                        
<include>org.apache.hadoop:hadoop-auth</include>
                                        
<include>org.apache.hadoop:hadoop-common</include>
@@ -54,15 +55,20 @@
                                        
<include>org.apache.ranger:ranger-plugins-common</include>
                                        
<include>org.apache.ranger:ranger-util</include>
                                        
<include>org.apache.zookeeper:zookeeper:jar:${zookeeper.version}</include>
-                                       
<include>com.fasterxml.jackson.core:jackson-annotations:jar:${atlas.jackson.version}</include>
-                                       
<include>com.fasterxml.jackson.core:jackson-core:jar:${atlas.jackson.version}</include>
-                                       
<include>com.fasterxml.jackson.core:jackson-databind:jar:${atlas.jackson.version}</include>
-                                       
<include>com.fasterxml.jackson.jaxrs:jackson-jaxrs-base:jar:${atlas.jackson.version}</include>
-                                       
<include>com.fasterxml.jackson.jaxrs:jackson-jaxrs-json-provider:jar:${atlas.jackson.version}</include>
-                                       
<include>org.codehaus.jackson:jackson-core-asl:jar:${codehaus.jackson.version}</include>
-                                       
<include>org.codehaus.jackson:jackson-jaxrs:jar:${codehaus.jackson.version}</include>
+                                       
<include>org.codehaus.jackson:jackson-core-asl</include>
+                                       
<include>org.codehaus.jackson:jackson-jaxrs</include>
+                                       
<include>org.codehaus.jackson:jackson-mapper-asl</include>
+                                       
<include>org.codehaus.jackson:jackson-xc</include>
                                        
<include>org.codehaus.jettison:jettison:jar:${jettison.version}</include>
+                                       
<include>org.json4s:json4s-native_${scala.binary.version}:jar:${json4s.version}</include>
+                                       
<include>org.json4s:json4s-core_${scala.binary.version}:jar:${json4s.version}</include>
+                                       
<include>org.json4s:json4s-ast_${scala.binary.version}:jar:${json4s.version}</include>
                                        
<include>org.scala-lang:scala-library:jar:${scala.version}</include>
+                                       
<include>org.scala-lang:scalap:jar:${scala.version}</include>
+                                       
<include>org.scala-lang:scala-compiler:jar:${scala.version}</include>
+                                       
<include>org.scala-lang:scala-reflect:jar:${scala.version}</include>
+                                       
<include>org.scala-lang.modules:scala-xml_${scala.binary.version}:jar:${scala.xml.version}</include>
+                                       
<include>org.scala-lang.modules:scala-parser-combinators_${scala.binary.version}:jar:${scala.xml.version}</include>
                                        <include>org.slf4j:slf4j-api</include>
                                        
<include>aopalliance:aopalliance:jar:${aopalliance.version}</include>
                                        
<include>commons-cli:commons-cli:jar:${commons.cli.version}</include>

http://git-wip-us.apache.org/repos/asf/ranger/blob/1ab501c3/tagsync/pom.xml
----------------------------------------------------------------------
diff --git a/tagsync/pom.xml b/tagsync/pom.xml
index 7e53641..8757f8d 100644
--- a/tagsync/pom.xml
+++ b/tagsync/pom.xml
@@ -55,11 +55,6 @@
             <version>${jersey-bundle.version}</version>
         </dependency>
         <dependency>
-            <groupId>com.sun.jersey.contribs</groupId>
-            <artifactId>jersey-multipart</artifactId>
-            <version>${sun-jersey-bundle.version}</version>
-        </dependency>
-        <dependency>
             <groupId>commons-cli</groupId>
             <artifactId>commons-cli</artifactId>
             <version>${commons.cli.version}</version>
@@ -100,14 +95,14 @@
             <version>${project.version}</version>
         </dependency>
         <dependency>
-            <groupId>org.codehaus.jackson</groupId>
-            <artifactId>jackson-core-asl</artifactId>
-            <version>${codehaus.jackson.version}</version>
+            <groupId>com.google.inject</groupId>
+            <artifactId>guice</artifactId>
+            <version>${guice.version}</version>
         </dependency>
         <dependency>
-            <groupId>org.codehaus.jackson</groupId>
-            <artifactId>jackson-jaxrs</artifactId>
-            <version>${codehaus.jackson.version}</version>
+            <groupId>com.google.inject.extensions</groupId>
+            <artifactId>guice-multibindings</artifactId>
+            <version>${guice.version}</version>
         </dependency>
         <dependency>
             <groupId>org.codehaus.jettison</groupId>
@@ -121,7 +116,7 @@
         </dependency>
         <dependency>
             <groupId>org.apache.atlas</groupId>
-            <artifactId>atlas-intg</artifactId>
+            <artifactId>atlas-typesystem</artifactId>
             <version>${atlas.version}</version>
         </dependency>
         <dependency>
@@ -131,11 +126,6 @@
         </dependency>
         <dependency>
             <groupId>org.apache.atlas</groupId>
-            <artifactId>atlas-client-v2</artifactId>
-            <version>${atlas.version}</version>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.atlas</groupId>
             <artifactId>atlas-common</artifactId>
             <version>${atlas.version}</version>
            <exclusions>
@@ -170,34 +160,53 @@
         </dependency>
         <dependency>
             <groupId>org.scala-lang</groupId>
+            <artifactId>scala-compiler</artifactId>
+            <version>${scala.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.scala-lang</groupId>
             <artifactId>scala-library</artifactId>
             <version>${scala.version}</version>
         </dependency>
         <dependency>
-            <groupId>com.fasterxml.jackson.core</groupId>
-            <artifactId>jackson-core</artifactId>
-            <version>${atlas.jackson.version}</version>
+            <groupId>org.scala-lang</groupId>
+            <artifactId>scalap</artifactId>
+            <version>${scala.version}</version>
         </dependency>
         <dependency>
-            <groupId>com.fasterxml.jackson.core</groupId>
-            <artifactId>jackson-databind</artifactId>
-            <version>${atlas.jackson.version}</version>
+            <groupId>org.scala-lang</groupId>
+            <artifactId>scala-reflect</artifactId>
+            <version>${scala.version}</version>
         </dependency>
         <dependency>
-            <groupId>com.fasterxml.jackson.jaxrs</groupId>
-            <artifactId>jackson-jaxrs-base</artifactId>
-            <version>${atlas.jackson.version}</version>
+            <groupId>org.scala-lang.modules</groupId>
+            <artifactId>scala-xml_${scala.binary.version}</artifactId>
+            <version>${scala.xml.version}</version>
         </dependency>
         <dependency>
-            <groupId>com.fasterxml.jackson.jaxrs</groupId>
-            <artifactId>jackson-jaxrs-json-provider</artifactId>
-            <version>${atlas.jackson.version}</version>
+            <groupId>org.scala-lang.modules</groupId>
+            
<artifactId>scala-parser-combinators_${scala.binary.version}</artifactId>
+            <version>${scala.xml.version}</version>
         </dependency>
         <dependency>
-            <groupId>com.fasterxml.jackson.core</groupId>
-            <artifactId>jackson-annotations</artifactId>
-            <version>${atlas.jackson.version}</version>
+            <groupId>org.json4s</groupId>
+            <artifactId>json4s-core_${scala.binary.version}</artifactId>
+            <version>${json4s.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.json4s</groupId>
+            <artifactId>json4s-native_${scala.binary.version}</artifactId>
+            <version>${json4s.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.json4s</groupId>
+            <artifactId>json4s-ast_${scala.binary.version}</artifactId>
+            <version>${json4s.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>com.thoughtworks.paranamer</groupId>
+            <artifactId>paranamer</artifactId>
+            <version>${paranamer.version}</version>
         </dependency>
-
     </dependencies>
 </project>

http://git-wip-us.apache.org/repos/asf/ranger/blob/1ab501c3/tagsync/src/main/java/org/apache/ranger/tagsync/process/TagSynchronizer.java
----------------------------------------------------------------------
diff --git 
a/tagsync/src/main/java/org/apache/ranger/tagsync/process/TagSynchronizer.java 
b/tagsync/src/main/java/org/apache/ranger/tagsync/process/TagSynchronizer.java
index 45997e4..b07cd34 100644
--- 
a/tagsync/src/main/java/org/apache/ranger/tagsync/process/TagSynchronizer.java
+++ 
b/tagsync/src/main/java/org/apache/ranger/tagsync/process/TagSynchronizer.java
@@ -107,7 +107,7 @@ public class TagSynchronizer {
 
                if (ret) {
                        LOG.info("Initializing TAG source and sink");
-                       ret = false;
+
                        tagSink = initializeTagSink(properties);
 
                        if (tagSink != null) {

http://git-wip-us.apache.org/repos/asf/ranger/blob/1ab501c3/tagsync/src/main/java/org/apache/ranger/tagsync/sink/tagadmin/TagAdminRESTSink.java
----------------------------------------------------------------------
diff --git 
a/tagsync/src/main/java/org/apache/ranger/tagsync/sink/tagadmin/TagAdminRESTSink.java
 
b/tagsync/src/main/java/org/apache/ranger/tagsync/sink/tagadmin/TagAdminRESTSink.java
index a1dc8f5..c34b6ea 100644
--- 
a/tagsync/src/main/java/org/apache/ranger/tagsync/sink/tagadmin/TagAdminRESTSink.java
+++ 
b/tagsync/src/main/java/org/apache/ranger/tagsync/sink/tagadmin/TagAdminRESTSink.java
@@ -65,13 +65,13 @@ public class TagAdminRESTSink implements TagSink, Runnable {
 
        @Override
        public boolean initialize(Properties properties) {
-               if (LOG.isDebugEnabled()) {
+               if(LOG.isDebugEnabled()) {
                        LOG.debug("==> TagAdminRESTSink.initialize()");
                }
 
                boolean ret = false;
 
-               String restUrl = TagSyncConfig.getTagAdminRESTUrl(properties);
+               String restUrl       = 
TagSyncConfig.getTagAdminRESTUrl(properties);
                String sslConfigFile = 
TagSyncConfig.getTagAdminRESTSslConfigFile(properties);
                String userName = TagSyncConfig.getTagAdminUserName(properties);
                String password = TagSyncConfig.getTagAdminPassword(properties);
@@ -89,19 +89,16 @@ public class TagAdminRESTSink implements TagSink, Runnable {
 
                if (StringUtils.isNotBlank(restUrl)) {
                        tagRESTClient = new RangerRESTClient(restUrl, 
sslConfigFile);
-                       if (!isKerberized) {
+                       if(!isKerberized) {
                                tagRESTClient.setBasicAuthInfo(userName, 
password);
                        }
-                       // Build and cache REST client. This will catch any 
errors in building REST client up-front
-                       tagRESTClient.getClient();
-
                        uploadWorkItems = new 
LinkedBlockingQueue<UploadWorkItem>();
                        ret = true;
                } else {
                        LOG.error("No value specified for property 
'ranger.tagsync.tagadmin.rest.url'!");
                }
 
-               if (LOG.isDebugEnabled()) {
+               if(LOG.isDebugEnabled()) {
                        LOG.debug("<== TagAdminRESTSink.initialize(), result=" 
+ ret);
                }
 

http://git-wip-us.apache.org/repos/asf/ranger/blob/1ab501c3/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasEntityWithTraits.java
----------------------------------------------------------------------
diff --git 
a/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasEntityWithTraits.java
 
b/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasEntityWithTraits.java
new file mode 100644
index 0000000..77dee01
--- /dev/null
+++ 
b/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasEntityWithTraits.java
@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.ranger.tagsync.source.atlas;
+
+import org.apache.atlas.AtlasException;
+import org.apache.atlas.typesystem.IReferenceableInstance;
+import org.apache.atlas.typesystem.IStruct;
+
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+
+public class AtlasEntityWithTraits {
+
+       private final IReferenceableInstance entity;
+       private final List<IStruct> traits;
+
+       public AtlasEntityWithTraits(IReferenceableInstance entity, 
List<IStruct> traits) {
+               this.entity = entity;
+               this.traits = traits;
+       }
+
+       public IReferenceableInstance getEntity() {
+               return entity;
+       }
+
+       public List<IStruct> getAllTraits() {
+               return traits == null ? new LinkedList<IStruct>() : traits;
+       }
+
+       @Override
+       public String toString( ) {
+               StringBuilder sb = new StringBuilder();
+
+               toString(sb);
+
+               return sb.toString();
+       }
+
+       public void toString(StringBuilder sb) {
+
+               sb.append("AtlasEntityWithTraits={ ");
+
+               sb.append("Entity-Id: " + entity.getId()._getId()).append(", ")
+                               .append("Entity-Type: " + 
entity.getTypeName()).append(", ")
+                               .append("Entity-Version: " + 
entity.getId().getVersion()).append(", ")
+                               .append("Entity-State: " + 
entity.getId().getStateAsString()).append(", ");
+
+               sb.append("Entity-Values={ ");
+               try {
+                       for (Map.Entry<String, Object> entry : 
entity.getValuesMap().entrySet()) {
+                               sb.append("{").append(entry.getKey()).append(", 
").append(entry.getValue()).append("}, ");
+                       }
+               } catch (AtlasException exception) {
+                               // Ignore
+               }
+               sb.append(" }");
+
+               sb.append(", Entity-Traits={ ");
+               for (IStruct trait : traits) {
+                       try {
+                               
sb.append("{traitType=").append(trait.getTypeName()).append(", ");
+                               Map<String, Object> traitValues = 
trait.getValuesMap();
+                               sb.append("{");
+                               for (Map.Entry<String, Object> valueEntry : 
traitValues.entrySet()) {
+                                       
sb.append("{").append(valueEntry.getKey()).append(", 
").append(valueEntry.getValue()).append("}");
+                               }
+                               sb.append("}");
+
+                               sb.append(" }");
+                       } catch (AtlasException exception) {
+                               // Ignore
+                       }
+               }
+               sb.append(" }");
+
+               sb.append(" }");
+
+       }
+
+}

http://git-wip-us.apache.org/repos/asf/ranger/blob/1ab501c3/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasHbaseResourceMapper.java
----------------------------------------------------------------------
diff --git 
a/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasHbaseResourceMapper.java
 
b/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasHbaseResourceMapper.java
index 33e804a..8b36a31 100644
--- 
a/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasHbaseResourceMapper.java
+++ 
b/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasHbaseResourceMapper.java
@@ -22,10 +22,10 @@ package org.apache.ranger.tagsync.source.atlas;
 import java.util.Map;
 import java.util.HashMap;
 
+import org.apache.atlas.typesystem.IReferenceableInstance;
 import org.apache.commons.lang.StringUtils;
 import org.apache.ranger.plugin.model.RangerPolicy.RangerPolicyResource;
 import org.apache.ranger.plugin.model.RangerServiceResource;
-import org.apache.ranger.tagsync.source.atlasrest.RangerAtlasEntity;
 
 public class AtlasHbaseResourceMapper extends AtlasResourceMapper {
        public static final String ENTITY_TYPE_HBASE_TABLE          = 
"hbase_table";
@@ -36,6 +36,10 @@ public class AtlasHbaseResourceMapper extends 
AtlasResourceMapper {
        public static final String RANGER_TYPE_HBASE_COLUMN_FAMILY  = 
"column-family";
        public static final String RANGER_TYPE_HBASE_COLUMN         = "column";
 
+       public static final String ENTITY_ATTRIBUTE_QUALIFIED_NAME  = 
"qualifiedName";
+       public static final String QUALIFIED_NAME_DELIMITER         = "\\.";
+       public static final Character QUALIFIED_NAME_DELIMITER_CHAR    = '.';
+
        public static final String[] SUPPORTED_ENTITY_TYPES = { 
ENTITY_TYPE_HBASE_TABLE, ENTITY_TYPE_HBASE_COLUMN_FAMILY, 
ENTITY_TYPE_HBASE_COLUMN };
 
        public AtlasHbaseResourceMapper() {
@@ -43,8 +47,8 @@ public class AtlasHbaseResourceMapper extends 
AtlasResourceMapper {
        }
 
        @Override
-       public RangerServiceResource buildResource(final RangerAtlasEntity 
entity) throws Exception {
-               String qualifiedName = 
(String)entity.getAttributes().get(AtlasResourceMapper.ENTITY_ATTRIBUTE_QUALIFIED_NAME);
+       public RangerServiceResource buildResource(final IReferenceableInstance 
entity) throws Exception {
+               String qualifiedName = getEntityAttribute(entity, 
ENTITY_ATTRIBUTE_QUALIFIED_NAME, String.class);
                if (StringUtils.isEmpty(qualifiedName)) {
                        throw new Exception("attribute '" +  
ENTITY_ATTRIBUTE_QUALIFIED_NAME + "' not found in entity");
                }
@@ -60,7 +64,7 @@ public class AtlasHbaseResourceMapper extends 
AtlasResourceMapper {
                }
 
                String entityType  = entity.getTypeName();
-               String entityGuid  = entity.getGuid();
+               String entityGuid  = entity.getId() != null ? 
entity.getId()._getId() : null;
                String serviceName = getRangerServiceName(clusterName);
 
                Map<String, RangerPolicyResource> elements = new 
HashMap<String, RangerPolicyResource>();

http://git-wip-us.apache.org/repos/asf/ranger/blob/1ab501c3/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasHdfsResourceMapper.java
----------------------------------------------------------------------
diff --git 
a/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasHdfsResourceMapper.java
 
b/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasHdfsResourceMapper.java
index 378542c..06bff90 100644
--- 
a/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasHdfsResourceMapper.java
+++ 
b/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasHdfsResourceMapper.java
@@ -22,12 +22,12 @@ package org.apache.ranger.tagsync.source.atlas;
 import java.util.HashMap;
 import java.util.Map;
 
+import org.apache.atlas.typesystem.IReferenceableInstance;
 import org.apache.commons.lang.StringUtils;
 import org.apache.hadoop.fs.Path;
 import org.apache.ranger.plugin.model.RangerPolicy;
 import org.apache.ranger.plugin.model.RangerPolicy.RangerPolicyResource;
 import org.apache.ranger.plugin.model.RangerServiceResource;
-import org.apache.ranger.tagsync.source.atlasrest.RangerAtlasEntity;
 
 public class AtlasHdfsResourceMapper extends AtlasResourceMapper {
        public static final String ENTITY_TYPE_HDFS_PATH = "hdfs_path";
@@ -35,6 +35,7 @@ public class AtlasHdfsResourceMapper extends 
AtlasResourceMapper {
 
        public static final String ENTITY_ATTRIBUTE_PATH           = "path";
        public static final String ENTITY_ATTRIBUTE_CLUSTER_NAME   = 
"clusterName";
+       public static final String ENTITY_ATTRIBUTE_QUALIFIED_NAME = 
"qualifiedName";
 
        public static final String[] SUPPORTED_ENTITY_TYPES = { 
ENTITY_TYPE_HDFS_PATH };
 
@@ -55,10 +56,10 @@ public class AtlasHdfsResourceMapper extends 
AtlasResourceMapper {
        }
 
        @Override
-       public RangerServiceResource buildResource(final RangerAtlasEntity 
entity) throws Exception {
-               String path          = 
(String)entity.getAttributes().get(ENTITY_ATTRIBUTE_PATH);
-               String clusterName   = 
(String)entity.getAttributes().get(ENTITY_ATTRIBUTE_CLUSTER_NAME);
-               String qualifiedName = 
(String)entity.getAttributes().get(AtlasResourceMapper.ENTITY_ATTRIBUTE_QUALIFIED_NAME);
+       public RangerServiceResource buildResource(final IReferenceableInstance 
entity) throws Exception {
+               String path          = getEntityAttribute(entity, 
ENTITY_ATTRIBUTE_PATH, String.class);
+               String clusterName   = getEntityAttribute(entity, 
ENTITY_ATTRIBUTE_CLUSTER_NAME, String.class);
+               String qualifiedName = getEntityAttribute(entity, 
ENTITY_ATTRIBUTE_QUALIFIED_NAME, String.class);
 
                if(StringUtils.isEmpty(path)) {
                        path = getResourceNameFromQualifiedName(qualifiedName);
@@ -80,7 +81,7 @@ public class AtlasHdfsResourceMapper extends 
AtlasResourceMapper {
                        }
                }
 
-               String  entityGuid  = entity.getGuid();
+               String  entityGuid  = entity.getId() != null ? 
entity.getId()._getId() : null;
                String  serviceName = getRangerServiceName(clusterName);
                Boolean isExcludes  = Boolean.FALSE;
                Boolean isRecursive = Boolean.TRUE;

http://git-wip-us.apache.org/repos/asf/ranger/blob/1ab501c3/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasHiveResourceMapper.java
----------------------------------------------------------------------
diff --git 
a/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasHiveResourceMapper.java
 
b/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasHiveResourceMapper.java
index 3e0a97f..a359622 100644
--- 
a/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasHiveResourceMapper.java
+++ 
b/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasHiveResourceMapper.java
@@ -22,10 +22,10 @@ package org.apache.ranger.tagsync.source.atlas;
 import java.util.Map;
 import java.util.HashMap;
 
+import org.apache.atlas.typesystem.IReferenceableInstance;
 import org.apache.commons.lang.StringUtils;
 import org.apache.ranger.plugin.model.RangerPolicy.RangerPolicyResource;
 import org.apache.ranger.plugin.model.RangerServiceResource;
-import org.apache.ranger.tagsync.source.atlasrest.RangerAtlasEntity;
 
 public class AtlasHiveResourceMapper extends AtlasResourceMapper {
        public static final String ENTITY_TYPE_HIVE_DB     = "hive_db";
@@ -36,6 +36,9 @@ public class AtlasHiveResourceMapper extends 
AtlasResourceMapper {
        public static final String RANGER_TYPE_HIVE_TABLE  = "table";
        public static final String RANGER_TYPE_HIVE_COLUMN = "column";
 
+       public static final String ENTITY_ATTRIBUTE_QUALIFIED_NAME = 
"qualifiedName";
+       public static final String QUALIFIED_NAME_DELIMITER        = "\\.";
+
        public static final String[] SUPPORTED_ENTITY_TYPES = { 
ENTITY_TYPE_HIVE_DB, ENTITY_TYPE_HIVE_TABLE, ENTITY_TYPE_HIVE_COLUMN };
 
        public AtlasHiveResourceMapper() {
@@ -43,8 +46,8 @@ public class AtlasHiveResourceMapper extends 
AtlasResourceMapper {
        }
 
        @Override
-       public RangerServiceResource buildResource(final RangerAtlasEntity 
entity) throws Exception {
-               String qualifiedName = 
(String)entity.getAttributes().get(AtlasResourceMapper.ENTITY_ATTRIBUTE_QUALIFIED_NAME);
+       public RangerServiceResource buildResource(final IReferenceableInstance 
entity) throws Exception {
+               String qualifiedName = getEntityAttribute(entity, 
ENTITY_ATTRIBUTE_QUALIFIED_NAME, String.class);
                if (StringUtils.isEmpty(qualifiedName)) {
                        throw new Exception("attribute '" +  
ENTITY_ATTRIBUTE_QUALIFIED_NAME + "' not found in entity");
                }
@@ -60,7 +63,7 @@ public class AtlasHiveResourceMapper extends 
AtlasResourceMapper {
                }
 
                String   entityType  = entity.getTypeName();
-               String   entityGuid  = entity.getGuid();
+               String   entityGuid  = entity.getId() != null ? 
entity.getId()._getId() : null;
                String   serviceName = getRangerServiceName(clusterName);
                String[] resources   = 
resourceStr.split(QUALIFIED_NAME_DELIMITER);
                String   dbName      = resources.length > 0 ? resources[0] : 
null;

http://git-wip-us.apache.org/repos/asf/ranger/blob/1ab501c3/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasKafkaResourceMapper.java
----------------------------------------------------------------------
diff --git 
a/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasKafkaResourceMapper.java
 
b/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasKafkaResourceMapper.java
index 86e37c3..09ae5d1 100644
--- 
a/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasKafkaResourceMapper.java
+++ 
b/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasKafkaResourceMapper.java
@@ -22,16 +22,18 @@ package org.apache.ranger.tagsync.source.atlas;
 import java.util.HashMap;
 import java.util.Map;
 
+import org.apache.atlas.typesystem.IReferenceableInstance;
 import org.apache.commons.lang.StringUtils;
 import org.apache.ranger.plugin.model.RangerPolicy;
 import org.apache.ranger.plugin.model.RangerPolicy.RangerPolicyResource;
 import org.apache.ranger.plugin.model.RangerServiceResource;
-import org.apache.ranger.tagsync.source.atlasrest.RangerAtlasEntity;
 
 public class AtlasKafkaResourceMapper extends AtlasResourceMapper {
        public static final String ENTITY_TYPE_KAFKA_TOPIC = "kafka_topic";
        public static final String RANGER_TYPE_KAFKA_TOPIC = "topic";
 
+       public static final String ENTITY_ATTRIBUTE_QUALIFIED_NAME = 
"qualifiedName";
+
        public static final String[] SUPPORTED_ENTITY_TYPES = { 
ENTITY_TYPE_KAFKA_TOPIC };
 
        public AtlasKafkaResourceMapper() {
@@ -39,8 +41,12 @@ public class AtlasKafkaResourceMapper extends 
AtlasResourceMapper {
        }
 
        @Override
-       public RangerServiceResource buildResource(final RangerAtlasEntity 
entity) throws Exception {
-               String qualifiedName = 
(String)entity.getAttributes().get(AtlasResourceMapper.ENTITY_ATTRIBUTE_QUALIFIED_NAME);
+       public RangerServiceResource buildResource(final IReferenceableInstance 
entity) throws Exception {
+               String qualifiedName = getEntityAttribute(entity, 
ENTITY_ATTRIBUTE_QUALIFIED_NAME, String.class);
+
+               if(StringUtils.isEmpty(qualifiedName)) {
+                       throwExceptionWithMessage("attribute '" + 
ENTITY_ATTRIBUTE_QUALIFIED_NAME +  "' not found in entity");
+               }
 
                String topic = getResourceNameFromQualifiedName(qualifiedName);
 
@@ -55,17 +61,16 @@ public class AtlasKafkaResourceMapper extends 
AtlasResourceMapper {
                }
 
                if(StringUtils.isEmpty(clusterName)) {
-                       throwExceptionWithMessage("attribute '" + 
ENTITY_ATTRIBUTE_QUALIFIED_NAME +  "' not found in entity");
+                       throwExceptionWithMessage("Cluster name not found in 
attribute '" + ENTITY_ATTRIBUTE_QUALIFIED_NAME +  "'");
                }
 
-
                Map<String, RangerPolicyResource> elements = new 
HashMap<String, RangerPolicy.RangerPolicyResource>();
                Boolean isExcludes  = Boolean.FALSE;
                Boolean isRecursive = Boolean.TRUE;
 
                elements.put(RANGER_TYPE_KAFKA_TOPIC, new 
RangerPolicyResource(topic, isExcludes, isRecursive));
 
-               String  entityGuid  = entity.getGuid();
+               String  entityGuid  = entity.getId() != null ? 
entity.getId()._getId() : null;
                String  serviceName = getRangerServiceName(clusterName);
 
                return new RangerServiceResource(entityGuid, serviceName, 
elements);

http://git-wip-us.apache.org/repos/asf/ranger/blob/1ab501c3/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasNotificationMapper.java
----------------------------------------------------------------------
diff --git 
a/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasNotificationMapper.java
 
b/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasNotificationMapper.java
index 91cf606..f007ae5 100644
--- 
a/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasNotificationMapper.java
+++ 
b/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasNotificationMapper.java
@@ -19,9 +19,11 @@
 
 package org.apache.ranger.tagsync.source.atlas;
 
-import org.apache.atlas.v1.model.notification.EntityNotificationV1;
-import org.apache.atlas.v1.model.instance.Id;
-import org.apache.atlas.v1.model.instance.Referenceable;
+import org.apache.atlas.AtlasException;
+import org.apache.atlas.notification.entity.EntityNotification;
+import org.apache.atlas.typesystem.IReferenceableInstance;
+import org.apache.atlas.typesystem.IStruct;
+import org.apache.atlas.typesystem.persistence.Id;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.collections.MapUtils;
 import org.apache.commons.lang.StringUtils;
@@ -32,10 +34,9 @@ import org.apache.ranger.plugin.model.RangerTag;
 import org.apache.ranger.plugin.model.RangerTagDef;
 import org.apache.ranger.plugin.model.RangerTagDef.RangerTagAttributeDef;
 import org.apache.ranger.plugin.util.ServiceTags;
-import org.apache.ranger.tagsync.source.atlasrest.RangerAtlasEntity;
-import org.apache.ranger.tagsync.source.atlasrest.RangerAtlasEntityWithTags;
 
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -46,7 +47,7 @@ public class AtlasNotificationMapper {
 
        private static Map<String, Long> unhandledEventTypes = new 
HashMap<String, Long>();
 
-       private static void logUnhandledEntityNotification(EntityNotificationV1 
entityNotification) {
+       private static void logUnhandledEntityNotification(EntityNotification 
entityNotification) {
 
                final int REPORTING_INTERVAL_FOR_UNHANDLED_ENTITYTYPE_IN_MILLIS 
= 5 * 60 * 1000; // 5 minutes
 
@@ -76,30 +77,39 @@ public class AtlasNotificationMapper {
        }
 
        @SuppressWarnings("unchecked")
-       public static ServiceTags 
processEntityNotification(EntityNotificationV1 entityNotification) {
+       public static ServiceTags processEntityNotification(EntityNotification 
entityNotification) {
 
                ServiceTags ret = null;
 
                if (isNotificationHandled(entityNotification)) {
                        try {
-                               RangerAtlasEntityWithTags entityWithTags = new 
RangerAtlasEntityWithTags(entityNotification);
-
-                               if (entityNotification.getOperationType() == 
EntityNotificationV1.OperationType.ENTITY_DELETE) {
-                                       ret = 
buildServiceTagsForEntityDeleteNotification(entityWithTags);
+                               IReferenceableInstance entity = 
entityNotification.getEntity();
+
+                               if (entity != null && 
AtlasResourceMapperUtil.isEntityTypeHandled(entity.getTypeName())) {
+                                       AtlasEntityWithTraits entityWithTraits 
= new AtlasEntityWithTraits(entity, entityNotification.getAllTraits());
+                                       if 
(entityNotification.getOperationType() == 
EntityNotification.OperationType.ENTITY_DELETE) {
+                                               ret = 
buildServiceTagsForEntityDeleteNotification(entityWithTraits);
+                                       } else {
+                                               if (entity.getId().getState() 
== Id.EntityState.ACTIVE) {
+                                                       ret = 
buildServiceTags(entityWithTraits, null);
+                                               } else {
+                                                       if 
(LOG.isDebugEnabled()) {
+                                                               
LOG.debug("Ignoring entityNotification for entity that is not ACTIVE: " + 
entityWithTraits);
+                                                       }
+                                               }
+                                       }
                                } else {
-                                       ret = buildServiceTags(entityWithTags, 
null);
+                                       
logUnhandledEntityNotification(entityNotification);
                                }
 
                        } catch (Exception exception) {
                                LOG.error("createServiceTags() failed!! ", 
exception);
                        }
-               } else {
-                       logUnhandledEntityNotification(entityNotification);
                }
                return ret;
        }
 
-       public static Map<String, ServiceTags> 
processAtlasEntities(List<RangerAtlasEntityWithTags> atlasEntities) {
+       public static Map<String, ServiceTags> 
processAtlasEntities(List<AtlasEntityWithTraits> atlasEntities) {
                Map<String, ServiceTags> ret = null;
 
                try {
@@ -111,16 +121,17 @@ public class AtlasNotificationMapper {
                return ret;
        }
 
-       static private boolean isNotificationHandled(EntityNotificationV1 
entityNotification) {
+       static private boolean isNotificationHandled(EntityNotification 
entityNotification) {
                boolean ret = false;
 
-               EntityNotificationV1.OperationType opType = 
entityNotification.getOperationType();
+               EntityNotification.OperationType opType = 
entityNotification.getOperationType();
 
-               if (opType != null) {
+               if(opType != null) {
                        switch (opType) {
-                               case ENTITY_CREATE:
-                                       ret = 
CollectionUtils.isNotEmpty(entityNotification.getAllTraits());
+                               case ENTITY_CREATE: {
+                                       LOG.debug("ENTITY_CREATE notification 
is not handled, as Ranger will get necessary information from any subsequent 
TRAIT_ADDED notification");
                                        break;
+                               }
                                case ENTITY_UPDATE:
                                case ENTITY_DELETE:
                                case TRAIT_ADD:
@@ -131,38 +142,30 @@ public class AtlasNotificationMapper {
                                }
                                default:
                                        LOG.error(opType + ": unknown 
notification received - not handled");
-                                       break;
-                       }
-                       if (ret) {
-                               final Referenceable entity = 
entityNotification.getEntity();
-
-                               ret = entity != null
-                                               && entity.getId().getState() == 
Id.EntityState.ACTIVE
-                                               && 
AtlasResourceMapperUtil.isEntityTypeHandled(entity.getTypeName());
                        }
                }
 
                return ret;
        }
 
-       static private ServiceTags 
buildServiceTagsForEntityDeleteNotification(RangerAtlasEntityWithTags 
entityWithTags) throws Exception {
+       static private ServiceTags 
buildServiceTagsForEntityDeleteNotification(AtlasEntityWithTraits 
entityWithTraits) throws Exception {
                final ServiceTags ret;
 
-               RangerAtlasEntity entity = entityWithTags.getEntity();
+               IReferenceableInstance entity = entityWithTraits.getEntity();
 
-               String guid = entity.getGuid();
+               String guid = entity.getId()._getId();
                if (StringUtils.isNotBlank(guid)) {
                        ret = new ServiceTags();
                        RangerServiceResource serviceResource = new 
RangerServiceResource();
                        serviceResource.setGuid(guid);
                        ret.getServiceResources().add(serviceResource);
                } else {
-                       ret = buildServiceTags(entityWithTags, null);
+                       ret = buildServiceTags(entityWithTraits, null);
                        if (ret != null) {
                                // tag-definitions should NOT be deleted as 
part of service-resource delete
-                               ret.setTagDefinitions(MapUtils.EMPTY_MAP);
+                               ret.setTagDefinitions(Collections.<Long, 
RangerTagDef>emptyMap());
                                // Ranger deletes tags associated with deleted 
service-resource
-                               ret.setTags(MapUtils.EMPTY_MAP);
+                               ret.setTags(Collections.<Long, 
RangerTag>emptyMap());
                        }
                }
 
@@ -173,13 +176,13 @@ public class AtlasNotificationMapper {
                return ret;
        }
 
-       static private Map<String, ServiceTags> 
buildServiceTags(List<RangerAtlasEntityWithTags> entitiesWithTags) throws 
Exception {
+       static private Map<String, ServiceTags> 
buildServiceTags(List<AtlasEntityWithTraits> entitiesWithTraits) throws 
Exception {
 
                Map<String, ServiceTags> ret = new HashMap<String, 
ServiceTags>();
 
-               for (RangerAtlasEntityWithTags element : entitiesWithTags) {
-                       RangerAtlasEntity entity = element.getEntity();
-                       if (entity != null) {
+               for (AtlasEntityWithTraits element : entitiesWithTraits) {
+                       IReferenceableInstance entity = element.getEntity();
+                       if (entity != null && entity.getId().getState() == 
Id.EntityState.ACTIVE) {
                                buildServiceTags(element, ret);
                        } else {
                                if (LOG.isDebugEnabled()) {
@@ -238,15 +241,15 @@ public class AtlasNotificationMapper {
                return ret;
        }
 
-       static private ServiceTags buildServiceTags(RangerAtlasEntityWithTags 
entityWithTags, Map<String, ServiceTags> serviceTagsMap) throws Exception {
+       static private ServiceTags buildServiceTags(AtlasEntityWithTraits 
entityWithTraits, Map<String, ServiceTags> serviceTagsMap) throws Exception {
                ServiceTags            ret             = null;
-               RangerAtlasEntity entity          = entityWithTags.getEntity();
+               IReferenceableInstance entity          = 
entityWithTraits.getEntity();
                RangerServiceResource  serviceResource = 
AtlasResourceMapperUtil.getRangerServiceResource(entity);
 
                if (serviceResource != null) {
 
-                       List<RangerTag>     tags        = 
getTags(entityWithTags);
-                       List<RangerTagDef>  tagDefs     = 
getTagDefs(entityWithTags);
+                       List<RangerTag>     tags        = 
getTags(entityWithTraits);
+                       List<RangerTagDef>  tagDefs     = 
getTagDefs(entityWithTraits);
                        String              serviceName = 
serviceResource.getServiceName();
 
                        ret = createOrGetServiceTags(serviceTagsMap, 
serviceName);
@@ -276,12 +279,12 @@ public class AtlasNotificationMapper {
                                }
                        } else {
                                if (LOG.isDebugEnabled()) {
-                                       LOG.debug("Entity " + entityWithTags + 
" does not have any tags associated with it when full-sync is being done.");
+                                       LOG.debug("Entity " + entityWithTraits 
+ " does not have any tags associated with it when full-sync is being done.");
                                        LOG.debug("Will not add this entity to 
serviceTags, so that this entity, if exists,  will be removed from ranger");
                                }
                        }
                } else {
-                       LOG.error("Failed to build serviceResource for entity:" 
+ entity.getGuid());
+                       LOG.error("Failed to build serviceResource for entity:" 
+ entity.getId()._getId());
                }
 
                return ret;
@@ -304,33 +307,58 @@ public class AtlasNotificationMapper {
                return ret;
        }
 
-       static private List<RangerTag> getTags(RangerAtlasEntityWithTags 
entityWithTags) {
+       static private List<RangerTag> getTags(AtlasEntityWithTraits 
entityWithTraits) {
                List<RangerTag> ret = new ArrayList<RangerTag>();
 
-               if (entityWithTags != null && 
MapUtils.isNotEmpty(entityWithTags.getTags())) {
-                       Map<String, Map<String, String>> tags = 
entityWithTags.getTags();
+               if(entityWithTraits != null && 
CollectionUtils.isNotEmpty(entityWithTraits.getAllTraits())) {
+                       List<IStruct> traits = entityWithTraits.getAllTraits();
+
+                       for (IStruct trait : traits) {
+                               Map<String, String> tagAttrs = new 
HashMap<String, String>();
 
-                       for (Map.Entry<String, Map<String, String>> tag : 
tags.entrySet()) {
-                               ret.add(new RangerTag(null, tag.getKey(), 
tag.getValue(), RangerTag.OWNER_SERVICERESOURCE));
+                               try {
+                                       Map<String, Object> attrs = 
trait.getValuesMap();
+
+                                       if(MapUtils.isNotEmpty(attrs)) {
+                                               for (Map.Entry<String, Object> 
attrEntry : attrs.entrySet()) {
+                                                       String attrName  = 
attrEntry.getKey();
+                                                       Object attrValue = 
attrEntry.getValue();
+
+                                                       tagAttrs.put(attrName, 
attrValue != null ? attrValue.toString() : null);
+                                               }
+                                       }
+                               } catch (AtlasException exception) {
+                                       LOG.error("Could not get values for 
trait:" + trait.getTypeName(), exception);
+                               }
+
+                               ret.add(new RangerTag(null, 
trait.getTypeName(), tagAttrs, RangerTag.OWNER_SERVICERESOURCE));
                        }
                }
 
                return ret;
        }
 
-       static private List<RangerTagDef> getTagDefs(RangerAtlasEntityWithTags 
entityWithTags) {
+       static private List<RangerTagDef> getTagDefs(AtlasEntityWithTraits 
entityWithTraits) {
                List<RangerTagDef> ret = new ArrayList<RangerTagDef>();
 
-               if (entityWithTags != null && 
MapUtils.isNotEmpty(entityWithTags.getTags())) {
-                       Map<String, Map<String, String>> tags = 
entityWithTags.getTags();
+               if(entityWithTraits != null && 
CollectionUtils.isNotEmpty(entityWithTraits.getAllTraits())) {
+                       List<IStruct> traits = entityWithTraits.getAllTraits();
+
+                       for (IStruct trait : traits) {
+                               RangerTagDef tagDef = new 
RangerTagDef(trait.getTypeName(), "Atlas");
+
+                               try {
+                                       Map<String, Object> attrs = 
trait.getValuesMap();
 
-                       for (Map.Entry<String, Map<String, String>> tag : 
tags.entrySet()) {
-                               RangerTagDef tagDef = new 
RangerTagDef(tag.getKey(), "Atlas");
-                               if (MapUtils.isNotEmpty(tag.getValue())) {
-                                       for (String attributeName : 
tag.getValue().keySet()) {
-                                               
tagDef.getAttributeDefs().add(new RangerTagAttributeDef(attributeName, 
entityWithTags.getTagAttributeType(tag.getKey(), attributeName)));
+                                       if(MapUtils.isNotEmpty(attrs)) {
+                                               for (String attrName : 
attrs.keySet()) {
+                                                       
tagDef.getAttributeDefs().add(new RangerTagAttributeDef(attrName, "string"));
+                                               }
                                        }
+                               } catch (AtlasException exception) {
+                                       LOG.error("Could not get values for 
trait:" + trait.getTypeName(), exception);
                                }
+
                                ret.add(tagDef);
                        }
                }

http://git-wip-us.apache.org/repos/asf/ranger/blob/1ab501c3/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasResourceMapper.java
----------------------------------------------------------------------
diff --git 
a/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasResourceMapper.java
 
b/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasResourceMapper.java
index 5d067a5..8ececdf 100644
--- 
a/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasResourceMapper.java
+++ 
b/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasResourceMapper.java
@@ -20,20 +20,19 @@
 package org.apache.ranger.tagsync.source.atlas;
 
 import java.util.Properties;
+import java.util.Map;
 
+import org.apache.atlas.AtlasException;
+import org.apache.atlas.typesystem.IReferenceableInstance;
 import org.apache.commons.lang.StringUtils;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.ranger.plugin.model.RangerServiceResource;
-import org.apache.ranger.tagsync.source.atlasrest.RangerAtlasEntity;
 
 public abstract class AtlasResourceMapper {
        private static final Log LOG = 
LogFactory.getLog(AtlasResourceMapper.class);
 
        public static final String TAGSYNC_DEFAULT_CLUSTER_NAME = 
"ranger.tagsync.atlas.default.cluster.name";
-       public static final String ENTITY_ATTRIBUTE_QUALIFIED_NAME = 
"qualifiedName";
-       public static final String QUALIFIED_NAME_DELIMITER        = "\\.";
-       public static final Character QUALIFIED_NAME_DELIMITER_CHAR    = '.';
 
        protected static final String TAGSYNC_SERVICENAME_MAPPER_PROP_PREFIX    
              = "ranger.tagsync.atlas.";
        protected static final String TAGSYNC_SERVICENAME_MAPPER_PROP_SUFFIX    
              = ".ranger.service";
@@ -74,7 +73,7 @@ public abstract class AtlasResourceMapper {
                this.defaultClusterName = properties != null ? 
properties.getProperty(TAGSYNC_DEFAULT_CLUSTER_NAME) : null;
        }
 
-       abstract public RangerServiceResource buildResource(final 
RangerAtlasEntity entity) throws Exception;
+       abstract public RangerServiceResource buildResource(final 
IReferenceableInstance entity) throws Exception;
 
        protected String getCustomRangerServiceName(String atlasInstanceName) {
                if(properties != null) {
@@ -119,4 +118,21 @@ public abstract class AtlasResourceMapper {
 
                throw new Exception(msg);
        }
+
+       static protected <T> T getEntityAttribute(IReferenceableInstance 
entity, String name, Class<T> type) {
+               T ret = null;
+
+               try {
+                       Map<String, Object> valueMap = entity.getValuesMap();
+                       ret = getAttribute(valueMap, name, type);
+               } catch (AtlasException exception) {
+                       LOG.error("Cannot get map of values for entity: " + 
entity.getId()._getId(), exception);
+               }
+
+               return ret;
+       }
+
+       static protected <T> T getAttribute(Map<String, Object> map, String 
name, Class<T> type) {
+               return type.cast(map.get(name));
+       }
 }

http://git-wip-us.apache.org/repos/asf/ranger/blob/1ab501c3/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasResourceMapperUtil.java
----------------------------------------------------------------------
diff --git 
a/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasResourceMapperUtil.java
 
b/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasResourceMapperUtil.java
index cd2cb63..40a639b 100644
--- 
a/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasResourceMapperUtil.java
+++ 
b/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasResourceMapperUtil.java
@@ -19,6 +19,7 @@
 
 package org.apache.ranger.tagsync.source.atlas;
 
+import org.apache.atlas.typesystem.IReferenceableInstance;
 import org.apache.commons.lang.StringUtils;
 import org.apache.ranger.plugin.model.RangerServiceResource;
 
@@ -27,13 +28,14 @@ import java.util.*;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.ranger.tagsync.process.TagSyncConfig;
-import org.apache.ranger.tagsync.source.atlasrest.RangerAtlasEntity;
 
 public class AtlasResourceMapperUtil {
        private static final Log LOG = 
LogFactory.getLog(AtlasResourceMapperUtil.class);
 
        private static Map<String, AtlasResourceMapper> atlasResourceMappers = 
new HashMap<String, AtlasResourceMapper>();
 
+       private static final String MAPPER_NAME_DELIMITER = ",";
+
        public static boolean isEntityTypeHandled(String entityTypeName) {
                if (LOG.isDebugEnabled()) {
                        LOG.debug("==> isEntityTypeHandled(entityTypeName=" + 
entityTypeName + ")");
@@ -50,9 +52,9 @@ public class AtlasResourceMapperUtil {
                return ret;
        }
 
-       public static RangerServiceResource 
getRangerServiceResource(RangerAtlasEntity atlasEntity) {
+       public static RangerServiceResource 
getRangerServiceResource(IReferenceableInstance atlasEntity) {
                if (LOG.isDebugEnabled()) {
-                       LOG.debug("==> getRangerServiceResource(" + 
atlasEntity.getGuid() +")");
+                       LOG.debug("==> getRangerServiceResource(" + 
atlasEntity.getId()._getId() +")");
                }
 
                RangerServiceResource resource = null;
@@ -63,63 +65,59 @@ public class AtlasResourceMapperUtil {
                        try {
                                resource = mapper.buildResource(atlasEntity);
                        } catch (Exception exception) {
-                               LOG.error("Could not get serviceResource for 
atlas entity:" + atlasEntity.getGuid() + ": ", exception);
+                               LOG.error("Could not get serviceResource for 
atlas entity:" + atlasEntity.getId()._getId() + ": ", exception);
                        }
                }
 
                if (LOG.isDebugEnabled()) {
-                       LOG.debug("<== getRangerServiceResource(" + 
atlasEntity.getGuid() +"): resource=" + resource);
+                       LOG.debug("<== getRangerServiceResource(" + 
atlasEntity.getId()._getId() +"): resource=" + resource);
                }
 
                return resource;
        }
 
        static public boolean initializeAtlasResourceMappers(Properties 
properties) {
-               final String MAPPER_NAME_DELIMITER = ",";
-
                String customMapperNames = 
TagSyncConfig.getCustomAtlasResourceMappers(properties);
 
                if (LOG.isDebugEnabled()) {
                        LOG.debug("==> 
initializeAtlasResourceMappers.initializeAtlasResourceMappers(" + 
customMapperNames + ")");
                }
-               boolean ret = true;
 
-               List<String> mapperNames = new ArrayList<String>();
-               
mapperNames.add("org.apache.ranger.tagsync.source.atlas.AtlasHiveResourceMapper");
-               
mapperNames.add("org.apache.ranger.tagsync.source.atlas.AtlasHdfsResourceMapper");
-               
mapperNames.add("org.apache.ranger.tagsync.source.atlas.AtlasHbaseResourceMapper");
-               
mapperNames.add("org.apache.ranger.tagsync.source.atlas.AtlasKafkaResourceMapper");
+               // Initialize the default mappers
+               initializeAtlasResourceMapper(new AtlasHiveResourceMapper(), 
properties);
+               initializeAtlasResourceMapper(new AtlasHdfsResourceMapper(), 
properties);
+               initializeAtlasResourceMapper(new AtlasHbaseResourceMapper(), 
properties);
+               initializeAtlasResourceMapper(new AtlasKafkaResourceMapper(), 
properties);
+               initializeAtlasResourceMapper(new AtlasStormResourceMapper(), 
properties);
 
+               // Initialize the custom mappers
+               boolean ret = true;
                if (StringUtils.isNotBlank(customMapperNames)) {
                        for (String customMapperName : 
customMapperNames.split(MAPPER_NAME_DELIMITER)) {
-                               mapperNames.add(customMapperName.trim());
-                       }
-               }
-
-               for (String mapperName : mapperNames) {
-                       try {
-                               Class<?> clazz = Class.forName(mapperName);
-                               AtlasResourceMapper resourceMapper = 
(AtlasResourceMapper) clazz.newInstance();
-
-                               resourceMapper.initialize(properties);
-
-                               for (String entityTypeName : 
resourceMapper.getSupportedEntityTypes()) {
-                                       add(entityTypeName, resourceMapper);
-                               }
-
-                       } catch (Exception exception) {
-                               LOG.error("Failed to create 
AtlasResourceMapper:" + mapperName + ": ", exception);
-                               ret = false;
+                           try {
+                               Class<?> clazz = 
Class.forName(customMapperName);
+                               AtlasResourceMapper resourceMapper = 
(AtlasResourceMapper) clazz.newInstance();
+
+                               initializeAtlasResourceMapper(resourceMapper, 
properties);
+                           } catch (Exception exception) {
+                               LOG.error("Failed to create 
AtlasResourceMapper:" + customMapperName + ": ", exception);
+                               ret = false;
+                           }
                        }
                }
 
                if (LOG.isDebugEnabled()) {
-                       LOG.debug("<== 
initializeAtlasResourceMappers.initializeAtlasResourceMappers(" + mapperNames + 
"): " + ret);
+                       LOG.debug("<== 
initializeAtlasResourceMappers.initializeAtlasResourceMappers(" + 
customMapperNames + "): " + ret);
                }
                return ret;
        }
 
-       private static void add(String entityType, AtlasResourceMapper mapper) {
-               atlasResourceMappers.put(entityType, mapper);
+       private static void initializeAtlasResourceMapper(AtlasResourceMapper 
resourceMapper, Properties properties) {
+           resourceMapper.initialize(properties);
+
+        for (String entityTypeName : resourceMapper.getSupportedEntityTypes()) 
{
+            atlasResourceMappers.put(entityTypeName, resourceMapper);
+        }
        }
+
 }

http://git-wip-us.apache.org/repos/asf/ranger/blob/1ab501c3/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasStormResourceMapper.java
----------------------------------------------------------------------
diff --git 
a/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasStormResourceMapper.java
 
b/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasStormResourceMapper.java
index 650968d..4ed01ca 100644
--- 
a/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasStormResourceMapper.java
+++ 
b/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasStormResourceMapper.java
@@ -22,15 +22,17 @@ package org.apache.ranger.tagsync.source.atlas;
 import java.util.HashMap;
 import java.util.Map;
 
+import org.apache.atlas.typesystem.IReferenceableInstance;
 import org.apache.commons.lang.StringUtils;
 import org.apache.ranger.plugin.model.RangerPolicy.RangerPolicyResource;
 import org.apache.ranger.plugin.model.RangerServiceResource;
-import org.apache.ranger.tagsync.source.atlasrest.RangerAtlasEntity;
 
 public class AtlasStormResourceMapper extends AtlasResourceMapper {
        public static final String ENTITY_TYPE_STORM_TOPOLOGY = 
"storm_topology";
        public static final String RANGER_TYPE_STORM_TOPOLOGY = "topology";
 
+       public static final String ENTITY_ATTRIBUTE_QUALIFIED_NAME = 
"qualifiedName";
+
        public static final String[] SUPPORTED_ENTITY_TYPES = { 
ENTITY_TYPE_STORM_TOPOLOGY };
 
        public AtlasStormResourceMapper() {
@@ -38,8 +40,8 @@ public class AtlasStormResourceMapper extends 
AtlasResourceMapper {
        }
 
        @Override
-    public RangerServiceResource buildResource(final RangerAtlasEntity entity) 
throws Exception {
-               String qualifiedName = 
(String)entity.getAttributes().get(AtlasResourceMapper.ENTITY_ATTRIBUTE_QUALIFIED_NAME);
+       public RangerServiceResource buildResource(final IReferenceableInstance 
entity) throws Exception {
+               String qualifiedName = getEntityAttribute(entity, 
ENTITY_ATTRIBUTE_QUALIFIED_NAME, String.class);
 
                String topology = 
getResourceNameFromQualifiedName(qualifiedName);
 
@@ -63,7 +65,7 @@ public class AtlasStormResourceMapper extends 
AtlasResourceMapper {
 
                elements.put(RANGER_TYPE_STORM_TOPOLOGY, new 
RangerPolicyResource(topology, isExcludes, isRecursive));
 
-               String entityGuid  = entity.getGuid();
+               String entityGuid  = entity.getId() != null ? 
entity.getId()._getId() : null;
                String serviceName = getRangerServiceName(clusterName);
 
                return new RangerServiceResource(entityGuid, serviceName, 
elements);

http://git-wip-us.apache.org/repos/asf/ranger/blob/1ab501c3/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasTagSource.java
----------------------------------------------------------------------
diff --git 
a/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasTagSource.java
 
b/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasTagSource.java
index 8c15ee5..c382db0 100644
--- 
a/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasTagSource.java
+++ 
b/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasTagSource.java
@@ -20,24 +20,24 @@
 package org.apache.ranger.tagsync.source.atlas;
 
 
+import org.apache.atlas.kafka.AtlasKafkaMessage;
 import org.apache.atlas.kafka.NotificationProvider;
-import org.apache.atlas.notification.NotificationConsumer;
-import org.apache.atlas.notification.NotificationInterface;
-import org.apache.atlas.v1.model.notification.EntityNotificationV1;
 import org.apache.commons.lang.StringUtils;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
-import org.apache.ranger.plugin.util.ServiceTags;
-import org.apache.ranger.tagsync.model.AbstractTagSource;
-import org.apache.atlas.kafka.AtlasKafkaMessage;
-import org.apache.kafka.common.TopicPartition;
-import org.apache.ranger.tagsync.source.atlasrest.RangerAtlasEntityWithTags;
 
+import org.apache.atlas.notification.NotificationConsumer;
+import org.apache.atlas.notification.NotificationInterface;
+import org.apache.atlas.notification.entity.EntityNotification;
+
+import org.apache.kafka.common.TopicPartition;
+import org.apache.ranger.tagsync.model.AbstractTagSource;
+import org.apache.ranger.plugin.util.ServiceTags;
 
 import java.io.IOException;
 import java.io.InputStream;
-import java.util.List;
 import java.util.Properties;
+import java.util.List;
 
 public class AtlasTagSource extends AbstractTagSource {
        private static final Log LOG = LogFactory.getLog(AtlasTagSource.class);
@@ -100,11 +100,10 @@ public class AtlasTagSource extends AbstractTagSource {
                }
 
                if (ret) {
-                       NotificationInterface notification = 
NotificationProvider.get();
-                       List<NotificationConsumer<EntityNotificationV1>> 
iterators = 
notification.createConsumers(NotificationInterface.NotificationType.ENTITIES, 
1);
-
-                       consumerTask = new ConsumerRunnable(iterators.get(0));
+            NotificationInterface notification = NotificationProvider.get();
+            List<NotificationConsumer<Object>> iterators = 
notification.createConsumers(NotificationInterface.NotificationType.ENTITIES, 
1);
 
+            consumerTask = new ConsumerRunnable(iterators.get(0));
                }
 
                if (LOG.isDebugEnabled()) {
@@ -138,59 +137,63 @@ public class AtlasTagSource extends AbstractTagSource {
                }
        }
 
-       private static String 
getPrintableEntityNotification(EntityNotificationV1 notification) {
+       private static String getPrintableEntityNotification(EntityNotification 
notification) {
                StringBuilder sb = new StringBuilder();
 
                sb.append("{ Notification-Type: 
").append(notification.getOperationType()).append(", ");
-        RangerAtlasEntityWithTags entityWithTags = new 
RangerAtlasEntityWithTags(notification);
-        sb.append(entityWithTags.toString());
-
+               AtlasEntityWithTraits entityWithTraits = new 
AtlasEntityWithTraits(notification.getEntity(), notification.getAllTraits());
+               sb.append(entityWithTraits.toString());
                sb.append("}");
                return sb.toString();
        }
 
        private class ConsumerRunnable implements Runnable {
 
-               private final NotificationConsumer<EntityNotificationV1> 
consumer;
+               private final NotificationConsumer<Object> consumer;
 
-               private 
ConsumerRunnable(NotificationConsumer<EntityNotificationV1> consumer) {
+               private ConsumerRunnable(NotificationConsumer<Object> consumer) 
{
                        this.consumer = consumer;
                }
 
-
                @Override
                public void run() {
                        if (LOG.isDebugEnabled()) {
                                LOG.debug("==> ConsumerRunnable.run()");
                        }
                        while (true) {
-                               try {
-                                       
List<AtlasKafkaMessage<EntityNotificationV1>> messages = 
consumer.receive(1000L);
-
-                                       for 
(AtlasKafkaMessage<EntityNotificationV1> message :  messages) {
-                                               EntityNotificationV1 
notification = message != null ? message.getMessage() : null;
-
-                                               if (notification != null) {
-                                                       if 
(LOG.isDebugEnabled()) {
-                                                               
LOG.debug("Notification=" + getPrintableEntityNotification(notification));
-                                                       }
-
-                                                       ServiceTags serviceTags 
= AtlasNotificationMapper.processEntityNotification(notification);
-                                                       if (serviceTags != 
null) {
-                                                               
updateSink(serviceTags);
-                                                       }
-
-                                                       TopicPartition 
partition = new TopicPartition("ATLAS_ENTITIES", message.getPartition());
-                                                       
consumer.commit(partition, message.getOffset());
-                                               } else {
-                                                       LOG.error("Null 
entityNotification received from Kafka!! Ignoring..");
-                                               }
-                                       }
-                               } catch (Exception exception) {
-                                       LOG.error("Caught exception..: ", 
exception);
-                                       return;
-                               }
-                       }
+                try {
+                    List<AtlasKafkaMessage<Object>> messages = 
consumer.receive(1000L);
+                    for (AtlasKafkaMessage<Object> message : messages) {
+                        Object kafkaMessage = message != null ? 
message.getMessage() : null;
+
+                        if (kafkaMessage != null) {
+                            EntityNotification notification = null;
+                            if (kafkaMessage instanceof EntityNotification) {
+                                notification = (EntityNotification) 
kafkaMessage;
+                            } else {
+                                LOG.warn("Received Kafka notification of 
unexpected type:[" + kafkaMessage.getClass().toString() + "], Ignoring...");
+                            }
+                            if (notification != null) {
+                                if (LOG.isDebugEnabled()) {
+                                    LOG.debug("Notification=" + 
getPrintableEntityNotification(notification));
+                                }
+
+                                ServiceTags serviceTags = 
AtlasNotificationMapper.processEntityNotification(notification);
+                                if (serviceTags != null) {
+                                    updateSink(serviceTags);
+                                }
+                            }
+                            TopicPartition partition = new 
TopicPartition("ATLAS_ENTITIES", message.getPartition());
+                            consumer.commit(partition, message.getOffset());
+                        } else {
+                            LOG.error("Null message received from Kafka!! 
Ignoring..");
+                        }
+                    }
+                } catch (Exception exception) {
+                    LOG.error("Caught exception..: ", exception);
+                    return;
+                }
+            }
                }
        }
 }

http://git-wip-us.apache.org/repos/asf/ranger/blob/1ab501c3/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlasrest/AtlasRESTTagSource.java
----------------------------------------------------------------------
diff --git 
a/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlasrest/AtlasRESTTagSource.java
 
b/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlasrest/AtlasRESTTagSource.java
index b715869..4e0ae90 100644
--- 
a/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlasrest/AtlasRESTTagSource.java
+++ 
b/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlasrest/AtlasRESTTagSource.java
@@ -20,67 +20,35 @@
 package org.apache.ranger.tagsync.source.atlasrest;
 
 import com.google.gson.Gson;
+
 import com.google.gson.GsonBuilder;
-import org.apache.atlas.AtlasClientV2;
-import org.apache.atlas.AtlasServiceException;
-import org.apache.atlas.exception.AtlasBaseException;
-import org.apache.atlas.model.SearchFilter;
-import org.apache.atlas.model.discovery.AtlasSearchResult;
-import org.apache.atlas.model.discovery.SearchParameters;
-import org.apache.atlas.model.instance.AtlasClassification;
-import org.apache.atlas.model.instance.AtlasEntity;
-import org.apache.atlas.model.instance.AtlasEntityHeader;
-import org.apache.atlas.model.typedef.AtlasBaseTypeDef;
-import org.apache.atlas.model.typedef.AtlasTypesDef;
-import org.apache.atlas.type.AtlasBuiltInTypes;
-import org.apache.atlas.type.AtlasClassificationType;
-import org.apache.atlas.type.AtlasStructType;
-import org.apache.atlas.type.AtlasType;
-import org.apache.atlas.type.AtlasTypeRegistry;
+
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.collections.MapUtils;
 import org.apache.commons.lang.StringUtils;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
-import org.apache.hadoop.security.UserGroupInformation;
-import org.apache.ranger.plugin.util.ServiceTags;
+import org.apache.ranger.plugin.util.RangerRESTClient;
 import org.apache.ranger.tagsync.model.AbstractTagSource;
+import org.apache.ranger.plugin.util.ServiceTags;
 import org.apache.ranger.tagsync.model.TagSink;
 import org.apache.ranger.tagsync.process.TagSyncConfig;
 import org.apache.ranger.tagsync.process.TagSynchronizer;
+import org.apache.ranger.tagsync.source.atlas.AtlasEntityWithTraits;
 import org.apache.ranger.tagsync.source.atlas.AtlasNotificationMapper;
 import org.apache.ranger.tagsync.source.atlas.AtlasResourceMapperUtil;
 
-import java.io.IOException;
-import java.text.DateFormat;
-import java.text.SimpleDateFormat;
-import java.util.ArrayList;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
-import java.util.Set;
-import java.util.TimeZone;
 
 public class AtlasRESTTagSource extends AbstractTagSource implements Runnable {
        private static final Log LOG = 
LogFactory.getLog(AtlasRESTTagSource.class);
 
-    private static final ThreadLocal<DateFormat> DATE_FORMATTER = new 
ThreadLocal<DateFormat>() {
-               @Override
-               protected DateFormat initialValue() {
-                       SimpleDateFormat dateFormat = new 
SimpleDateFormat(AtlasBaseTypeDef.SERIALIZED_DATE_FORMAT_STR);
-
-                       dateFormat.setTimeZone(TimeZone.getTimeZone("UTC"));
-
-                       return dateFormat;
-               }
-       };
-
        private long sleepTimeBetweenCycleInMillis;
-       private String[] restUrls         = null;
-       private boolean  isKerberized     = false;
-       private String[] userNamePassword = null;
+
+       private AtlasRESTUtil atlasRESTUtil = null;
 
        private Thread myThread = null;
 
@@ -127,26 +95,30 @@ public class AtlasRESTTagSource extends AbstractTagSource 
implements Runnable {
                boolean ret = 
AtlasResourceMapperUtil.initializeAtlasResourceMappers(properties);
 
                sleepTimeBetweenCycleInMillis = 
TagSyncConfig.getTagSourceAtlasDownloadIntervalInMillis(properties);
-               isKerberized = 
TagSyncConfig.getTagsyncKerberosIdentity(properties) != null;
+               final boolean isKerberized = 
TagSyncConfig.getTagsyncKerberosIdentity(properties) != null;
 
-               String restEndpoint       = 
TagSyncConfig.getAtlasRESTEndpoint(properties);
+               String restUrl       = 
TagSyncConfig.getAtlasRESTEndpoint(properties);
                String sslConfigFile = 
TagSyncConfig.getAtlasRESTSslConfigFile(properties);
-        this.userNamePassword = new String[] { 
TagSyncConfig.getAtlasRESTUserName(properties), 
TagSyncConfig.getAtlasRESTPassword(properties) };
+               String userName = 
TagSyncConfig.getAtlasRESTUserName(properties);
+               String password = 
TagSyncConfig.getAtlasRESTPassword(properties);
 
                if (LOG.isDebugEnabled()) {
-                       LOG.debug("restUrl=" + restEndpoint);
+                       LOG.debug("restUrl=" + restUrl);
                        LOG.debug("sslConfigFile=" + sslConfigFile);
-                       LOG.debug("userName=" + userNamePassword[0]);
+                       LOG.debug("userName=" + userName);
                        LOG.debug("kerberized=" + isKerberized);
                }
-        if (StringUtils.isNotEmpty(restEndpoint)) {
-            this.restUrls = restEndpoint.split(",");
-
-            for (int i = 0; i < restUrls.length; i++) {
-                if (!restUrls[i].endsWith("/")) {
-                    restUrls[i] += "/";
-                }
-            }
+
+               if (StringUtils.isNotEmpty(restUrl)) {
+                       if (!restUrl.endsWith("/")) {
+                               restUrl += "/";
+                       }
+                       RangerRESTClient atlasRESTClient = new 
RangerRESTClient(restUrl, sslConfigFile);
+
+                       if (!isKerberized) {
+                               atlasRESTClient.setBasicAuthInfo(userName, 
password);
+                       }
+                       atlasRESTUtil = new AtlasRESTUtil(atlasRESTClient, 
isKerberized);
                } else {
                        LOG.info("AtlasEndpoint not specified, Initial download 
of Atlas-entities cannot be done.");
                        ret = false;
@@ -202,15 +174,16 @@ public class AtlasRESTTagSource extends AbstractTagSource 
implements Runnable {
 
        public void synchUp() {
 
-               List<RangerAtlasEntityWithTags> rangerAtlasEntities = 
getAtlasActiveEntities();
+               List<AtlasEntityWithTraits> atlasEntities = 
atlasRESTUtil.getAtlasEntities();
 
-               if (CollectionUtils.isNotEmpty(rangerAtlasEntities)) {
+               if (CollectionUtils.isNotEmpty(atlasEntities)) {
                        if (LOG.isDebugEnabled()) {
-                               for (RangerAtlasEntityWithTags element : 
rangerAtlasEntities) {
+                               for (AtlasEntityWithTraits element : 
atlasEntities) {
                                        LOG.debug(element);
                                }
                        }
-                       Map<String, ServiceTags> serviceTagsMap = 
AtlasNotificationMapper.processAtlasEntities(rangerAtlasEntities);
+
+                       Map<String, ServiceTags> serviceTagsMap = 
AtlasNotificationMapper.processAtlasEntities(atlasEntities);
 
                        if (MapUtils.isNotEmpty(serviceTagsMap)) {
                                for (Map.Entry<String, ServiceTags> entry : 
serviceTagsMap.entrySet()) {
@@ -229,158 +202,5 @@ public class AtlasRESTTagSource extends AbstractTagSource 
implements Runnable {
 
        }
 
-       private List<RangerAtlasEntityWithTags> getAtlasActiveEntities() {
-               if (LOG.isDebugEnabled()) {
-                       LOG.debug("==> getAtlasActiveEntities()");
-               }
-               List<RangerAtlasEntityWithTags> ret = null;
-
-               SearchParameters searchParams = new SearchParameters();
-               AtlasTypeRegistry typeRegistry = new AtlasTypeRegistry();
-               AtlasTypeRegistry.AtlasTransientTypeRegistry tty = null;
-               AtlasSearchResult searchResult = null;
-
-               searchParams.setClassification("*");
-               searchParams.setIncludeClassificationAttributes(true);
-               searchParams.setOffset(0);
-               searchParams.setLimit(Integer.MAX_VALUE);
-
-               boolean commitUpdates = false;
-               try {
-                       AtlasClientV2 atlasClient = getAtlasClient();
-                       searchResult = atlasClient.facetedSearch(searchParams);
-                       AtlasTypesDef typesDef = atlasClient.getAllTypeDefs(new 
SearchFilter());
-                       tty = typeRegistry.lockTypeRegistryForUpdate();
-                       tty.addTypes(typesDef);
-                       commitUpdates = true;
-               } catch (AtlasServiceException | AtlasBaseException | 
IOException excp) {
-                       LOG.error("failed to download tags from Atlas", excp);
-               } catch (Exception unexpectedException) {
-                       LOG.error("Failed to download tags from Atlas due to 
unexpected exception", unexpectedException);
-               } finally {
-                       if (tty != null) {
-                               typeRegistry.releaseTypeRegistryForUpdate(tty, 
commitUpdates);
-                       }
-               }
-
-               if (commitUpdates && searchResult != null) {
-                       if (LOG.isDebugEnabled()) {
-                               LOG.debug(AtlasType.toJson(searchResult));
-                       }
-                       ret = new ArrayList<>();
-                       List<AtlasEntityHeader> entityHeaders = 
searchResult.getEntities();
-                       if (CollectionUtils.isNotEmpty(entityHeaders)) {
-                               for (AtlasEntityHeader header : entityHeaders) {
-                                       if 
(!header.getStatus().equals(AtlasEntity.Status.ACTIVE)) {
-                                               if (LOG.isDebugEnabled()) {
-                                                       LOG.debug("Skipping 
entity because it is not ACTIVE, header:[" + header + "]");
-                                               }
-                                               continue;
-                                       }
-
-                                       String typeName = header.getTypeName();
-                                       if 
(!AtlasResourceMapperUtil.isEntityTypeHandled(typeName)) {
-                                               if (LOG.isDebugEnabled()) {
-                                                       LOG.debug("Not fetching 
Atlas entities of type:[" + typeName + "]");
-                                               }
-                                               continue;
-                                       }
-
-                                       Map<String, Map<String, String>> 
allTagsForEntity = new HashMap<>();
-
-                                       for (AtlasClassification classification 
: header.getClassifications()) {
-                                               Map<String, Map<String, 
String>> tags = resolveTag(typeRegistry, classification.getTypeName(), 
classification.getAttributes());
-                                               if (tags != null) {
-                                                       
allTagsForEntity.putAll(tags);
-                                               }
-                                       }
-
-                                       if 
(MapUtils.isNotEmpty(allTagsForEntity)) {
-
-                                               RangerAtlasEntity entity = new 
RangerAtlasEntity(typeName, header.getGuid(), header.getAttributes());
-                                               RangerAtlasEntityWithTags 
entityWithTags = new RangerAtlasEntityWithTags(entity, allTagsForEntity, 
typeRegistry);
-                                               ret.add(entityWithTags);
-                                       }
-                               }
-                       }
-               }
-
-               if (LOG.isDebugEnabled()) {
-                       LOG.debug("<== getAtlasActiveEntities()");
-               }
-
-               return ret;
-       }
-
-    /*
-     * Returns a map of <tag-name, List<attributeName, [attributeValue, 
attributeType]>>
-     */
-       private Map<String, Map<String, String>> resolveTag(AtlasTypeRegistry 
typeRegistry, String typeName, Map<String, Object> attributes) {
-               Map<String, Map<String, String>> ret = new HashMap<>();
-
-               try {
-                       AtlasClassificationType classificationType = 
typeRegistry.getClassificationTypeByName(typeName);
-                       if (classificationType != null) {
-                               Map<String, String> allAttributes = new 
HashMap<>();
-                               if (MapUtils.isNotEmpty(attributes) && 
MapUtils.isNotEmpty(classificationType.getAllAttributes())) {
-                                       for (Map.Entry<String, Object> 
attribute : attributes.entrySet()) {
-                                               String name = 
attribute.getKey();
-                                               Object value = 
attribute.getValue();
-                                               if (value != null) {
-                                                       String stringValue = 
value.toString();
-                                                       
AtlasStructType.AtlasAttribute atlasAttribute = 
classificationType.getAttribute(name);
-                                                       if (atlasAttribute != 
null) {
-                                                               if (value 
instanceof Number) {
-                                                                       if 
(atlasAttribute.getAttributeType() instanceof AtlasBuiltInTypes.AtlasDateType) {
-                                                                               
stringValue = DATE_FORMATTER.get().format(value);
-                                                                       }
-                                                               }
-                                                               
allAttributes.put(name, stringValue);
-                                                       }
-                                               }
-                                       }
-                               }
-                               // Put most derived classificationType with all 
attributes
-                               ret.put(typeName, allAttributes);
-
-                               // Find base classification types
-                               Set<String> superTypeNames = 
classificationType.getAllSuperTypes();
-                               for (String superTypeName : superTypeNames) {
-                                       AtlasClassificationType superType = 
typeRegistry.getClassificationTypeByName(superTypeName);
-                                       if (superType != null) {
-                                               Map<String, String> 
attributeMap = new HashMap<>();
-                                               if 
(MapUtils.isNotEmpty(attributes) && 
MapUtils.isNotEmpty(superType.getAllAttributes())) {
-                                                       for (String name : 
superType.getAllAttributes().keySet()) {
-                                                               String 
stringValue = allAttributes.get(name);
-                                                               if (stringValue 
!= null) {
-                                                                       
attributeMap.put(name, stringValue);
-                                                               }
-                                                       }
-                                               }
-                                               ret.put(superTypeName, 
attributeMap);
-                                       }
-                               }
-                       }
-               } catch (Exception exception) {
-                       LOG.error("Error in resolving tags for type:[" + 
typeName + "]", exception);
-               }
-               return ret;
-       }
-
-       private AtlasClientV2 getAtlasClient() throws IOException {
-               final AtlasClientV2 ret;
-
-               if (isKerberized) {
-                       UserGroupInformation ugi = 
UserGroupInformation.getLoginUser();
-
-                       ugi.checkTGTAndReloginFromKeytab();
-
-                       ret = new AtlasClientV2(ugi, ugi.getShortUserName(), 
restUrls);
-               } else {
-                       ret = new AtlasClientV2(restUrls, userNamePassword);
-               }
-
-               return ret;
-       }
 }
 

http://git-wip-us.apache.org/repos/asf/ranger/blob/1ab501c3/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlasrest/AtlasRESTUtil.java
----------------------------------------------------------------------
diff --git 
a/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlasrest/AtlasRESTUtil.java
 
b/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlasrest/AtlasRESTUtil.java
new file mode 100644
index 0000000..00a101e
--- /dev/null
+++ 
b/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlasrest/AtlasRESTUtil.java
@@ -0,0 +1,325 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.ranger.tagsync.source.atlasrest;
+
+import com.google.gson.Gson;
+import com.sun.jersey.api.client.ClientResponse;
+import com.sun.jersey.api.client.WebResource;
+import org.apache.atlas.typesystem.IReferenceableInstance;
+import org.apache.atlas.typesystem.IStruct;
+import org.apache.atlas.typesystem.Struct;
+import org.apache.atlas.typesystem.json.InstanceSerialization;
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.collections.MapUtils;
+import org.apache.commons.lang.StringUtils;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.log4j.Logger;
+import org.apache.ranger.admin.client.datatype.RESTResponse;
+import org.apache.ranger.plugin.util.RangerRESTClient;
+import org.apache.ranger.tagsync.source.atlas.AtlasEntityWithTraits;
+import org.apache.ranger.tagsync.source.atlas.AtlasResourceMapperUtil;
+
+import java.io.IOException;
+import java.security.PrivilegedAction;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+
+@SuppressWarnings("unchecked")
+public class AtlasRESTUtil {
+       private static final Logger LOG = Logger.getLogger(AtlasRESTUtil.class);
+
+       private static final String REST_MIME_TYPE_JSON = "application/json";
+       private static final String API_ATLAS_TYPES    = "api/atlas/types";
+       private static final String API_ATLAS_ENTITIES = 
"api/atlas/entities?type=";
+       private static final String API_ATLAS_ENTITY   = "api/atlas/entities/";
+       private static final String API_ATLAS_TYPE     = "api/atlas/types/";
+
+       private static final String RESULTS_ATTRIBUTE               = "results";
+       private static final String DEFINITION_ATTRIBUTE            = 
"definition";
+       private static final String VALUES_ATTRIBUTE                = "values";
+       private static final String TRAITS_ATTRIBUTE                = "traits";
+       private static final String TYPE_NAME_ATTRIBUTE             = 
"typeName";
+       private static final String TRAIT_TYPES_ATTRIBUTE           = 
"traitTypes";
+       private static final String SUPER_TYPES_ATTRIBUTE           = 
"superTypes";
+       private static final String ATTRIBUTE_DEFINITIONS_ATTRIBUTE = 
"attributeDefinitions";
+       private static final String NAME_ATTRIBUTE                  = "name";
+
+       private final Gson gson = new Gson();
+
+       private final RangerRESTClient atlasRESTClient;
+
+       private final boolean isKerberized;
+
+       public AtlasRESTUtil(RangerRESTClient atlasRESTClient, boolean 
isKerberized) {
+               if (LOG.isDebugEnabled()) {
+                       LOG.debug("==> AtlasRESTUtil()");
+               }
+
+               this.atlasRESTClient = atlasRESTClient;
+
+               this.isKerberized = isKerberized;
+
+               if (LOG.isDebugEnabled()) {
+                       LOG.debug("<== AtlasRESTUtil()");
+               }
+       }
+
+       public List<AtlasEntityWithTraits> getAtlasEntities() {
+               if (LOG.isDebugEnabled()) {
+                       LOG.debug("==> getAtlasEntities()");
+               }
+
+               List<AtlasEntityWithTraits> ret = new 
ArrayList<AtlasEntityWithTraits>();
+
+               Map<String, Object> typesResponse = atlasAPI(API_ATLAS_TYPES);
+
+               List<String> types = getAttribute(typesResponse, 
RESULTS_ATTRIBUTE, List.class);
+
+               if (CollectionUtils.isNotEmpty(types)) {
+
+                       for (String type : types) {
+
+                               if 
(!AtlasResourceMapperUtil.isEntityTypeHandled(type)) {
+                                       if (LOG.isDebugEnabled()) {
+                                               LOG.debug("Not fetching Atlas 
entities of type: " + type);
+                                       }
+                                       continue;
+                               }
+
+                               Map<String, Object> entitiesResponse = 
atlasAPI(API_ATLAS_ENTITIES + type);
+
+                               List<String> guids = 
getAttribute(entitiesResponse, RESULTS_ATTRIBUTE, List.class);
+
+                               if (CollectionUtils.isEmpty(guids)) {
+                                       if (LOG.isDebugEnabled()) {
+                                               LOG.debug("No Atlas entities 
for type: " + type);
+                                       }
+                                       continue;
+                               }
+
+                               for (String guid : guids) {
+
+                                       Map<String, Object> entityResponse = 
atlasAPI(API_ATLAS_ENTITY + guid);
+
+                                       Map<String, Object> definition = 
getAttribute(entityResponse, DEFINITION_ATTRIBUTE, Map.class);
+
+                                       Map<String, Object> traitsAttribute = 
getAttribute(definition, TRAITS_ATTRIBUTE, Map.class);
+
+                                       List<IStruct> allTraits = new 
LinkedList<>();
+
+                                       if 
(MapUtils.isNotEmpty(traitsAttribute)) {
+
+                                               for (Map.Entry<String, Object> 
entry : traitsAttribute.entrySet()) {
+
+                                                       Map<String, Object> 
trait = (Map<String, Object>) entry.getValue();
+
+                                                       Map<String, Object> 
traitValues = getAttribute(trait, VALUES_ATTRIBUTE, Map.class);
+                                                       String traitTypeName = 
getAttribute(trait, TYPE_NAME_ATTRIBUTE, String.class);
+
+                                                       if 
(StringUtils.isEmpty(traitTypeName)) {
+                                                               continue;
+                                                       }
+
+                                                       List<IStruct> 
superTypes = getTraitSuperTypes(getTraitType(traitTypeName), traitValues);
+
+                                                       Struct trait1 = new 
Struct(traitTypeName, traitValues);
+
+                                                       allTraits.add(trait1);
+                                                       
allTraits.addAll(superTypes);
+                                               }
+                                       }
+
+                                       IReferenceableInstance entity = 
InstanceSerialization.fromJsonReferenceable(gson.toJson(definition), true);
+
+                                       if (entity != null) {
+                                               AtlasEntityWithTraits 
atlasEntity = new AtlasEntityWithTraits(entity, allTraits);
+                                               ret.add(atlasEntity);
+                                       } else {
+                                               if (LOG.isInfoEnabled()) {
+                                                       LOG.info("Could not 
create Atlas entity from its definition, type=" + type + ", guid=" + guid);
+                                               }
+                                       }
+
+                               }
+
+                       }
+                       if (LOG.isDebugEnabled()) {
+                               LOG.debug("<== getAtlasEntities()");
+                       }
+               }
+
+               return ret;
+       }
+
+       private Map<String, Object> getTraitType(String traitName) {
+               if (LOG.isDebugEnabled()) {
+                       LOG.debug("==> getTraitType(" + traitName + ")");
+               }
+               Map<String, Object> ret = null;
+
+               Map<String, Object> typeResponse = atlasAPI(API_ATLAS_TYPE + 
traitName);
+
+               Map<String, Object> definition = getAttribute(typeResponse, 
DEFINITION_ATTRIBUTE, Map.class);
+
+               List traitTypes = getAttribute(definition, 
TRAIT_TYPES_ATTRIBUTE, List.class);
+
+               if (CollectionUtils.isNotEmpty(traitTypes)) {
+                       ret = (Map<String, Object>) traitTypes.get(0);
+               }
+
+               if (LOG.isDebugEnabled()) {
+                       LOG.debug("<== getTraitType(" + traitName + ")");
+               }
+               return ret;
+       }
+
+       private List<IStruct> getTraitSuperTypes(Map<String, Object> traitType, 
Map<String, Object> values) {
+
+               if (LOG.isDebugEnabled()) {
+                       LOG.debug("==> getTraitSuperTypes()");
+               }
+               List<IStruct> ret = new LinkedList<>();
+
+               if (traitType != null) {
+
+                       List<String> superTypeNames = getAttribute(traitType, 
SUPER_TYPES_ATTRIBUTE, List.class);
+
+                       if (CollectionUtils.isNotEmpty(superTypeNames)) {
+                               for (String superTypeName : superTypeNames) {
+
+                                       Map<String, Object> superTraitType = 
getTraitType(superTypeName);
+
+                                       if (superTraitType != null) {
+                                               List<Map<String, Object>> 
attributeDefinitions = (List) 
superTraitType.get(ATTRIBUTE_DEFINITIONS_ATTRIBUTE);
+
+                                               Map<String, Object> 
superTypeValues = new HashMap<>();
+                                               for (Map<String, Object> 
attributeDefinition : attributeDefinitions) {
+
+                                                       String attributeName = 
attributeDefinition.get(NAME_ATTRIBUTE).toString();
+                                                       if 
(values.containsKey(attributeName)) {
+                                                               
superTypeValues.put(attributeName, values.get(attributeName));
+                                                       }
+                                               }
+
+                                               List<IStruct> superTraits = 
getTraitSuperTypes(getTraitType(superTypeName), values);
+
+                                               Struct superTrait = new 
Struct(superTypeName, superTypeValues);
+
+                                               ret.add(superTrait);
+                                               ret.addAll(superTraits);
+                                       }
+                               }
+                       }
+               }
+               if (LOG.isDebugEnabled()) {
+                       LOG.debug("<== getTraitSuperTypes()");
+               }
+               return ret;
+       }
+
+       private Map<String, Object> atlasAPI(final String endpoint) {
+
+               if (LOG.isDebugEnabled()) {
+                       LOG.debug("==> atlasAPI(" + endpoint + ")");
+               }
+               Map<String, Object> ret = new HashMap<String, Object>();
+
+               try {
+                       UserGroupInformation userGroupInformation = null;
+                       if (isKerberized) {
+                               userGroupInformation = 
UserGroupInformation.getLoginUser();
+
+                               try {
+                                       
userGroupInformation.checkTGTAndReloginFromKeytab();
+                               } catch (IOException ioe) {
+                                       LOG.error("Error renewing TGT and 
relogin", ioe);
+                                       userGroupInformation = null;
+                               }
+                       }
+                       if (userGroupInformation != null) {
+                               LOG.debug("Using kerberos authentication");
+                               if(LOG.isDebugEnabled()) {
+                                       LOG.debug("Using Principal = "+ 
userGroupInformation.getUserName());
+                               }
+                               ret = userGroupInformation.doAs(new 
PrivilegedAction<Map<String, Object>>() {
+                                       @Override
+                                       public Map<String, Object> run() {
+                                               try{
+                                                       return 
executeAtlasAPI(endpoint);
+                                               }catch (Exception e) {
+                                                       LOG.error("Atlas API 
failed with message : ", e);
+                                               }
+                                               return null;
+                                       }
+                               });
+                       } else {
+                               LOG.debug("Using basic authentication");
+                               ret = executeAtlasAPI(endpoint);
+                       }
+               } catch (Exception exception) {
+                       LOG.error("Exception when fetching Atlas objects.", 
exception);
+                       ret = null;
+               }
+
+               if (LOG.isDebugEnabled()) {
+                       LOG.debug("<== atlasAPI(" + endpoint + ")");
+               }
+               return ret;
+       }
+
+       private Map<String, Object> executeAtlasAPI(final String endpoint) {
+
+               if (LOG.isDebugEnabled()) {
+                       LOG.debug("==> executeAtlasAPI(" + endpoint + ")");
+               }
+
+               Map<String, Object> ret = new HashMap<String, Object>();
+
+               try {
+                       final WebResource webResource = 
atlasRESTClient.getResource(endpoint);
+
+                       ClientResponse response = 
webResource.accept(REST_MIME_TYPE_JSON).type(REST_MIME_TYPE_JSON).get(ClientResponse.class);
+
+                       if (response != null && response.getStatus() == 200) {
+                               ret = response.getEntity(ret.getClass());
+                       } else {
+                               RESTResponse resp = 
RESTResponse.fromClientResponse(response);
+                               LOG.error("Error getting atlas data request=" + 
webResource.toString()
+                                               + ", response=" + 
resp.toString());
+                       }
+               } catch (Exception exception) {
+                       LOG.error("Exception when fetching Atlas objects.", 
exception);
+                       ret = null;
+               }
+
+               if (LOG.isDebugEnabled()) {
+                       LOG.debug("<== executeAtlasAPI(" + endpoint + ")");
+               }
+
+               return ret;
+       }
+
+       private <T> T getAttribute(Map<String, Object> map, String name, 
Class<T> type) {
+               return MapUtils.isNotEmpty(map) ? type.cast(map.get(name)) : 
null;
+       }
+
+}

Reply via email to