You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@atlas.apache.org by ni...@apache.org on 2019/03/05 14:36:25 UTC
[atlas] branch branch-1.0 updated (ea008cb -> 36d8014)
This is an automated email from the ASF dual-hosted git repository.
nixon pushed a change to branch branch-1.0
in repository https://gitbox.apache.org/repos/asf/atlas.git.
from ea008cb ATLAS-3065: added type-patch to remove legacy attributes
new b3df689 ATLAS-3020: Audit APIs for classification updates.
new 752587a ATLAS-3029: Audit APIs for classification updates.
new 6f3aa81 ATLAS-3029, ATLAS-3020: Audit APIs for classification updates. Part-2.
new bf59b4f ATLAS-3015: Classification Updater tool.
new 3a8cda4 ATLAS-3046: Classification Updater tool. Unique name used.
new c27aeef ATLAS-3046: Part 3. Updated case for handling deleted classifications.
new 36d8014 ATLAS-3066 : UI : Fix various table layouts & improvements.
The 7 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../main/java/org/apache/atlas/AtlasClientV2.java | 21 +-
dashboardv2/public/css/scss/common.scss | 2 +-
dashboardv2/public/js/utils/CommonViewFunction.js | 10 +-
.../public/js/views/audit/AuditTableLayoutView.js | 2 +-
.../js/views/detail_page/DetailPageLayoutView.js | 2 +-
.../instance/AtlasEntityHeaders.java} | 25 +-
.../audit/CassandraBasedAuditRepository.java | 7 +
.../repository/audit/EntityAuditRepository.java | 9 +
.../audit/HBaseBasedAuditRepository.java | 52 +-
.../audit/InMemoryEntityAuditRepository.java | 17 +
.../audit/NoopEntityAuditRepository.java | 7 +
.../repository/store/graph/AtlasEntityStore.java | 3 +
.../store/graph/v2/AtlasEntityStoreV2.java | 7 +
.../store/graph/v2/ClassificationAssociator.java | 318 ++++++++++++
.../store/graph/v2/EntityGraphRetriever.java | 2 +-
.../graph/v2/ClassificationAssociatorTest.java | 240 +++++++++
.../col-entity-None.json | 10 +
.../col-entity-PII-FIN_PII.json | 32 ++
.../classification-association/col-entity-PII.json | 22 +
.../col-entity-T1-prop-Tn-No-Guid.json | 31 ++
.../col-entity-T1-prop-Tn.json | 34 ++
.../classification-association/header-FIN_PII.json | 32 ++
.../classification-association/header-None.json | 21 +
.../header-PII-VENDOR_PII.json | 42 ++
.../classification-association/header-PII.json | 32 ++
.../header-Tx-prop-T1-No-Guid.json | 39 ++
.../header-Tx-prop-T1.json | 42 ++
.../json/classification-association/header-Tx.json | 26 +
.../classification-association/header-empty.json | 3 +
.../classification-updater}/pom.xml | 81 ++-
.../org/apache/atlas/tools/BulkFetchAndUpdate.java | 549 +++++++++++++++++++++
.../src/main/resources/atlas-log4j.xml | 23 +-
.../src/main/resources/update-classifications.sh | 90 ++++
.../java/org/apache/atlas/web/rest/EntityREST.java | 59 ++-
34 files changed, 1802 insertions(+), 90 deletions(-)
copy intg/src/main/java/org/apache/atlas/{v1/model/typedef/TraitTypeDefinition.java => model/instance/AtlasEntityHeaders.java} (72%)
create mode 100644 repository/src/main/java/org/apache/atlas/repository/store/graph/v2/ClassificationAssociator.java
create mode 100644 repository/src/test/java/org/apache/atlas/repository/store/graph/v2/ClassificationAssociatorTest.java
create mode 100644 repository/src/test/resources/json/classification-association/col-entity-None.json
create mode 100644 repository/src/test/resources/json/classification-association/col-entity-PII-FIN_PII.json
create mode 100644 repository/src/test/resources/json/classification-association/col-entity-PII.json
create mode 100644 repository/src/test/resources/json/classification-association/col-entity-T1-prop-Tn-No-Guid.json
create mode 100644 repository/src/test/resources/json/classification-association/col-entity-T1-prop-Tn.json
create mode 100644 repository/src/test/resources/json/classification-association/header-FIN_PII.json
create mode 100644 repository/src/test/resources/json/classification-association/header-None.json
create mode 100644 repository/src/test/resources/json/classification-association/header-PII-VENDOR_PII.json
create mode 100644 repository/src/test/resources/json/classification-association/header-PII.json
create mode 100644 repository/src/test/resources/json/classification-association/header-Tx-prop-T1-No-Guid.json
create mode 100644 repository/src/test/resources/json/classification-association/header-Tx-prop-T1.json
create mode 100644 repository/src/test/resources/json/classification-association/header-Tx.json
create mode 100644 repository/src/test/resources/json/classification-association/header-empty.json
copy {addons/hdfs-model => tools/classification-updater}/pom.xml (63%)
create mode 100644 tools/classification-updater/src/main/java/org/apache/atlas/tools/BulkFetchAndUpdate.java
copy addons/hive-bridge/src/main/resources/atlas-hive-import-log4j.xml => tools/classification-updater/src/main/resources/atlas-log4j.xml (64%)
create mode 100644 tools/classification-updater/src/main/resources/update-classifications.sh
[atlas] 05/07: ATLAS-3046: Classification Updater tool. Unique name
used.
Posted by ni...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
nixon pushed a commit to branch branch-1.0
in repository https://gitbox.apache.org/repos/asf/atlas.git
commit 3a8cda4d2da7b1bd08c8233cc9d7c5076314dfc9
Author: Ashutosh Mestry <am...@hortonworks.com>
AuthorDate: Thu Feb 7 22:17:18 2019 -0800
ATLAS-3046: Classification Updater tool. Unique name used.
---
.../org/apache/atlas/tools/BulkFetchAndUpdate.java | 99 ++++++++++++++++------
1 file changed, 71 insertions(+), 28 deletions(-)
diff --git a/tools/classification-updater/src/main/java/org/apache/atlas/tools/BulkFetchAndUpdate.java b/tools/classification-updater/src/main/java/org/apache/atlas/tools/BulkFetchAndUpdate.java
index 1e0b66d..19f8325 100644
--- a/tools/classification-updater/src/main/java/org/apache/atlas/tools/BulkFetchAndUpdate.java
+++ b/tools/classification-updater/src/main/java/org/apache/atlas/tools/BulkFetchAndUpdate.java
@@ -28,6 +28,8 @@ import org.apache.atlas.model.instance.AtlasClassification;
import org.apache.atlas.model.instance.AtlasEntityHeader;
import org.apache.atlas.model.instance.AtlasEntityHeaders;
import org.apache.atlas.model.typedef.AtlasClassificationDef;
+import org.apache.atlas.model.typedef.AtlasEntityDef;
+import org.apache.atlas.model.typedef.AtlasStructDef;
import org.apache.atlas.model.typedef.AtlasTypesDef;
import org.apache.atlas.type.AtlasType;
import org.apache.atlas.utils.AtlasJson;
@@ -300,6 +302,7 @@ public class BulkFetchAndUpdate {
private static final String ATTR_NAME_QUALIFIED_NAME = "qualifiedName";
private AtlasClientV2 atlasClientV2;
+ private Map<String, String> typeNameUniqueAttributeNameMap = new HashMap<>();
public Preparer(AtlasClientV2 atlasClientV2) {
this.atlasClientV2 = atlasClientV2;
@@ -322,7 +325,7 @@ public class BulkFetchAndUpdate {
try {
classificationDef.setGuid(null);
String json = AtlasType.toJson(classificationDef);
- fileWriter.write(json + "\n");
+ fileWriter.write(json);
} catch (Exception e) {
LOG.error("Error writing classifications: {}", e);
displayCrLf("Error writing classifications.");
@@ -348,11 +351,11 @@ public class BulkFetchAndUpdate {
}
try {
- AtlasEntityHeaders response = atlasClientV2.getEntityHeaders(fromTimestamp);
- int guidHeaderMapSize = response.getGuidHeaderMap().size();
+ AtlasEntityHeaders entityHeaders = atlasClientV2.getEntityHeaders(fromTimestamp);
+ int guidHeaderMapSize = entityHeaders.getGuidHeaderMap().size();
try {
displayCrLf("Read entities: " + guidHeaderMapSize);
- AtlasEntityHeaders updatedHeaders = removeEntityGuids(response);
+ AtlasEntityHeaders updatedHeaders = removeEntityGuids(entityHeaders);
fileWriter.write(AtlasType.toJson(updatedHeaders));
displayCrLf("Writing entities: " + updatedHeaders.getGuidHeaderMap().size());
@@ -368,12 +371,17 @@ public class BulkFetchAndUpdate {
}
private AtlasEntityHeaders removeEntityGuids(AtlasEntityHeaders headers) {
- Map<String, AtlasEntityHeader> qualifiedNameHeaderMap = new HashMap<>();
+ Map<String, AtlasEntityHeader> uniqueNameEntityHeaderMap = new HashMap<>();
for (AtlasEntityHeader header : headers.getGuidHeaderMap().values()) {
- String qualifiedName = getQualifiedName(header);
- displayCrLf("Processing: " + qualifiedName);
+ String uniqueName = getUniqueName(header);
+ if (StringUtils.isEmpty(uniqueName)) {
+ displayCrLf("UniqueName is empty. Ignoring: " + header.getGuid());
+ LOG.warn("UniqueName is empty. Ignoring: {}", AtlasJson.toJson(header));
+ continue;
+ }
+ displayCrLf("Processing: " + uniqueName);
if (header.getStatus() == DELETED) {
continue;
}
@@ -383,37 +391,72 @@ public class BulkFetchAndUpdate {
continue;
}
- boolean keyFound = qualifiedNameHeaderMap.containsKey(qualifiedName);
+ String key = String.format("%s:%s", header.getTypeName(), uniqueName);
+ boolean keyFound = uniqueNameEntityHeaderMap.containsKey(key);
if (!keyFound) {
- qualifiedNameHeaderMap.put(qualifiedName, header);
+ uniqueNameEntityHeaderMap.put(key, header);
}
- AtlasEntityHeader currentHeader = qualifiedNameHeaderMap.get(qualifiedName);
- for (AtlasClassification c : header.getClassifications()) {
- c.setEntityGuid(null);
-
- if (keyFound) {
- boolean found =
- currentHeader.getClassifications().stream().anyMatch(ox -> ox.getTypeName().equals(c.getTypeName()));
- if (!found) {
- currentHeader.getClassifications().add(c);
- } else {
- displayCrLf("Ignoring: " + c.toString());
- LOG.warn("Ignoring: {}", AtlasJson.toJson(c));
- }
+ updateClassificationsForHeader(header, uniqueNameEntityHeaderMap.get(key), keyFound);
+ displayCrLf("Processing: " + uniqueName);
+ }
+
+ displayCrLf("Processed: " + uniqueNameEntityHeaderMap.size());
+ headers.setGuidHeaderMap(uniqueNameEntityHeaderMap);
+ return headers;
+ }
+
+ private void updateClassificationsForHeader(AtlasEntityHeader header, AtlasEntityHeader currentHeader, boolean keyFound) {
+ for (AtlasClassification c : header.getClassifications()) {
+ c.setEntityGuid(null);
+
+ if (keyFound) {
+ boolean found =
+ currentHeader.getClassifications().stream().anyMatch(ox -> ox.getTypeName().equals(c.getTypeName()));
+ if (!found) {
+ currentHeader.getClassifications().add(c);
+ } else {
+ displayCrLf("Ignoring: " + c.toString());
+ LOG.warn("Ignoring: {}", AtlasJson.toJson(c));
}
}
+ }
+ }
- displayCrLf("Processing: " + qualifiedName);
+ private String getUniqueName(AtlasEntityHeader header) {
+ String uniqueAttributeName = ATTR_NAME_QUALIFIED_NAME;
+ if (!header.getAttributes().containsKey(ATTR_NAME_QUALIFIED_NAME)) {
+ uniqueAttributeName = getUniqueAttribute(header.getTypeName());
}
- displayCrLf("Processed: " + qualifiedNameHeaderMap.size());
- headers.setGuidHeaderMap(qualifiedNameHeaderMap);
- return headers;
+ Object attrValue = header.getAttribute(uniqueAttributeName);
+ if (attrValue == null) {
+ LOG.warn("Unique Attribute Value: empty: {}", AtlasJson.toJson(header));
+ return StringUtils.EMPTY;
+ }
+
+ return attrValue.toString();
}
- private String getQualifiedName(AtlasEntityHeader header) {
- return (String) header.getAttribute(ATTR_NAME_QUALIFIED_NAME);
+ private String getUniqueAttribute(String typeName) {
+ try {
+ if (typeNameUniqueAttributeNameMap.containsKey(typeName)) {
+ return typeNameUniqueAttributeNameMap.get(typeName);
+ }
+
+ AtlasEntityDef entityDef = atlasClientV2.getEntityDefByName(typeName);
+ for (AtlasStructDef.AtlasAttributeDef ad : entityDef.getAttributeDefs()) {
+ if (ad.getIsUnique()) {
+ typeNameUniqueAttributeNameMap.put(typeName, ad.getName());
+ return ad.getName();
+ }
+ }
+ } catch (AtlasServiceException e) {
+ LOG.error("Error fetching type: {}", typeName, e);
+ return null;
+ }
+
+ return null;
}
private List<AtlasClassificationDef> getAllClassificationsDefs() throws Exception {
[atlas] 03/07: ATLAS-3029,
ATLAS-3020: Audit APIs for classification updates. Part-2.
Posted by ni...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
nixon pushed a commit to branch branch-1.0
in repository https://gitbox.apache.org/repos/asf/atlas.git
commit 6f3aa81452000e432a1b9a4b4abbae88cd172f63
Author: Ashutosh Mestry <am...@hortonworks.com>
AuthorDate: Mon Jan 21 22:31:13 2019 -0800
ATLAS-3029, ATLAS-3020: Audit APIs for classification updates. Part-2.
---
.../store/graph/v2/ClassificationAssociator.java | 4 ++-
.../graph/v2/ClassificationAssociatorTest.java | 5 +++
.../col-entity-T1-prop-Tn-No-Guid.json | 31 +++++++++++++++++
.../header-Tx-prop-T1-No-Guid.json | 39 ++++++++++++++++++++++
4 files changed, 78 insertions(+), 1 deletion(-)
diff --git a/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/ClassificationAssociator.java b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/ClassificationAssociator.java
index 057598c..628c1cc 100644
--- a/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/ClassificationAssociator.java
+++ b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/ClassificationAssociator.java
@@ -310,7 +310,9 @@ public class ClassificationAssociator {
return list;
}
- return list.stream().filter(x -> x.getEntityGuid().equals(guid)).collect(Collectors.toList());
+ return list.stream().filter(x -> x != null &&
+ (StringUtils.isEmpty(guid) || StringUtils.isEmpty(x.getEntityGuid()))
+ || x.getEntityGuid().equals(guid)).collect(Collectors.toList());
}
}
}
diff --git a/repository/src/test/java/org/apache/atlas/repository/store/graph/v2/ClassificationAssociatorTest.java b/repository/src/test/java/org/apache/atlas/repository/store/graph/v2/ClassificationAssociatorTest.java
index ab5bb2b..16074b6 100644
--- a/repository/src/test/java/org/apache/atlas/repository/store/graph/v2/ClassificationAssociatorTest.java
+++ b/repository/src/test/java/org/apache/atlas/repository/store/graph/v2/ClassificationAssociatorTest.java
@@ -163,6 +163,7 @@ public class ClassificationAssociatorTest {
@Test
public void updaterTests() throws IOException {
+ updaterAssert("header-None", "col-entity-None");
updaterAssert("header-PII", "col-entity-None", PROCESS_ADD + ":PII");
updaterAssert("header-PII", "col-entity-PII", new String[]{PROCESS_UPDATE + ":PII"});
updaterAssert("header-None", "col-entity-PII", new String[]{PROCESS_DELETE + ":PII"});
@@ -181,6 +182,10 @@ public class ClassificationAssociatorTest {
updaterAssert("header-Tx-prop-T1", "col-entity-T1-prop-Tn",
PROCESS_DELETE + ":T1",
PROCESS_ADD + ":Tx");
+ updaterAssert("header-Tx-prop-T1-No-Guid", "col-entity-T1-prop-Tn-No-Guid",
+ PROCESS_DELETE + ":Tn",
+ PROCESS_UPDATE + ":T1",
+ PROCESS_ADD + ":Tx");
}
diff --git a/repository/src/test/resources/json/classification-association/col-entity-T1-prop-Tn-No-Guid.json b/repository/src/test/resources/json/classification-association/col-entity-T1-prop-Tn-No-Guid.json
new file mode 100644
index 0000000..4a6fa80
--- /dev/null
+++ b/repository/src/test/resources/json/classification-association/col-entity-T1-prop-Tn-No-Guid.json
@@ -0,0 +1,31 @@
+{
+ "typeName": "hive_column",
+ "attributes": {
+ "owner": "hive",
+ "createTime": 1547071410000,
+ "qualifiedName": "stocks.daily@cl1",
+ "name": "daily"
+ },
+ "status": "ACTIVE",
+ "displayText": "daily",
+ "classifications": [
+ {
+ "typeName": "T1",
+ "attributes": {},
+ "entityStatus": "ACTIVE",
+ "propagate": false,
+ "validityPeriods": [],
+ "removePropagationsOnEntityDelete": false
+ },
+ {
+ "typeName": "Tn",
+ "attributes": {},
+ "entityStatus": "ACTIVE",
+ "propagate": false,
+ "validityPeriods": [],
+ "removePropagationsOnEntityDelete": false
+ }
+ ],
+ "meaningNames": [],
+ "meanings": []
+}
diff --git a/repository/src/test/resources/json/classification-association/header-Tx-prop-T1-No-Guid.json b/repository/src/test/resources/json/classification-association/header-Tx-prop-T1-No-Guid.json
new file mode 100644
index 0000000..0771851
--- /dev/null
+++ b/repository/src/test/resources/json/classification-association/header-Tx-prop-T1-No-Guid.json
@@ -0,0 +1,39 @@
+{
+ "guidHeaderMap": {
+ "0ce68113-77fe-4ed1-9585-69371202bd74": {
+ "typeName": "hive_column",
+ "attributes": {
+ "owner": "hive",
+ "qualifiedName": "hortoniabank.us_customers.nationalid@cl1",
+ "name": "nationalid"
+ },
+ "status": "ACTIVE",
+ "displayText": "nationalid",
+ "classificationNames": [
+ "T1", "Tx"
+ ],
+ "classifications": [
+ {
+ "typeName": "Tx",
+ "attributes": {
+ "type": "ssn"
+ },
+ "entityStatus": "ACTIVE",
+ "propagate": true,
+ "removePropagationsOnEntityDelete": false
+ },
+ {
+ "typeName": "T1",
+ "attributes": {
+ "type": "ssn"
+ },
+ "entityStatus": "ACTIVE",
+ "propagate": true,
+ "removePropagationsOnEntityDelete": false
+ }
+ ],
+ "meaningNames": [],
+ "meanings": []
+ }
+ }
+}
[atlas] 06/07: ATLAS-3046: Part 3. Updated case for handling
deleted classifications.
Posted by ni...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
nixon pushed a commit to branch branch-1.0
in repository https://gitbox.apache.org/repos/asf/atlas.git
commit c27aeef3e4dec369763fe04786bf43a0795faaeb
Author: Ashutosh Mestry <am...@hortonworks.com>
AuthorDate: Mon Feb 11 23:35:11 2019 -0800
ATLAS-3046: Part 3. Updated case for handling deleted classifications.
---
.../src/main/java/org/apache/atlas/tools/BulkFetchAndUpdate.java | 6 +-----
1 file changed, 1 insertion(+), 5 deletions(-)
diff --git a/tools/classification-updater/src/main/java/org/apache/atlas/tools/BulkFetchAndUpdate.java b/tools/classification-updater/src/main/java/org/apache/atlas/tools/BulkFetchAndUpdate.java
index 19f8325..c1e801f 100644
--- a/tools/classification-updater/src/main/java/org/apache/atlas/tools/BulkFetchAndUpdate.java
+++ b/tools/classification-updater/src/main/java/org/apache/atlas/tools/BulkFetchAndUpdate.java
@@ -386,12 +386,8 @@ public class BulkFetchAndUpdate {
continue;
}
- header.setGuid(null);
- if (header.getClassifications().size() == 0) {
- continue;
- }
-
String key = String.format("%s:%s", header.getTypeName(), uniqueName);
+ header.setGuid(null);
boolean keyFound = uniqueNameEntityHeaderMap.containsKey(key);
if (!keyFound) {
uniqueNameEntityHeaderMap.put(key, header);
[atlas] 04/07: ATLAS-3015: Classification Updater tool.
Posted by ni...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
nixon pushed a commit to branch branch-1.0
in repository https://gitbox.apache.org/repos/asf/atlas.git
commit bf59b4f973b6ccf689c6bfe7ae8162f1e0885211
Author: Ashutosh Mestry <am...@hortonworks.com>
AuthorDate: Wed Jan 23 11:13:34 2019 -0800
ATLAS-3015: Classification Updater tool.
---
tools/classification-updater/pom.xml | 114 +++++
.../org/apache/atlas/tools/BulkFetchAndUpdate.java | 510 +++++++++++++++++++++
.../src/main/resources/atlas-log4j.xml | 42 ++
.../src/main/resources/update-classifications.sh | 90 ++++
4 files changed, 756 insertions(+)
diff --git a/tools/classification-updater/pom.xml b/tools/classification-updater/pom.xml
new file mode 100644
index 0000000..4e23537
--- /dev/null
+++ b/tools/classification-updater/pom.xml
@@ -0,0 +1,114 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ 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.
+ -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <artifactId>apache-atlas</artifactId>
+ <groupId>org.apache.atlas</groupId>
+ <version>2.0.0-SNAPSHOT</version>
+ <relativePath>../../</relativePath>
+ </parent>
+ <artifactId>atlas-classification-updater</artifactId>
+ <description>Apache Atlas classification updater Module</description>
+ <name>Apache Atlas classification updater</name>
+ <packaging>jar</packaging>
+
+ <properties>
+ <calcite.version>0.9.2-incubating</calcite.version>
+ </properties>
+
+ <dependencies>
+ <!-- Logging -->
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-api</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-log4j12</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.atlas</groupId>
+ <artifactId>atlas-client-v1</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.atlas</groupId>
+ <artifactId>atlas-client-v2</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.atlas</groupId>
+ <artifactId>atlas-notification</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.atlas</groupId>
+ <artifactId>hdfs-model</artifactId>
+ </dependency>
+
+ <!-- to bring up atlas server for integration tests -->
+
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ <version>${jackson.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.testng</groupId>
+ <artifactId>testng</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-all</artifactId>
+ </dependency>
+
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-jar-plugin</artifactId>
+ <version>2.4</version>
+ <executions>
+ <execution>
+ <goals>
+ <goal>test-jar</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-compiler-plugin</artifactId>
+ <configuration>
+ <source>1.8</source>
+ <target>1.8</target>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+</project>
diff --git a/tools/classification-updater/src/main/java/org/apache/atlas/tools/BulkFetchAndUpdate.java b/tools/classification-updater/src/main/java/org/apache/atlas/tools/BulkFetchAndUpdate.java
new file mode 100644
index 0000000..1e0b66d
--- /dev/null
+++ b/tools/classification-updater/src/main/java/org/apache/atlas/tools/BulkFetchAndUpdate.java
@@ -0,0 +1,510 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.atlas.tools;
+
+import com.sun.jersey.core.util.MultivaluedMapImpl;
+import org.apache.atlas.ApplicationProperties;
+import org.apache.atlas.AtlasClientV2;
+import org.apache.atlas.AtlasException;
+import org.apache.atlas.AtlasServiceException;
+import org.apache.atlas.model.SearchFilter;
+import org.apache.atlas.model.instance.AtlasClassification;
+import org.apache.atlas.model.instance.AtlasEntityHeader;
+import org.apache.atlas.model.instance.AtlasEntityHeaders;
+import org.apache.atlas.model.typedef.AtlasClassificationDef;
+import org.apache.atlas.model.typedef.AtlasTypesDef;
+import org.apache.atlas.type.AtlasType;
+import org.apache.atlas.utils.AtlasJson;
+import org.apache.atlas.utils.AuthenticationUtil;
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.DefaultParser;
+import org.apache.commons.cli.Options;
+import org.apache.commons.cli.ParseException;
+import org.apache.commons.configuration.Configuration;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.ws.rs.core.MultivaluedMap;
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.FileReader;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.text.SimpleDateFormat;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.TimeZone;
+import java.util.function.Consumer;
+
+import static org.apache.atlas.model.instance.AtlasEntity.Status.DELETED;
+
+public class BulkFetchAndUpdate {
+ private static final Logger LOG = LoggerFactory.getLogger(BulkFetchAndUpdate.class);
+
+ private static final String DATE_FORMAT_SUPPORTED = "yyyy-MM-dd'T'HH:mm:ss";
+ private static final String OPTION_FROM = "f";
+
+ private static final String APPLICATION_PROPERTY_ATLAS_ENDPOINT = "atlas.rest.address";
+ private static final String SYSTEM_PROPERTY_USER_DIR = "user.dir";
+ private static final String STEP_PREPARE = "prepare";
+ private static final String STEP_UPDATE = "update";
+ private static final int EXIT_CODE_SUCCESS = 0;
+ private static final int EXIT_CODE_FAILED = 1;
+ private static final String DEFAULT_ATLAS_URL = "http://localhost:21000/";
+ private static final String FILE_CLASSIFICATION_DEFS = "classification-definitions.json";
+ private static final String FILE_ENTITY_HEADERS = "entity-headers.json";
+
+ private final static String[] filesToUse = new String[] {
+ FILE_CLASSIFICATION_DEFS,
+ FILE_ENTITY_HEADERS
+ };
+
+ public static void main(String[] args) {
+ int exitCode = EXIT_CODE_FAILED;
+
+ try {
+ long fromTimestamp = 0L;
+ CommandLine cmd = getCommandLine(args);
+
+ String stepToExecute = cmd.getOptionValue("s").trim();
+ String uid = cmd.getOptionValue("u");
+ String pwd = cmd.getOptionValue("p");
+ String directory = cmd.getOptionValue("d");
+ String fromTime = cmd.getOptionValue(OPTION_FROM);
+ String basePath = getDirectory(directory);
+
+ displayCrLf(basePath);
+
+ String[] atlasEndpoint = getAtlasRESTUrl();
+ if (atlasEndpoint == null || atlasEndpoint.length == 0) {
+ atlasEndpoint = new String[]{DEFAULT_ATLAS_URL};
+ }
+
+ if (StringUtils.equals(stepToExecute, STEP_PREPARE)) {
+ if (StringUtils.isEmpty(fromTime)) {
+ displayCrLf("'fromTime' is empty" + fromTime);
+ printUsage();
+ return;
+ }
+
+ fromTimestamp = getTimestamp(fromTime);
+ displayCrLf("fromTimestamp: " + fromTimestamp);
+ if (fromTimestamp == 0L) {
+ printUsage();
+ return;
+ }
+ }
+
+ process(stepToExecute, basePath, atlasEndpoint, uid, pwd, fromTimestamp);
+
+ exitCode = EXIT_CODE_SUCCESS;
+ } catch (ParseException e) {
+ LOG.error("Failed to parse arguments. Error: ", e.getMessage());
+ printUsage();
+ } catch (Exception e) {
+ LOG.error("Failed!", e);
+ displayCrLf("Failed: " + e.getMessage());
+ }
+
+ System.exit(exitCode);
+ }
+
+ private static long getTimestamp(String str) {
+ try {
+ if (StringUtils.isEmpty(str)) {
+ return 0;
+ }
+
+ TimeZone utc = TimeZone.getDefault();
+ SimpleDateFormat simpleDateFormat = new SimpleDateFormat(DATE_FORMAT_SUPPORTED);
+ simpleDateFormat.setTimeZone(utc);
+
+ return simpleDateFormat.parse(str).getTime();
+ } catch (java.text.ParseException e) {
+ displayCrLf("Unsupported date format: " + str);
+ return 0;
+ }
+ }
+
+ private static void process(String stepToExecute, String basePath, String[] atlasEndpoint, String uid, String pwd, long fromTimestamp) throws Exception {
+ AtlasClientV2 atlasClientV2 = getAtlasClientV2(atlasEndpoint, new String[]{uid, pwd});
+
+ switch (stepToExecute) {
+ case STEP_PREPARE: {
+ Preparer p = new Preparer(atlasClientV2);
+ p.run(basePath, fromTimestamp);
+ }
+ break;
+
+ case STEP_UPDATE: {
+ Updater u = new Updater(atlasClientV2);
+ u.run(basePath);
+ }
+ break;
+
+ default:
+ printUsage();
+ break;
+ }
+ }
+
+ private static String getDirectory(String directory) {
+ String basePath = System.getProperty(SYSTEM_PROPERTY_USER_DIR) + File.separatorChar;
+ if (StringUtils.isNotEmpty(directory) && checkDirectoryExists(directory)) {
+ basePath = directory + File.separatorChar;
+ } else {
+ display("Using directory: ");
+ }
+
+ return basePath;
+ }
+
+ private static CommandLine getCommandLine(String[] args) throws ParseException {
+ Options options = new Options();
+ options.addRequiredOption("s", "step", true, "Step to run.");
+ options.addOption("u", "user", true, "User name.");
+ options.addOption("p", "password", true, "Password name.");
+ options.addOption("d", "dir", true, "Directory for reading/writing data.");
+ options.addOption(OPTION_FROM, "fromDate", true, "Date, in YYYY-MM-DD format, from where to start reading.");
+
+ return new DefaultParser().parse(options, args);
+ }
+
+ private static void printUsage() {
+ System.out.println();
+ displayCrLf("Usage: classification-updater.sh [-s <step>] [-f <from time>] [-t <optional: to time>] [-d <dir>]");
+ displayCrLf(" step: Specify which step to execute:");
+ displayCrLf(" prepare: prepare classifications and associated entities.");
+ displayCrLf(" update: update classifications and entities.");
+ displayCrLf(" dir: [optional] Directory where read/write will happen.");
+ displayCrLf(" If not specified, current directory will be used.");
+ displayCrLf(" from: [mandatory for 'prepare' step, optional for 'update' step] Date, in YYYY-MM-DD format, from where audits need to be read.");
+ displayCrLf(" If not specified, current directory will be used.");
+ System.out.println();
+ }
+
+ private static String[] getAtlasRESTUrl() {
+ Configuration atlasConf = null;
+ try {
+ atlasConf = ApplicationProperties.get();
+ return atlasConf.getStringArray(APPLICATION_PROPERTY_ATLAS_ENDPOINT);
+ } catch (AtlasException e) {
+ return new String[]{DEFAULT_ATLAS_URL};
+ }
+ }
+
+ private static AtlasClientV2 getAtlasClientV2(String[] atlasEndpoint, String[] uidPwdFromCommandLine) throws IOException {
+ AtlasClientV2 atlasClientV2;
+ if (!AuthenticationUtil.isKerberosAuthenticationEnabled()) {
+ String[] uidPwd = (uidPwdFromCommandLine[0] == null || uidPwdFromCommandLine[1] == null)
+ ? AuthenticationUtil.getBasicAuthenticationInput()
+ : uidPwdFromCommandLine;
+
+ atlasClientV2 = new AtlasClientV2(atlasEndpoint, uidPwd);
+
+ } else {
+ UserGroupInformation ugi = UserGroupInformation.getCurrentUser();
+ atlasClientV2 = new AtlasClientV2(ugi, ugi.getShortUserName(), atlasEndpoint);
+ }
+ return atlasClientV2;
+ }
+
+ private static void displayCrLf(String... formatMessage) {
+ displayFn(System.out::println, formatMessage);
+ }
+
+ private static void display(String... formatMessage) {
+ displayFn(System.out::print, formatMessage);
+ }
+
+ private static void displayFn(Consumer<String> fn, String... formatMessage) {
+ if (formatMessage.length == 1) {
+ fn.accept(formatMessage[0]);
+ } else {
+ fn.accept(String.format(formatMessage[0], formatMessage[1]));
+ }
+ }
+ private static void closeReader(BufferedReader bufferedReader) {
+ try {
+ if (bufferedReader == null) {
+ return;
+ }
+
+ bufferedReader.close();
+ } catch (IOException ex) {
+ LOG.error("closeReader", ex);
+ }
+ }
+
+ private static BufferedReader getBufferedReader(String basePath, String fileName) throws FileNotFoundException {
+ return new BufferedReader(new FileReader(basePath + fileName));
+ }
+
+ private static boolean fileCheck(String basePath, String[] files, boolean existCheck) {
+ boolean ret = true;
+ for (String f : files) {
+ ret = ret && fileCheck(basePath, f, existCheck);
+ }
+
+ return ret;
+ }
+
+ private static boolean fileCheck(String basePath, String file, boolean existCheck) {
+ String errorMessage = existCheck ? "does not exist" : "exists" ;
+ if (checkFileExists(basePath + file) != existCheck) {
+ displayCrLf(String.format("File '%s' %s!", basePath + file, errorMessage));
+ return false;
+ }
+
+ return true;
+ }
+
+ private static boolean checkFileExists(String fileName) {
+ File f = new File(fileName);
+ return f.exists() && !f.isDirectory();
+ }
+
+ private static boolean checkDirectoryExists(String fileName) {
+ File f = new File(fileName);
+ return f.exists() && f.isDirectory();
+ }
+
+ private static FileWriter getFileWriter(String basePath, String fileName) throws IOException {
+ String filePath = basePath + fileName;
+ displayCrLf("Creating %s", filePath);
+ return new FileWriter(filePath, true);
+ }
+
+ private static class Preparer {
+ private static final String ATTR_NAME_QUALIFIED_NAME = "qualifiedName";
+
+ private AtlasClientV2 atlasClientV2;
+
+ public Preparer(AtlasClientV2 atlasClientV2) {
+ this.atlasClientV2 = atlasClientV2;
+ }
+
+ public void run(String basePath, long fromTimestamp) throws Exception {
+ if (!fileCheck(basePath, filesToUse, false)) return;
+
+ displayCrLf("Starting: from: " + fromTimestamp + " to: " + "current time (" + System.currentTimeMillis() + ")...");
+ writeClassificationDefs(basePath, FILE_CLASSIFICATION_DEFS, getAllClassificationsDefs());
+ writeEntityHeaders(basePath, FILE_ENTITY_HEADERS, fromTimestamp);
+ displayCrLf("Done!");
+ }
+
+ private void writeClassificationDefs(String basePath, String fileName, List<AtlasClassificationDef> classificationDefs) throws IOException {
+ FileWriter fileWriter = null;
+ try {
+ fileWriter = getFileWriter(basePath, fileName);
+ for (AtlasClassificationDef classificationDef : classificationDefs) {
+ try {
+ classificationDef.setGuid(null);
+ String json = AtlasType.toJson(classificationDef);
+ fileWriter.write(json + "\n");
+ } catch (Exception e) {
+ LOG.error("Error writing classifications: {}", e);
+ displayCrLf("Error writing classifications.");
+ }
+ }
+ }
+ finally {
+ if (fileWriter != null) {
+ fileWriter.close();
+ }
+ }
+ }
+
+ private void writeEntityHeaders(String basePath, String fileName, long fromTimestamp) throws AtlasServiceException, IOException {
+ FileWriter fileWriter = null;
+
+ try {
+ fileWriter = getFileWriter(basePath, fileName);
+ } catch (IOException e) {
+ LOG.error("Error opening {}/{}", basePath, fileName, e);
+ displayCrLf("Error opening: %", basePath + File.separatorChar + fileName);
+ return;
+ }
+
+ try {
+ AtlasEntityHeaders response = atlasClientV2.getEntityHeaders(fromTimestamp);
+ int guidHeaderMapSize = response.getGuidHeaderMap().size();
+ try {
+ displayCrLf("Read entities: " + guidHeaderMapSize);
+ AtlasEntityHeaders updatedHeaders = removeEntityGuids(response);
+ fileWriter.write(AtlasType.toJson(updatedHeaders));
+
+ displayCrLf("Writing entities: " + updatedHeaders.getGuidHeaderMap().size());
+ } catch (Exception e) {
+ LOG.error("Error writing: {}", guidHeaderMapSize, e);
+ displayCrLf("Error writing: " + e.toString());
+ }
+ } finally {
+ if (fileWriter != null) {
+ fileWriter.close();
+ }
+ }
+ }
+
+ private AtlasEntityHeaders removeEntityGuids(AtlasEntityHeaders headers) {
+ Map<String, AtlasEntityHeader> qualifiedNameHeaderMap = new HashMap<>();
+
+ for (AtlasEntityHeader header : headers.getGuidHeaderMap().values()) {
+ String qualifiedName = getQualifiedName(header);
+ displayCrLf("Processing: " + qualifiedName);
+
+ if (header.getStatus() == DELETED) {
+ continue;
+ }
+
+ header.setGuid(null);
+ if (header.getClassifications().size() == 0) {
+ continue;
+ }
+
+ boolean keyFound = qualifiedNameHeaderMap.containsKey(qualifiedName);
+ if (!keyFound) {
+ qualifiedNameHeaderMap.put(qualifiedName, header);
+ }
+
+ AtlasEntityHeader currentHeader = qualifiedNameHeaderMap.get(qualifiedName);
+ for (AtlasClassification c : header.getClassifications()) {
+ c.setEntityGuid(null);
+
+ if (keyFound) {
+ boolean found =
+ currentHeader.getClassifications().stream().anyMatch(ox -> ox.getTypeName().equals(c.getTypeName()));
+ if (!found) {
+ currentHeader.getClassifications().add(c);
+ } else {
+ displayCrLf("Ignoring: " + c.toString());
+ LOG.warn("Ignoring: {}", AtlasJson.toJson(c));
+ }
+ }
+ }
+
+ displayCrLf("Processing: " + qualifiedName);
+ }
+
+ displayCrLf("Processed: " + qualifiedNameHeaderMap.size());
+ headers.setGuidHeaderMap(qualifiedNameHeaderMap);
+ return headers;
+ }
+
+ private String getQualifiedName(AtlasEntityHeader header) {
+ return (String) header.getAttribute(ATTR_NAME_QUALIFIED_NAME);
+ }
+
+ private List<AtlasClassificationDef> getAllClassificationsDefs() throws Exception {
+ MultivaluedMap<String, String> searchParams = new MultivaluedMapImpl();
+ searchParams.add(SearchFilter.PARAM_TYPE, "CLASSIFICATION");
+ SearchFilter searchFilter = new SearchFilter(searchParams);
+
+ AtlasTypesDef typesDef = atlasClientV2.getAllTypeDefs(searchFilter);
+ displayCrLf("Found classifications: " + typesDef.getClassificationDefs().size());
+ return typesDef.getClassificationDefs();
+ }
+ }
+
+ private static class Updater {
+ private AtlasClientV2 atlasClientV2;
+
+ public Updater(AtlasClientV2 atlasClientV2) {
+
+ this.atlasClientV2 = atlasClientV2;
+ }
+
+ public void run(String basePath) throws Exception {
+ if (!fileCheck(basePath, filesToUse, true)) return;
+
+ displayCrLf("Starting...");
+ readAndCreateOrUpdateClassificationDefs(basePath, FILE_CLASSIFICATION_DEFS);
+ readEntityUpdates(basePath, FILE_ENTITY_HEADERS);
+ displayCrLf("Done!");
+ }
+
+ private void readEntityUpdates(String basePath, String fileName) throws IOException {
+ BufferedReader bufferedReader = null;
+ try {
+ bufferedReader = getBufferedReader(basePath, fileName);
+ String json = bufferedReader.readLine();
+ if (StringUtils.isEmpty(json)) {
+ displayCrLf("Empty file encountered: %s", fileName);
+ return;
+ }
+
+ AtlasEntityHeaders response = AtlasType.fromJson(json, AtlasEntityHeaders.class);
+ displayCrLf("Found :" + response.getGuidHeaderMap().size());
+ String output = atlasClientV2.setClassifications(response);
+ displayCrLf(output);
+ } catch (AtlasServiceException e) {
+ displayCrLf("Error updating. Please see log for details.");
+ LOG.error("Error updating. {}", e);
+ } finally {
+ closeReader(bufferedReader);
+ }
+ }
+
+ private void readAndCreateOrUpdateClassificationDefs(String basePath, String fileName) throws Exception {
+ BufferedReader bufferedReader = null;
+ try {
+ bufferedReader = getBufferedReader(basePath, fileName);
+
+ for (String cd; (cd = bufferedReader.readLine()) != null; ) {
+ AtlasClassificationDef classificationDef = AtlasType.fromJson(cd, AtlasClassificationDef.class);
+ createOrUpdateClassification(classificationDef);
+ }
+ } finally {
+ closeReader(bufferedReader);
+ }
+ }
+
+ private void createOrUpdateClassification(AtlasClassificationDef classificationDef) {
+ String name = classificationDef.getName();
+ AtlasTypesDef typesDef = new AtlasTypesDef(null, null, Collections.singletonList(classificationDef), null, null);
+ try {
+ display("%s -> ", name);
+ atlasClientV2.createAtlasTypeDefs(typesDef);
+ displayCrLf(" [Done]");
+ } catch (AtlasServiceException e) {
+ LOG.error("{} skipped!", name, e);
+ displayCrLf(" [Skipped]", name);
+ updateClassification(classificationDef);
+ }
+ }
+
+ private void updateClassification(AtlasClassificationDef classificationDef) {
+ String name = classificationDef.getName();
+ AtlasTypesDef typesDef = new AtlasTypesDef(null, null, Collections.singletonList(classificationDef), null, null);
+ try {
+ display("Update: %s -> ", name);
+ atlasClientV2.updateAtlasTypeDefs(typesDef);
+ displayCrLf(" [Done]");
+ } catch (AtlasServiceException e) {
+ LOG.error("{} skipped!", name, e);
+ displayCrLf(" [Skipped]", name);
+ }
+ }
+ }
+}
diff --git a/tools/classification-updater/src/main/resources/atlas-log4j.xml b/tools/classification-updater/src/main/resources/atlas-log4j.xml
new file mode 100644
index 0000000..9658edf
--- /dev/null
+++ b/tools/classification-updater/src/main/resources/atlas-log4j.xml
@@ -0,0 +1,42 @@
+<?xml version="1.0" encoding="UTF-8" ?>
+<!--
+ ~ 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.
+ -->
+
+<!DOCTYPE log4j:configuration SYSTEM "log4j.dtd">
+
+<log4j:configuration xmlns:log4j="http://jakarta.apache.org/log4j/">
+ <appender name="FILE" class="org.apache.log4j.RollingFileAppender">
+ <param name="File" value="/var/log/atlas/classification-updater.log"/>
+ <param name="Append" value="true"/>
+ <param name="maxFileSize" value="100MB" />
+ <param name="maxBackupIndex" value="20" />
+ <layout class="org.apache.log4j.PatternLayout">
+ <param name="ConversionPattern" value="%d %-5p - [%t:%x] ~ %m (%C{1}:%L)%n"/>
+ </layout>
+ </appender>
+
+ <logger name="org.apache.atlas.tools.BulkFetchAndUpdate" additivity="false">
+ <level value="info"/>
+ <appender-ref ref="FILE"/>
+ </logger>
+
+ <root>
+ <priority value="warn"/>
+ <appender-ref ref="FILE"/>
+ </root>
+</log4j:configuration>
diff --git a/tools/classification-updater/src/main/resources/update-classifications.sh b/tools/classification-updater/src/main/resources/update-classifications.sh
new file mode 100644
index 0000000..c11d5e1
--- /dev/null
+++ b/tools/classification-updater/src/main/resources/update-classifications.sh
@@ -0,0 +1,90 @@
+#!/bin/bash
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License. See accompanying LICENSE file.
+#
+# resolve links - $0 may be a softlink
+PRG="${0}"
+
+[[ `uname -s` == *"CYGWIN"* ]] && CYGWIN=true
+
+while [ -h "${PRG}" ]; do
+ ls=`ls -ld "${PRG}"`
+ link=`expr "$ls" : '.*-> \(.*\)$'`
+ if expr "$link" : '/.*' > /dev/null; then
+ PRG="$link"
+ else
+ PRG=`dirname "${PRG}"`/"$link"
+ fi
+done
+
+BASEDIR=`dirname ${PRG}`
+
+if test -z "${JAVA_HOME}"
+then
+ JAVA_BIN=`which java`
+ JAR_BIN=`which jar`
+else
+ JAVA_BIN="${JAVA_HOME}/bin/java"
+ JAR_BIN="${JAVA_HOME}/bin/jar"
+fi
+export JAVA_BIN
+
+if [ ! -e "${JAVA_BIN}" ] || [ ! -e "${JAR_BIN}" ]; then
+ echo "$JAVA_BIN and/or $JAR_BIN not found on the system. Please make sure java and jar commands are available."
+ exit 1
+fi
+
+# Construct ATLAS_CONF where atlas-properties reside
+export ATLAS_CONF=/usr/hdp/current/atlas-server/conf/
+
+# log dir for applications
+ATLAS_LOG_DIR="/var/log/atlas"
+ATLAS_LOG_FILE="classification-updater.log"
+LOG_CONFIG="${BASEDIR}/atlas-log4j.xml"
+
+# Construct Atlas classpath.
+for i in "/usr/hdp/current/atlas-server/server/webapp/atlas/WEB-INF/lib/"*.jar; do
+ ATLASCPPATH="${ATLASCPPATH}:$i"
+done
+
+for i in "${BASEDIR}/"*.jar; do
+ ATLASCPPATH="${ATLASCPPATH}:$i"
+done
+
+echo "Logging: ${ATLAS_LOG_DIR}/${ATLAS_LOG_FILE}"
+echo "Log config: ${LOG_CONFIG}"
+
+TIME=`date +%Y%m%d%H%M%s`
+CP="${ATLASCPPATH}:${ATLAS_CONF}"
+
+# If running in cygwin, convert pathnames and classpath to Windows format.
+if [ "${CYGWIN}" == "true" ]
+then
+ ATLAS_LOG_DIR=`cygpath -w ${ATLAS_LOG_DIR}`
+ ATLAS_LOG_FILE=`cygpath -w ${ATLAS_LOG_FILE}`
+ CP=`cygpath -w -p ${CP}`
+fi
+
+JAVA_PROPERTIES="$ATLAS_OPTS -Datlas.log.dir=$ATLAS_LOG_DIR -Datlas.log.file=$ATLAS_LOG_FILE -Dlog4j.configuration=file://$LOG_CONFIG"
+
+IMPORT_ARGS=$@
+JVM_ARGS=
+
+JAVA_PROPERTIES="${JAVA_PROPERTIES} ${JVM_ARGS}"
+
+"${JAVA_BIN}" ${JAVA_PROPERTIES} -cp "${CP}" org.apache.atlas.tools.BulkFetchAndUpdate $IMPORT_ARGS
+
+RETVAL=$?
+[ $RETVAL -eq 0 ] && echo Done!
+[ $RETVAL -ne 0 ] && echo Failed!
+exit $RETVAL
[atlas] 02/07: ATLAS-3029: Audit APIs for classification updates.
Posted by ni...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
nixon pushed a commit to branch branch-1.0
in repository https://gitbox.apache.org/repos/asf/atlas.git
commit 752587a9d94214402ef4765917a4e93358777e6e
Author: Ashutosh Mestry <am...@hortonworks.com>
AuthorDate: Mon Jan 21 21:32:42 2019 -0800
ATLAS-3029: Audit APIs for classification updates.
---
.../store/graph/v2/ClassificationAssociator.java | 14 +++++++-------
1 file changed, 7 insertions(+), 7 deletions(-)
diff --git a/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/ClassificationAssociator.java b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/ClassificationAssociator.java
index 11d008f..057598c 100644
--- a/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/ClassificationAssociator.java
+++ b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/ClassificationAssociator.java
@@ -171,7 +171,7 @@ public class ClassificationAssociator {
operationListMap.put(op, results);
}
- private void addClassifications(String entityToBeChangedGuid, String typeName, String qualifiedName, List<AtlasClassification> list) {
+ private void addClassifications(String entityGuid, String typeName, String qualifiedName, List<AtlasClassification> list) {
if (CollectionUtils.isEmpty(list)) {
return;
}
@@ -179,16 +179,16 @@ public class ClassificationAssociator {
String status = STATUS_DONE;
String classificationNames = getClassificationNames(list);
try {
- entitiesStore.addClassifications(entityToBeChangedGuid, list);
+ entitiesStore.addClassifications(entityGuid, list);
} catch (AtlasBaseException e) {
status = STATUS_PARTIAL;
LOG.warn("{}:{}:{} -> {}: {}.", PROCESS_UPDATE, typeName, qualifiedName, classificationNames, status);
}
- summarize(PROCESS_ADD, entityToBeChangedGuid, typeName, qualifiedName, classificationNames, status);
+ summarize(PROCESS_ADD, entityGuid, typeName, qualifiedName, classificationNames, status);
}
- private void updateClassifications(String entityToBeChangedGuid, String typeName, String qualifiedName, List<AtlasClassification> list) {
+ private void updateClassifications(String entityGuid, String typeName, String qualifiedName, List<AtlasClassification> list) {
if (CollectionUtils.isEmpty(list)) {
return;
}
@@ -197,16 +197,16 @@ public class ClassificationAssociator {
String classificationNames = getClassificationNames(list);
try {
- entitiesStore.updateClassifications(entityToBeChangedGuid, list);
+ entitiesStore.updateClassifications(entityGuid, list);
} catch (AtlasBaseException e) {
status = STATUS_PARTIAL;
LOG.warn("{}:{}:{} -> {}: {}.", PROCESS_UPDATE, typeName, qualifiedName, classificationNames, status);
}
- summarize(PROCESS_UPDATE, entityToBeChangedGuid, typeName, qualifiedName, classificationNames, status);
+ summarize(PROCESS_UPDATE, entityGuid, typeName, qualifiedName, classificationNames, status);
}
- private void deleteClassifications(String typeName, String entityGuid, String qualifiedName, List<AtlasClassification> list) {
+ private void deleteClassifications(String entityGuid, String typeName, String qualifiedName, List<AtlasClassification> list) {
if (CollectionUtils.isEmpty(list)) {
return;
}
[atlas] 07/07: ATLAS-3066 : UI : Fix various table layouts &
improvements.
Posted by ni...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
nixon pushed a commit to branch branch-1.0
in repository https://gitbox.apache.org/repos/asf/atlas.git
commit 36d8014dc5006da5f815ad8a592a0f0ecf0c1182
Author: gutkaBinit <bi...@gmail.com>
AuthorDate: Tue Mar 5 16:28:38 2019 +0530
ATLAS-3066 : UI : Fix various table layouts & improvements.
Signed-off-by: nixonrodrigues <ni...@apache.org>
---
dashboardv2/public/css/scss/common.scss | 2 +-
dashboardv2/public/js/utils/CommonViewFunction.js | 10 +++++-----
dashboardv2/public/js/views/audit/AuditTableLayoutView.js | 2 +-
.../public/js/views/detail_page/DetailPageLayoutView.js | 2 +-
4 files changed, 8 insertions(+), 8 deletions(-)
diff --git a/dashboardv2/public/css/scss/common.scss b/dashboardv2/public/css/scss/common.scss
index 35acc43..484879b 100644
--- a/dashboardv2/public/css/scss/common.scss
+++ b/dashboardv2/public/css/scss/common.scss
@@ -118,7 +118,7 @@ pre {
overflow: hidden;
&.shrink {
- height: 100px;
+ height: 112px;
white-space: -moz-pre-wrap;
/* Mozilla, supported since 1999 */
white-space: -pre-wrap;
diff --git a/dashboardv2/public/js/utils/CommonViewFunction.js b/dashboardv2/public/js/utils/CommonViewFunction.js
index 8532d49..e9477c3 100644
--- a/dashboardv2/public/js/utils/CommonViewFunction.js
+++ b/dashboardv2/public/js/utils/CommonViewFunction.js
@@ -89,7 +89,7 @@ define(['require', 'utils/Utils', 'modules/Modal', 'utils/Messages', 'utils/Enum
return numberFormat(val);
}
} else {
- return val;
+ return val || "N/A";
}
},
fetchInputOutputValue = function(id, defEntity) {
@@ -252,16 +252,16 @@ define(['require', 'utils/Utils', 'modules/Modal', 'utils/Messages', 'utils/Enum
}
if (isTable) {
var htmlTag = '<div class="scroll-y">' + getValue(val) + '</div>';
- if (_.isObject(valueObject[key])) {
+ if (_.isObject(valueObject[key]) && !_.isEmpty(valueObject[key])) {
var matchedLinkString = val.match(/href|value-loader\w*/g),
matchedJson = val.match(/json-value|json-string\w*/g),
- isMatchLinkStringIsSingle = matchedLinkString && matchedLinkString.length == 1,
+ isMatchLinkStringIsSingle = matchedLinkString && matchedLinkString.length <= 5,
isMatchJSONStringIsSingle = matchedJson && matchedJson.length == 1,
expandCollapseButton = "";
if ((matchedJson && !isMatchJSONStringIsSingle) || (matchedLinkString && !isMatchLinkStringIsSingle)) {
- var expandCollapseButton = '<button class="expand-collapse-button"><i class="fa"></i></button>'
+ expandCollapseButton = '<button class="expand-collapse-button"><i class="fa"></i></button>';
+ htmlTag = '<pre class="shrink code-block ' + (isMatchJSONStringIsSingle ? 'fixed-height' : '') + '">' + expandCollapseButton + '<code>' + val + '</code></pre>';
}
- var htmlTag = '<pre class="shrink code-block ' + (isMatchJSONStringIsSingle ? 'fixed-height' : '') + '">' + expandCollapseButton + '<code>' + val + '</code></pre>';
}
table += '<tr><td>' + _.escape(key) + '</td><td>' + htmlTag + '</td></tr>';
} else {
diff --git a/dashboardv2/public/js/views/audit/AuditTableLayoutView.js b/dashboardv2/public/js/views/audit/AuditTableLayoutView.js
index a30158c..5874a18 100644
--- a/dashboardv2/public/js/views/audit/AuditTableLayoutView.js
+++ b/dashboardv2/public/js/views/audit/AuditTableLayoutView.js
@@ -153,9 +153,9 @@ define(['require',
if (that.pervOld.length === 0) {
options.previous.attr('disabled', true);
}
- that.renderTableLayoutView();
}
}
+ that.renderTableLayoutView();
that.$('.fontLoader').hide();
that.$('.tableOverlay').hide();
that.$('.auditTable').show(); // Only for first time table show because we never hide after first render.
diff --git a/dashboardv2/public/js/views/detail_page/DetailPageLayoutView.js b/dashboardv2/public/js/views/detail_page/DetailPageLayoutView.js
index 4b04fea..18c545b 100644
--- a/dashboardv2/public/js/views/detail_page/DetailPageLayoutView.js
+++ b/dashboardv2/public/js/views/detail_page/DetailPageLayoutView.js
@@ -247,7 +247,7 @@ define(['require',
if (schemaOptions && schemaOptions.hasOwnProperty('schemaElementsAttribute') && schemaOptions.schemaElementsAttribute !== "") {
this.$('.schemaTable').show();
this.renderSchemaLayoutView(_.extend({}, obj, {
- attribute: collectionJSON.attributes[schemaOptions.schemaElementsAttribute]
+ attribute: collectionJSON.attributes[schemaOptions.schemaElementsAttribute] || collectionJSON.relationshipAttributes[schemaOptions.schemaElementsAttribute]
}));
} else if (this.value && this.value.tabActive == "schema") {
Utils.setUrl({
[atlas] 01/07: ATLAS-3020: Audit APIs for classification updates.
Posted by ni...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
nixon pushed a commit to branch branch-1.0
in repository https://gitbox.apache.org/repos/asf/atlas.git
commit b3df6895f311589a7cee19073052ebb38ce8b513
Author: Ashutosh Mestry <am...@hortonworks.com>
AuthorDate: Mon Jan 21 20:34:50 2019 -0800
ATLAS-3020: Audit APIs for classification updates.
---
.../main/java/org/apache/atlas/AtlasClientV2.java | 21 +-
.../atlas/model/instance/AtlasEntityHeaders.java | 56 ++++
.../audit/CassandraBasedAuditRepository.java | 7 +
.../repository/audit/EntityAuditRepository.java | 9 +
.../audit/HBaseBasedAuditRepository.java | 52 +++-
.../audit/InMemoryEntityAuditRepository.java | 17 ++
.../audit/NoopEntityAuditRepository.java | 7 +
.../repository/store/graph/AtlasEntityStore.java | 3 +
.../store/graph/v2/AtlasEntityStoreV2.java | 7 +
.../store/graph/v2/ClassificationAssociator.java | 316 +++++++++++++++++++++
.../store/graph/v2/EntityGraphRetriever.java | 2 +-
.../graph/v2/ClassificationAssociatorTest.java | 235 +++++++++++++++
.../col-entity-None.json | 10 +
.../col-entity-PII-FIN_PII.json | 32 +++
.../classification-association/col-entity-PII.json | 22 ++
.../col-entity-T1-prop-Tn.json | 34 +++
.../classification-association/header-FIN_PII.json | 32 +++
.../classification-association/header-None.json | 21 ++
.../header-PII-VENDOR_PII.json | 42 +++
.../classification-association/header-PII.json | 32 +++
.../header-Tx-prop-T1.json | 42 +++
.../json/classification-association/header-Tx.json | 26 ++
.../classification-association/header-empty.json | 3 +
.../java/org/apache/atlas/web/rest/EntityREST.java | 59 +++-
24 files changed, 1081 insertions(+), 6 deletions(-)
diff --git a/client/client-v2/src/main/java/org/apache/atlas/AtlasClientV2.java b/client/client-v2/src/main/java/org/apache/atlas/AtlasClientV2.java
index 7c8caee..33466e5 100644
--- a/client/client-v2/src/main/java/org/apache/atlas/AtlasClientV2.java
+++ b/client/client-v2/src/main/java/org/apache/atlas/AtlasClientV2.java
@@ -27,7 +27,9 @@ import org.apache.atlas.model.instance.AtlasClassification;
import org.apache.atlas.model.instance.AtlasClassification.AtlasClassifications;
import org.apache.atlas.model.instance.AtlasEntity.AtlasEntitiesWithExtInfo;
import org.apache.atlas.model.instance.AtlasEntity.AtlasEntityWithExtInfo;
+import org.apache.atlas.model.instance.AtlasEntityHeader;
import org.apache.atlas.model.instance.AtlasRelationship;
+import org.apache.atlas.model.instance.AtlasEntityHeaders;
import org.apache.atlas.model.instance.EntityMutationResponse;
import org.apache.atlas.model.lineage.AtlasLineageInfo;
import org.apache.atlas.model.lineage.AtlasLineageInfo.LineageDirection;
@@ -72,6 +74,8 @@ public class AtlasClientV2 extends AtlasBaseClient {
// Relationships APIs
private static final String RELATIONSHIPS_URI = BASE_URI + "v2/relationship/";
+ private static final String BULK_HEADERS = "bulk/headers";
+ private static final String BULK_SET_CLASSIFICATIONS = "bulk/setClassifications";
public AtlasClientV2(String[] baseUrl, String[] basicAuthUserNamePassword) {
super(baseUrl, basicAuthUserNamePassword);
@@ -326,13 +330,26 @@ public class AtlasClientV2 extends AtlasBaseClient {
}
public void deleteClassifications(String guid, List<AtlasClassification> classifications) throws AtlasServiceException {
- callAPI(formatPathParameters(API_V2.GET_CLASSIFICATIONS, guid), AtlasClassifications.class, classifications);
+ for (AtlasClassification c : classifications) {
+ callAPI(formatPathParameters(API_V2.DELETE_CLASSIFICATION, guid, c.getTypeName()), AtlasClassifications.class, classifications);
+ }
}
public void deleteClassification(String guid, String classificationName) throws AtlasServiceException {
callAPI(formatPathParameters(API_V2.DELETE_CLASSIFICATION, guid, classificationName), null, null);
}
+ public AtlasEntityHeaders getEntityHeaders(long tagUpdateStartTime) throws AtlasServiceException {
+ MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
+ queryParams.add("tagUpdateStartTime", Long.toString(tagUpdateStartTime));
+
+ return callAPI(API_V2.GET_BULK_HEADERS, AtlasEntityHeaders.class, queryParams);
+ }
+
+ public String setClassifications(AtlasEntityHeaders entityHeaders) throws AtlasServiceException {
+ return callAPI(API_V2.UPDATE_BULK_SET_CLASSIFICATIONS, String.class, entityHeaders);
+ }
+
/* Discovery calls */
public AtlasSearchResult dslSearch(final String query) throws AtlasServiceException {
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
@@ -480,6 +497,8 @@ public class AtlasClientV2 extends AtlasBaseClient {
public static final API_V2 DELETE_RELATIONSHIP_BY_GUID = new API_V2(RELATIONSHIPS_URI + "guid/", HttpMethod.DELETE, Response.Status.NO_CONTENT);
public static final API_V2 CREATE_RELATIONSHIP = new API_V2(RELATIONSHIPS_URI , HttpMethod.POST, Response.Status.OK);
public static final API_V2 UPDATE_RELATIONSHIP = new API_V2(RELATIONSHIPS_URI , HttpMethod.PUT, Response.Status.OK);
+ public static final API_V2 GET_BULK_HEADERS = new API_V2(ENTITY_API + BULK_HEADERS, HttpMethod.GET, Response.Status.OK);
+ public static final API_V2 UPDATE_BULK_SET_CLASSIFICATIONS = new API_V2(ENTITY_API + AtlasClientV2.BULK_SET_CLASSIFICATIONS, HttpMethod.POST, Response.Status.OK);
private API_V2(String path, String method, Response.Status status) {
super(path, method, status);
diff --git a/intg/src/main/java/org/apache/atlas/model/instance/AtlasEntityHeaders.java b/intg/src/main/java/org/apache/atlas/model/instance/AtlasEntityHeaders.java
new file mode 100644
index 0000000..11c6fc7
--- /dev/null
+++ b/intg/src/main/java/org/apache/atlas/model/instance/AtlasEntityHeaders.java
@@ -0,0 +1,56 @@
+/**
+ * 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.atlas.model.instance;
+
+import com.fasterxml.jackson.annotation.JsonAutoDetect;
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+
+import javax.xml.bind.annotation.XmlAccessType;
+import javax.xml.bind.annotation.XmlAccessorType;
+import javax.xml.bind.annotation.XmlRootElement;
+
+import java.util.Map;
+
+import static com.fasterxml.jackson.annotation.JsonAutoDetect.Visibility.NONE;
+import static com.fasterxml.jackson.annotation.JsonAutoDetect.Visibility.PUBLIC_ONLY;
+
+@JsonAutoDetect(getterVisibility=PUBLIC_ONLY, setterVisibility=PUBLIC_ONLY, fieldVisibility=NONE)
+@JsonSerialize(include=JsonSerialize.Inclusion.NON_NULL)
+@JsonIgnoreProperties(ignoreUnknown=true)
+@XmlRootElement
+@XmlAccessorType(XmlAccessType.PROPERTY)
+public class AtlasEntityHeaders {
+ Map<String, AtlasEntityHeader> guidHeaderMap;
+
+ public AtlasEntityHeaders() {
+ }
+
+ public AtlasEntityHeaders(Map<String, AtlasEntityHeader> guidEntityHeaderMap) {
+ guidHeaderMap = guidEntityHeaderMap;
+ }
+
+ public void setGuidHeaderMap(Map<String, AtlasEntityHeader> guidHeaderMap) {
+ this.guidHeaderMap = guidHeaderMap;
+ }
+
+ public Map<String, AtlasEntityHeader> getGuidHeaderMap() {
+ return guidHeaderMap;
+ }
+}
diff --git a/repository/src/main/java/org/apache/atlas/repository/audit/CassandraBasedAuditRepository.java b/repository/src/main/java/org/apache/atlas/repository/audit/CassandraBasedAuditRepository.java
index eb78f8f..b8131bd 100644
--- a/repository/src/main/java/org/apache/atlas/repository/audit/CassandraBasedAuditRepository.java
+++ b/repository/src/main/java/org/apache/atlas/repository/audit/CassandraBasedAuditRepository.java
@@ -30,6 +30,7 @@ import org.apache.atlas.EntityAuditEvent;
import org.apache.atlas.annotation.ConditionalOnAtlasProperty;
import org.apache.atlas.exception.AtlasBaseException;
import org.apache.atlas.model.audit.EntityAuditEventV2;
+import org.apache.commons.lang.NotImplementedException;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -39,6 +40,7 @@ import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import javax.inject.Singleton;
@@ -188,6 +190,11 @@ public class CassandraBasedAuditRepository extends AbstractStorageBasedAuditRepo
}
@Override
+ public Set<String> getEntitiesWithTagChanges(long fromTimestamp, long toTimestamp) throws AtlasBaseException {
+ throw new NotImplementedException();
+ }
+
+ @Override
public void start() throws AtlasException {
initApplicationProperties();
initializeSettings();
diff --git a/repository/src/main/java/org/apache/atlas/repository/audit/EntityAuditRepository.java b/repository/src/main/java/org/apache/atlas/repository/audit/EntityAuditRepository.java
index aab2d5b..2a47e39 100644
--- a/repository/src/main/java/org/apache/atlas/repository/audit/EntityAuditRepository.java
+++ b/repository/src/main/java/org/apache/atlas/repository/audit/EntityAuditRepository.java
@@ -24,6 +24,7 @@ import org.apache.atlas.model.audit.EntityAuditEventV2;
import org.apache.atlas.exception.AtlasBaseException;
import java.util.List;
+import java.util.Set;
/**
* Interface for repository for storing entity audit events
@@ -77,6 +78,14 @@ public interface EntityAuditRepository {
*/
List<EntityAuditEventV2> listEventsV2(String entityId, String startKey, short n) throws AtlasBaseException;
+ /***
+ * List events for given time range where classifications have been added, deleted or updated.
+ * @param fromTimestamp from timestamp
+ * @param toTimestamp to timestamp
+ * @return events that match the range
+ * @throws AtlasBaseException
+ */
+ Set<String> getEntitiesWithTagChanges(long fromTimestamp, long toTimestamp) throws AtlasBaseException;
/**
* List events for the given entity id in decreasing order of timestamp, from the given timestamp. Returns n results
diff --git a/repository/src/main/java/org/apache/atlas/repository/audit/HBaseBasedAuditRepository.java b/repository/src/main/java/org/apache/atlas/repository/audit/HBaseBasedAuditRepository.java
index 6f4415f..5f01293 100644
--- a/repository/src/main/java/org/apache/atlas/repository/audit/HBaseBasedAuditRepository.java
+++ b/repository/src/main/java/org/apache/atlas/repository/audit/HBaseBasedAuditRepository.java
@@ -50,7 +50,10 @@ import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
+import org.apache.hadoop.hbase.filter.BinaryPrefixComparator;
+import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.PageFilter;
+import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.io.compress.Compression;
import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;
import org.apache.hadoop.hbase.regionserver.BloomType;
@@ -64,9 +67,10 @@ import javax.inject.Singleton;
import java.io.Closeable;
import java.io.IOException;
import java.util.ArrayList;
-import java.util.Iterator;
+import java.util.HashSet;
import java.util.List;
import java.util.Properties;
+import java.util.Set;
import static org.apache.atlas.EntityAuditEvent.EntityAuditAction.TAG_ADD;
import static org.apache.atlas.EntityAuditEvent.EntityAuditAction.TAG_DELETE;
@@ -546,6 +550,52 @@ public class HBaseBasedAuditRepository extends AbstractStorageBasedAuditReposito
}
@Override
+ public Set<String> getEntitiesWithTagChanges(long fromTimestamp, long toTimestamp) throws AtlasBaseException {
+ final String classificationUpdatesAction = "CLASSIFICATION_";
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Listing events for fromTimestamp {}, toTimestamp {}, action {}", fromTimestamp, toTimestamp);
+ }
+
+ Table table = null;
+ ResultScanner scanner = null;
+
+ try {
+ Set<String> guids = new HashSet<>();
+
+ table = connection.getTable(tableName);
+
+ byte[] filterValue = Bytes.toBytes(classificationUpdatesAction);
+ BinaryPrefixComparator binaryPrefixComparator = new BinaryPrefixComparator(filterValue);
+ SingleColumnValueFilter filter = new SingleColumnValueFilter(COLUMN_FAMILY, COLUMN_ACTION, CompareFilter.CompareOp.EQUAL, binaryPrefixComparator);
+ Scan scan = new Scan().setFilter(filter).setTimeRange(fromTimestamp, toTimestamp);
+
+ Result result;
+ scanner = table.getScanner(scan);
+ while ((result = scanner.next()) != null) {
+ EntityAuditEvent event = fromKey(result.getRow());
+
+ if (event == null) {
+ continue;
+ }
+
+ guids.add(event.getEntityId());
+ }
+
+ return guids;
+ } catch (IOException e) {
+ throw new AtlasBaseException(e);
+ } finally {
+ try {
+ close(scanner);
+ close(table);
+ } catch (AtlasException e) {
+ throw new AtlasBaseException(e);
+ }
+ }
+ }
+
+ @Override
public void start() throws AtlasException {
Configuration configuration = ApplicationProperties.get();
startInternal(configuration, getHBaseConfiguration(configuration));
diff --git a/repository/src/main/java/org/apache/atlas/repository/audit/InMemoryEntityAuditRepository.java b/repository/src/main/java/org/apache/atlas/repository/audit/InMemoryEntityAuditRepository.java
index dca3b85..ad6ec94 100644
--- a/repository/src/main/java/org/apache/atlas/repository/audit/InMemoryEntityAuditRepository.java
+++ b/repository/src/main/java/org/apache/atlas/repository/audit/InMemoryEntityAuditRepository.java
@@ -21,6 +21,7 @@ package org.apache.atlas.repository.audit;
import org.apache.atlas.AtlasException;
import org.apache.atlas.EntityAuditEvent;
import org.apache.atlas.annotation.ConditionalOnAtlasProperty;
+import org.apache.atlas.exception.AtlasBaseException;
import org.apache.atlas.model.audit.EntityAuditEventV2;
import org.apache.commons.collections.CollectionUtils;
import org.springframework.stereotype.Component;
@@ -28,7 +29,9 @@ import org.springframework.stereotype.Component;
import javax.inject.Singleton;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.HashSet;
import java.util.List;
+import java.util.Set;
import java.util.SortedMap;
import java.util.TreeMap;
@@ -119,6 +122,20 @@ public class InMemoryEntityAuditRepository implements EntityAuditRepository {
}
@Override
+ public Set<String> getEntitiesWithTagChanges(long fromTimestamp, long toTimestamp) throws AtlasBaseException {
+ Set<String> events = new HashSet<>();
+
+ for (EntityAuditEventV2 event : auditEventsV2.values()) {
+ long timestamp = event.getTimestamp();
+ if (timestamp > fromTimestamp && timestamp <= toTimestamp) {
+ events.add(event.getEntityId());
+ }
+ }
+
+ return events;
+ }
+
+ @Override
public List<Object> listEvents(String entityId, String startKey, short maxResults) {
List events = listEventsV2(entityId, startKey, maxResults);
diff --git a/repository/src/main/java/org/apache/atlas/repository/audit/NoopEntityAuditRepository.java b/repository/src/main/java/org/apache/atlas/repository/audit/NoopEntityAuditRepository.java
index e3a6078..4bb68d5 100644
--- a/repository/src/main/java/org/apache/atlas/repository/audit/NoopEntityAuditRepository.java
+++ b/repository/src/main/java/org/apache/atlas/repository/audit/NoopEntityAuditRepository.java
@@ -20,12 +20,14 @@ package org.apache.atlas.repository.audit;
import org.apache.atlas.EntityAuditEvent;
import org.apache.atlas.annotation.ConditionalOnAtlasProperty;
+import org.apache.atlas.exception.AtlasBaseException;
import org.apache.atlas.model.audit.EntityAuditEventV2;
import org.springframework.stereotype.Component;
import javax.inject.Singleton;
import java.util.Collections;
import java.util.List;
+import java.util.Set;
/**
* Implementation that completely disables the audit repository.
@@ -66,6 +68,11 @@ public class NoopEntityAuditRepository implements EntityAuditRepository {
}
@Override
+ public Set<String> getEntitiesWithTagChanges(long fromTimestamp, long toTimestamp) throws AtlasBaseException {
+ return Collections.emptySet();
+ }
+
+ @Override
public List<Object> listEvents(String entityId, String startKey, short n) {
return Collections.emptyList();
}
diff --git a/repository/src/main/java/org/apache/atlas/repository/store/graph/AtlasEntityStore.java b/repository/src/main/java/org/apache/atlas/repository/store/graph/AtlasEntityStore.java
index 750fa17..1da1138 100644
--- a/repository/src/main/java/org/apache/atlas/repository/store/graph/AtlasEntityStore.java
+++ b/repository/src/main/java/org/apache/atlas/repository/store/graph/AtlasEntityStore.java
@@ -24,6 +24,7 @@ import org.apache.atlas.model.instance.AtlasClassification;
import org.apache.atlas.model.instance.AtlasEntity.AtlasEntitiesWithExtInfo;
import org.apache.atlas.model.instance.AtlasEntity.AtlasEntityWithExtInfo;
import org.apache.atlas.model.instance.AtlasEntityHeader;
+import org.apache.atlas.model.instance.AtlasEntityHeaders;
import org.apache.atlas.model.instance.AtlasObjectId;
import org.apache.atlas.model.instance.EntityMutationResponse;
import org.apache.atlas.repository.store.graph.v2.EntityStream;
@@ -216,4 +217,6 @@ public interface AtlasEntityStore {
List<AtlasClassification> getClassifications(String guid) throws AtlasBaseException;
AtlasClassification getClassification(String guid, String classificationName) throws AtlasBaseException;
+
+ String setClassifications(AtlasEntityHeaders entityHeaders);
}
diff --git a/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/AtlasEntityStoreV2.java b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/AtlasEntityStoreV2.java
index a622fb5..18eb337 100644
--- a/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/AtlasEntityStoreV2.java
+++ b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/AtlasEntityStoreV2.java
@@ -687,6 +687,13 @@ public class AtlasEntityStoreV2 implements AtlasEntityStore {
return ret;
}
+ @Override
+ @GraphTransaction
+ public String setClassifications(AtlasEntityHeaders entityHeaders) {
+ ClassificationAssociator.Updater associator = new ClassificationAssociator.Updater(typeRegistry, this);
+ return associator.setClassifications(entityHeaders.getGuidHeaderMap());
+ }
+
private EntityMutationResponse createOrUpdate(EntityStream entityStream, boolean isPartialUpdate, boolean replaceClassifications) throws AtlasBaseException {
if (LOG.isDebugEnabled()) {
LOG.debug("==> createOrUpdate()");
diff --git a/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/ClassificationAssociator.java b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/ClassificationAssociator.java
new file mode 100644
index 0000000..11d008f
--- /dev/null
+++ b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/ClassificationAssociator.java
@@ -0,0 +1,316 @@
+/**
+ * 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.atlas.repository.store.graph.v2;
+
+import org.apache.atlas.exception.AtlasBaseException;
+import org.apache.atlas.model.instance.AtlasClassification;
+import org.apache.atlas.model.instance.AtlasEntityHeader;
+import org.apache.atlas.model.instance.AtlasEntityHeaders;
+import org.apache.atlas.repository.audit.EntityAuditRepository;
+import org.apache.atlas.repository.graphdb.AtlasVertex;
+import org.apache.atlas.repository.store.graph.AtlasEntityStore;
+import org.apache.atlas.type.AtlasEntityType;
+import org.apache.atlas.type.AtlasTypeRegistry;
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.collections.MapUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Component;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+@Component
+public class ClassificationAssociator {
+ private static final Logger LOG = LoggerFactory.getLogger(ClassificationAssociator.class);
+
+ public static class Retriever {
+ private final EntityAuditRepository auditRepository;
+ private final EntityGraphRetriever entityRetriever;
+
+ public Retriever(AtlasTypeRegistry typeRegistry, EntityAuditRepository auditRepository) {
+ this.entityRetriever = new EntityGraphRetriever(typeRegistry);
+ this.auditRepository = auditRepository;
+ }
+
+ Retriever(EntityGraphRetriever entityGraphRetriever, EntityAuditRepository auditRepository) {
+ this.entityRetriever = entityGraphRetriever;
+ this.auditRepository = auditRepository;
+ }
+
+ public AtlasEntityHeaders get(long fromTimestamp, long toTimestamp) throws AtlasBaseException {
+ toTimestamp = incrementTimestamp(toTimestamp);
+ Set<String> guids = auditRepository.getEntitiesWithTagChanges(fromTimestamp, toTimestamp);
+
+ Map<String, AtlasEntityHeader> guidEntityHeaderMap = new HashMap<>();
+ for (String guid : guids) {
+ AtlasEntityHeader entityHeader = entityRetriever.toAtlasEntityHeaderWithClassifications(guid);
+
+ guidEntityHeaderMap.put(guid, entityHeader);
+ }
+
+ guids.clear();
+ return new AtlasEntityHeaders(guidEntityHeaderMap);
+ }
+
+ private long incrementTimestamp(long t) {
+ return t + 1;
+ }
+ }
+
+ public static class Updater {
+ static final String ATTR_NAME_QUALIFIED_NAME = "qualifiedName";
+ static final String STATUS_DONE = "(Done)";
+ static final String STATUS_SKIPPED = "(Skipped)";
+ static final String STATUS_PARTIAL = "(Partial)";
+
+ private static final String PROCESS_FORMAT = "%s:%s:%s:%s -> %s:%s";
+ static final String PROCESS_ADD = "Add";
+ static final String PROCESS_UPDATE = "Update";
+ static final String PROCESS_DELETE = "Delete";
+ static final String JSONIFY_STRING_FORMAT = "\"%s\",";
+
+ private final AtlasTypeRegistry typeRegistry;
+ private final AtlasEntityStore entitiesStore;
+ private final EntityGraphRetriever entityRetriever;
+ private StringBuilder actionSummary = new StringBuilder();
+
+ public Updater(AtlasTypeRegistry typeRegistry, AtlasEntityStore entitiesStore) {
+ this.typeRegistry = typeRegistry;
+ this.entitiesStore = entitiesStore;
+ entityRetriever = new EntityGraphRetriever(typeRegistry);
+ }
+
+ public String setClassifications(Map<String, AtlasEntityHeader> map) {
+ for (AtlasEntityHeader incomingEntityHeader : map.values()) {
+ String typeName = incomingEntityHeader.getTypeName();
+ AtlasEntityType entityType = typeRegistry.getEntityTypeByName(typeName);
+ if (entityType == null) {
+ LOG.warn("Entity type: {}: Not found: {}!", typeName, STATUS_SKIPPED);
+ summarizeFormat("%s: %s", typeName, STATUS_SKIPPED);
+ continue;
+ }
+
+ String qualifiedName = getQualifiedName(incomingEntityHeader);
+ AtlasEntityHeader entityToBeChanged = getByUniqueAttributes(entityType, qualifiedName, incomingEntityHeader.getAttributes());
+ if (entityToBeChanged == null) {
+ summarizeFormat("Entity:%s:%s:[Not found]:%s", entityType.getTypeName(), qualifiedName, STATUS_SKIPPED);
+ continue;
+ }
+
+
+ String guid = entityToBeChanged.getGuid();
+ Map<String, List<AtlasClassification>> operationListMap = computeChanges(incomingEntityHeader, entityToBeChanged);
+ commitChanges(guid, typeName, qualifiedName, operationListMap);
+ }
+
+ return getJsonArray(actionSummary);
+ }
+
+ private void commitChanges(String entityGuid, String typeName, String qualifiedName,
+ Map<String, List<AtlasClassification>> operationListMap) {
+ if (MapUtils.isEmpty(operationListMap)) {
+ return;
+ }
+
+ deleteClassifications(entityGuid, typeName, qualifiedName, operationListMap.get(PROCESS_DELETE));
+ updateClassifications(entityGuid, typeName, qualifiedName, operationListMap.get(PROCESS_UPDATE));
+ addClassifications(entityGuid, typeName, qualifiedName, operationListMap.get(PROCESS_ADD));
+
+ operationListMap.clear();
+ }
+
+ private Map<String, List<AtlasClassification>> computeChanges(AtlasEntityHeader incomingEntityHeader, AtlasEntityHeader entityToBeUpdated) {
+ if (incomingEntityHeader == null || entityToBeUpdated == null) {
+ return null;
+ }
+
+ ListOps<AtlasClassification> listOps = new ListOps<>();
+ List<AtlasClassification> incomingClassifications = listOps.filter(incomingEntityHeader.getGuid(), incomingEntityHeader.getClassifications());
+ List<AtlasClassification> entityClassifications = listOps.filter(entityToBeUpdated.getGuid(), entityToBeUpdated.getClassifications());
+
+ if (CollectionUtils.isEmpty(incomingClassifications) && CollectionUtils.isEmpty(entityClassifications)) {
+ return null;
+ }
+
+ Map<String, List<AtlasClassification>> operationListMap = new HashMap<>();
+
+ bucket(PROCESS_DELETE, operationListMap, listOps.subtract(entityClassifications, incomingClassifications));
+ bucket(PROCESS_UPDATE, operationListMap, listOps.intersect(incomingClassifications, entityClassifications));
+ bucket(PROCESS_ADD, operationListMap, listOps.subtract(incomingClassifications, entityClassifications));
+
+ return operationListMap;
+ }
+
+ private void bucket(String op, Map<String, List<AtlasClassification>> operationListMap, List<AtlasClassification> results) {
+ if (CollectionUtils.isEmpty(results)) {
+ return;
+ }
+
+ operationListMap.put(op, results);
+ }
+
+ private void addClassifications(String entityToBeChangedGuid, String typeName, String qualifiedName, List<AtlasClassification> list) {
+ if (CollectionUtils.isEmpty(list)) {
+ return;
+ }
+
+ String status = STATUS_DONE;
+ String classificationNames = getClassificationNames(list);
+ try {
+ entitiesStore.addClassifications(entityToBeChangedGuid, list);
+ } catch (AtlasBaseException e) {
+ status = STATUS_PARTIAL;
+ LOG.warn("{}:{}:{} -> {}: {}.", PROCESS_UPDATE, typeName, qualifiedName, classificationNames, status);
+ }
+
+ summarize(PROCESS_ADD, entityToBeChangedGuid, typeName, qualifiedName, classificationNames, status);
+ }
+
+ private void updateClassifications(String entityToBeChangedGuid, String typeName, String qualifiedName, List<AtlasClassification> list) {
+ if (CollectionUtils.isEmpty(list)) {
+ return;
+ }
+
+ String status = STATUS_DONE;
+ String classificationNames = getClassificationNames(list);
+
+ try {
+ entitiesStore.updateClassifications(entityToBeChangedGuid, list);
+ } catch (AtlasBaseException e) {
+ status = STATUS_PARTIAL;
+ LOG.warn("{}:{}:{} -> {}: {}.", PROCESS_UPDATE, typeName, qualifiedName, classificationNames, status);
+ }
+
+ summarize(PROCESS_UPDATE, entityToBeChangedGuid, typeName, qualifiedName, classificationNames, status);
+ }
+
+ private void deleteClassifications(String typeName, String entityGuid, String qualifiedName, List<AtlasClassification> list) {
+ if (CollectionUtils.isEmpty(list)) {
+ return;
+ }
+
+ String status = STATUS_DONE;
+ String classificationTypeName = getClassificationNames(list);
+ for (AtlasClassification c : list) {
+ try {
+ entitiesStore.deleteClassification(entityGuid, c.getTypeName());
+ } catch (AtlasBaseException e) {
+ status = STATUS_PARTIAL;
+ LOG.warn("{}:{}:{} -> {}: Skipped!", entityGuid, typeName, qualifiedName, c.getTypeName());
+ }
+ }
+
+ summarize(PROCESS_DELETE, entityGuid, typeName, qualifiedName, classificationTypeName, status);
+ }
+
+ AtlasEntityHeader getByUniqueAttributes(AtlasEntityType entityType, String qualifiedName, Map<String, Object> attrValues) {
+ try {
+ AtlasVertex vertex = AtlasGraphUtilsV2.findByUniqueAttributes(entityType, attrValues);
+ if (vertex == null) {
+ return null;
+ }
+
+ return entityRetriever.toAtlasEntityHeaderWithClassifications(vertex);
+ } catch (AtlasBaseException e) {
+ LOG.warn("{}:{} could not be processed!", entityType, qualifiedName);
+ return null;
+ } catch (Exception ex) {
+ LOG.error("{}:{} could not be processed!", entityType, qualifiedName, ex);
+ return null;
+ }
+ }
+
+ private String getClassificationNames(List<AtlasClassification> list) {
+ return list.stream().map(AtlasClassification::getTypeName).collect(Collectors.joining(", "));
+ }
+
+ private String getQualifiedName(AtlasEntityHeader entityHeader) {
+ return (String) entityHeader.getAttribute(ATTR_NAME_QUALIFIED_NAME);
+ }
+
+ private void summarize(String... s) {
+ summarizeFormat(PROCESS_FORMAT, s);
+ }
+
+ private void summarizeFormat(String format, String... s) {
+ summarize(String.format(format, s));
+ }
+
+ private void summarize(String s) {
+ actionSummary.append(String.format(JSONIFY_STRING_FORMAT, s));
+ }
+
+ private String getJsonArray(StringBuilder actionSummary) {
+ return "[" + StringUtils.removeEnd(actionSummary.toString(), ",") + "]";
+ }
+ }
+
+ private static class ListOps<V extends AtlasClassification> {
+ public List<V> intersect(List<V> lhs, List<V> rhs) {
+ if (CollectionUtils.isEmpty(rhs)) {
+ return null;
+ }
+
+ List<V> result = new ArrayList<>();
+ for (V c : rhs) {
+ V found = findFrom(lhs, c);
+ if (found != null) {
+ result.add(found);
+ }
+ }
+
+ return result;
+ }
+
+ public List<V> subtract(List<V> lhs, List<V> rhs) {
+ if (CollectionUtils.isEmpty(lhs)) {
+ return null;
+ }
+
+ List<V> result = new ArrayList<>();
+ for (V c : lhs) {
+ V found = findFrom(rhs, c);
+ if (found == null) {
+ result.add(c);
+ }
+ }
+
+ return result;
+ }
+
+ private V findFrom(List<V> reference, V check) {
+ return (V) CollectionUtils.find(reference, ox ->
+ ((V) ox).getTypeName().equals(check.getTypeName()));
+ }
+
+ public List<V> filter(String guid, List<V> list) {
+ if (CollectionUtils.isEmpty(list)) {
+ return list;
+ }
+
+ return list.stream().filter(x -> x.getEntityGuid().equals(guid)).collect(Collectors.toList());
+ }
+ }
+}
diff --git a/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/EntityGraphRetriever.java b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/EntityGraphRetriever.java
index 477e98f..d87d1ae 100644
--- a/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/EntityGraphRetriever.java
+++ b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/EntityGraphRetriever.java
@@ -97,7 +97,7 @@ import static org.apache.atlas.type.AtlasStructType.AtlasAttribute.AtlasRelation
import static org.apache.atlas.type.AtlasStructType.AtlasAttribute.AtlasRelationshipEdgeDirection.OUT;
@Component
-public final class EntityGraphRetriever {
+public class EntityGraphRetriever {
private static final Logger LOG = LoggerFactory.getLogger(EntityGraphRetriever.class);
private static final String TERM_RELATION_NAME = "AtlasGlossarySemanticAssignment";
diff --git a/repository/src/test/java/org/apache/atlas/repository/store/graph/v2/ClassificationAssociatorTest.java b/repository/src/test/java/org/apache/atlas/repository/store/graph/v2/ClassificationAssociatorTest.java
new file mode 100644
index 0000000..ab5bb2b
--- /dev/null
+++ b/repository/src/test/java/org/apache/atlas/repository/store/graph/v2/ClassificationAssociatorTest.java
@@ -0,0 +1,235 @@
+/**
+ * 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.atlas.repository.store.graph.v2;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import org.apache.atlas.model.instance.AtlasEntityHeader;
+import org.apache.atlas.model.instance.AtlasEntityHeaders;
+import org.apache.atlas.repository.audit.EntityAuditRepository;
+import org.apache.atlas.repository.store.graph.AtlasEntityStore;
+import org.apache.atlas.type.AtlasEntityType;
+import org.apache.atlas.type.AtlasTypeRegistry;
+import org.apache.atlas.utils.AtlasJson;
+import org.apache.atlas.utils.TestResourceFileUtils;
+import org.apache.commons.lang.StringUtils;
+import org.testng.annotations.Test;
+
+import java.io.IOException;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+import static org.apache.atlas.repository.store.graph.v2.ClassificationAssociator.Updater.PROCESS_ADD;
+import static org.apache.atlas.repository.store.graph.v2.ClassificationAssociator.Updater.PROCESS_DELETE;
+import static org.apache.atlas.repository.store.graph.v2.ClassificationAssociator.Updater.PROCESS_UPDATE;
+import static org.apache.atlas.repository.store.graph.v2.ClassificationAssociator.Updater.STATUS_DONE;
+import static org.apache.atlas.repository.store.graph.v2.ClassificationAssociator.Updater.STATUS_SKIPPED;
+import static org.mockito.Matchers.anyLong;
+import static org.mockito.Matchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertTrue;
+import static org.testng.FileAssert.fail;
+
+public class ClassificationAssociatorTest {
+ private static final String TABLE_GUID = "df122fc3-5555-40f8-a30f-3090b8a622f8";
+ private static String TEST_FILES_SUBDIR = "classification-association";
+ private static String MESSAGE_SEPARATOR = ":";
+ private static String ENTITY_NAME_SEPARATOR = "->";
+
+ private class ClassificationAssociatorUpdaterForSpy extends ClassificationAssociator.Updater {
+ private final String entityFileName;
+
+ public ClassificationAssociatorUpdaterForSpy(AtlasTypeRegistry typeRegistry, AtlasEntityStore entitiesStore) {
+ super(typeRegistry, entitiesStore);
+ this.entityFileName = StringUtils.EMPTY;
+ }
+
+ public ClassificationAssociatorUpdaterForSpy(AtlasTypeRegistry typeRegistry, AtlasEntityStore entitiesStore, String entityFileName) {
+ super(typeRegistry, entitiesStore);
+ this.entityFileName = entityFileName;
+ }
+
+ @Override
+ AtlasEntityHeader getByUniqueAttributes(AtlasEntityType entityType, String qualifiedName, Map<String, Object> attrValues) {
+ try {
+ if (StringUtils.isEmpty(entityFileName)) {
+ return null;
+ }
+
+ return getEntityHeaderFromFile(entityFileName);
+ } catch (IOException e) {
+ fail(entityFileName + " could not be loaded.");
+ return null;
+ }
+ }
+ }
+
+ @Test
+ public void auditScanYieldsNothing_EmptyHeadersReturned() {
+ AtlasEntityHeaders actualEntityHeaders = setupRetriever("header-empty", 0, 0, null);
+
+ assertNotNull(actualEntityHeaders);
+ assertEquals(actualEntityHeaders.getGuidHeaderMap().size(),0);
+ }
+
+ @Test
+ public void auditScanYieldsOneEntity_EntityHeadersHasOneElementWithClassification() {
+ AtlasEntityHeaders actualEntityHeaders = setupRetriever("header-Tx", 0, 0, TABLE_GUID);
+
+ assertNotNull(actualEntityHeaders);
+ assertEquals(actualEntityHeaders.getGuidHeaderMap().size(), 1);
+ assertTrue(actualEntityHeaders.getGuidHeaderMap().containsKey(TABLE_GUID));
+ assertEquals(actualEntityHeaders.getGuidHeaderMap().get(TABLE_GUID).getGuid(), TABLE_GUID);
+ assertNotNull(actualEntityHeaders.getGuidHeaderMap().get(TABLE_GUID).getClassifications());
+ assertEquals(actualEntityHeaders.getGuidHeaderMap().get(TABLE_GUID).getClassifications().size(), 1);
+ }
+
+ private AtlasEntityHeaders setupRetriever(String headersFile, int fromTimestamp, int toTimestamp, final String tableGuid) {
+ AtlasEntityHeader entityHeaderWithClassification = null;
+ try {
+ Set<String> guids = new HashSet<String>();
+ entityHeaderWithClassification = TestResourceFileUtils.readObjectFromJson(TEST_FILES_SUBDIR, headersFile, AtlasEntityHeader.class);
+ if (!StringUtils.isEmpty(tableGuid)) {
+ guids.add(tableGuid);
+ }
+
+ EntityAuditRepository auditRepository = mock(EntityAuditRepository.class);
+ when(auditRepository.getEntitiesWithTagChanges(anyLong(), anyLong())).thenReturn(guids);
+
+ EntityGraphRetriever entityGraphRetriever = mock(EntityGraphRetriever.class);
+ when(entityGraphRetriever.toAtlasEntityHeaderWithClassifications(TABLE_GUID)).thenReturn(entityHeaderWithClassification);
+
+ ClassificationAssociator.Retriever retriever = new ClassificationAssociator.Retriever(entityGraphRetriever, auditRepository);
+ return retriever.get(fromTimestamp, toTimestamp);
+ }
+ catch (Exception ex) {
+ fail("Exception!");
+ return null;
+ }
+ }
+
+ @Test
+ public void updaterIncorrectType_ReturnsError() throws IOException {
+ AtlasEntityHeaders entityHeaderMap = getEntityHeaderMapFromFile("header-PII");
+ AtlasEntityStore entitiesStore = mock(AtlasEntityStore.class);
+
+ AtlasTypeRegistry typeRegistry = mock(AtlasTypeRegistry.class);
+ when(typeRegistry.getEntityTypeByName(anyString())).thenReturn(null);
+
+ ClassificationAssociator.Updater updater = new ClassificationAssociator.Updater(typeRegistry, entitiesStore);
+ String summary = updater.setClassifications(entityHeaderMap.getGuidHeaderMap());
+
+ assertTrue(summary.contains("hive_"));
+ assertTrue(summary.contains(STATUS_SKIPPED));
+ }
+
+ @Test
+ public void updaterCorrectTypeEntityNotFound_Skipped() throws IOException {
+ AtlasEntityHeaders entityHeaderMap = getEntityHeaderMapFromFile("header-PII");
+ AtlasEntityType hiveTable = mock(AtlasEntityType.class);
+ AtlasEntityStore entitiesStore = mock(AtlasEntityStore.class);
+ AtlasTypeRegistry typeRegistry = mock(AtlasTypeRegistry.class);
+
+ when(typeRegistry.getEntityTypeByName(anyString())).thenReturn(hiveTable);
+ when(hiveTable.getTypeName()).thenReturn("hive_column");
+
+ ClassificationAssociatorUpdaterForSpy updater = new ClassificationAssociatorUpdaterForSpy(typeRegistry, entitiesStore);
+ String summary = updater.setClassifications(entityHeaderMap.getGuidHeaderMap());
+
+ TypeReference<String[]> typeReference = new TypeReference<String[]>() {};
+ String[] summaryArray = AtlasJson.fromJson(summary, typeReference);
+ assertEquals(summaryArray.length, 1);
+ assertSummaryElement(summaryArray[0], "Entity", STATUS_SKIPPED, "");
+ }
+
+ @Test
+ public void updaterTests() throws IOException {
+ updaterAssert("header-PII", "col-entity-None", PROCESS_ADD + ":PII");
+ updaterAssert("header-PII", "col-entity-PII", new String[]{PROCESS_UPDATE + ":PII"});
+ updaterAssert("header-None", "col-entity-PII", new String[]{PROCESS_DELETE + ":PII"});
+ updaterAssert("header-PII-VENDOR_PII", "col-entity-PII-FIN_PII",
+ PROCESS_DELETE + ":FIN_PII",
+ PROCESS_UPDATE + ":PII",
+ PROCESS_ADD + ":VENDOR_PII");
+ updaterAssert("header-None", "col-entity-None", new String[]{});
+ updaterAssert("header-FIN_PII", "col-entity-PII",
+ PROCESS_DELETE + ":PII",
+ PROCESS_ADD + ":FIN_PII");
+ }
+
+ @Test
+ public void updater_filterPropagatedClassifications() throws IOException {
+ updaterAssert("header-Tx-prop-T1", "col-entity-T1-prop-Tn",
+ PROCESS_DELETE + ":T1",
+ PROCESS_ADD + ":Tx");
+ }
+
+
+ private void assertSummaryElement(String summaryElement, String operation, String status, String classificationName) {
+ String[] splits = StringUtils.split(summaryElement, MESSAGE_SEPARATOR);
+ String[] nameSplits = StringUtils.split(splits[3], ENTITY_NAME_SEPARATOR);
+ if (nameSplits.length > 1) {
+ assertEquals(nameSplits[1].trim(), classificationName);
+ }
+
+ assertEquals(splits[0], operation);
+ assertEquals(splits[4], status);
+ }
+
+ private String[] setupUpdater(String entityHeaderFileName, String entityFileName, int expectedSummaryLength) throws IOException {
+ AtlasEntityHeaders entityHeaderMap = getEntityHeaderMapFromFile(entityHeaderFileName);
+
+ AtlasEntityType hiveTable = mock(AtlasEntityType.class);
+ AtlasEntityStore entitiesStore = mock(AtlasEntityStore.class);
+ AtlasTypeRegistry typeRegistry = mock(AtlasTypeRegistry.class);
+
+ when(typeRegistry.getEntityTypeByName(anyString())).thenReturn(hiveTable);
+ when(hiveTable.getTypeName()).thenReturn("hive_column");
+
+ ClassificationAssociatorUpdaterForSpy updater = new ClassificationAssociatorUpdaterForSpy(typeRegistry, entitiesStore, entityFileName);
+ String summary = updater.setClassifications(entityHeaderMap.getGuidHeaderMap());
+
+ TypeReference<String[]> typeReference = new TypeReference<String[]>() {};
+ String[] summaryArray = AtlasJson.fromJson(summary, typeReference);
+ assertEquals(summaryArray.length, expectedSummaryLength);
+
+ return summaryArray;
+ }
+
+ private AtlasEntityHeader getEntityHeaderFromFile(String entityJson) throws IOException {
+ return TestResourceFileUtils.readObjectFromJson(TEST_FILES_SUBDIR, entityJson, AtlasEntityHeader.class);
+ }
+
+ private AtlasEntityHeaders getEntityHeaderMapFromFile(String filename) throws IOException {
+ return TestResourceFileUtils.readObjectFromJson(TEST_FILES_SUBDIR, filename, AtlasEntityHeaders.class);
+ }
+
+ private void updaterAssert(String incoming, String entity, String... opNamePair) throws IOException {
+ String[] summary = setupUpdater(incoming, entity, opNamePair.length);
+
+ for (int i = 0; i < opNamePair.length; i++) {
+ String s = opNamePair[i];
+ String[] splits = StringUtils.split(s, ":");
+ assertSummaryElement(summary[i], splits[0], STATUS_DONE, splits[1]);
+ }
+ }
+}
diff --git a/repository/src/test/resources/json/classification-association/col-entity-None.json b/repository/src/test/resources/json/classification-association/col-entity-None.json
new file mode 100644
index 0000000..6c04a3d
--- /dev/null
+++ b/repository/src/test/resources/json/classification-association/col-entity-None.json
@@ -0,0 +1,10 @@
+{
+ "typeName": "hive_column",
+ "attributes": {
+ "owner": "hive",
+ "qualifiedName": "hortoniabank.us_customers.nationalid@cl1",
+ "name": "nationalid"
+ },
+ "guid": "0ce68113-77fe-4ed1-9585-69371202bd74",
+ "status": "ACTIVE"
+}
diff --git a/repository/src/test/resources/json/classification-association/col-entity-PII-FIN_PII.json b/repository/src/test/resources/json/classification-association/col-entity-PII-FIN_PII.json
new file mode 100644
index 0000000..283d863
--- /dev/null
+++ b/repository/src/test/resources/json/classification-association/col-entity-PII-FIN_PII.json
@@ -0,0 +1,32 @@
+{
+ "typeName": "hive_column",
+ "attributes": {
+ "owner": "hive",
+ "qualifiedName": "hortoniabank.us_customers.nationalid@cl1",
+ "name": "nationalid"
+ },
+ "guid": "0ce68113-77fe-4ed1-9585-69371202bd74",
+ "status": "ACTIVE",
+ "classifications": [
+ {
+ "typeName": "PII",
+ "attributes": {
+ "type": "ssn"
+ },
+ "entityGuid": "0ce68113-77fe-4ed1-9585-69371202bd74",
+ "entityStatus": "ACTIVE",
+ "propagate": true,
+ "removePropagationsOnEntityDelete": false
+ },
+ {
+ "typeName": "FIN_PII",
+ "attributes": {
+ "type": "ssn"
+ },
+ "entityGuid": "0ce68113-77fe-4ed1-9585-69371202bd74",
+ "entityStatus": "ACTIVE",
+ "propagate": true,
+ "removePropagationsOnEntityDelete": false
+ }
+ ]
+}
diff --git a/repository/src/test/resources/json/classification-association/col-entity-PII.json b/repository/src/test/resources/json/classification-association/col-entity-PII.json
new file mode 100644
index 0000000..af34e7d
--- /dev/null
+++ b/repository/src/test/resources/json/classification-association/col-entity-PII.json
@@ -0,0 +1,22 @@
+{
+ "typeName": "hive_column",
+ "attributes": {
+ "owner": "hive",
+ "qualifiedName": "hortoniabank.us_customers.nationalid@cl1",
+ "name": "nationalid"
+ },
+ "guid": "0ce68113-77fe-4ed1-9585-69371202bd74",
+ "status": "ACTIVE",
+ "classifications": [
+ {
+ "typeName": "PII",
+ "attributes": {
+ "type": "ssn"
+ },
+ "entityGuid": "0ce68113-77fe-4ed1-9585-69371202bd74",
+ "entityStatus": "ACTIVE",
+ "propagate": true,
+ "removePropagationsOnEntityDelete": false
+ }
+ ]
+}
diff --git a/repository/src/test/resources/json/classification-association/col-entity-T1-prop-Tn.json b/repository/src/test/resources/json/classification-association/col-entity-T1-prop-Tn.json
new file mode 100644
index 0000000..4f9cbd2
--- /dev/null
+++ b/repository/src/test/resources/json/classification-association/col-entity-T1-prop-Tn.json
@@ -0,0 +1,34 @@
+{
+ "typeName": "hive_column",
+ "attributes": {
+ "owner": "hive",
+ "createTime": 1547071410000,
+ "qualifiedName": "stocks.daily@cl1",
+ "name": "daily"
+ },
+ "guid": "df122fc3-5555-40f8-a30f-3090b8a622f8",
+ "status": "ACTIVE",
+ "displayText": "daily",
+ "classifications": [
+ {
+ "typeName": "T1",
+ "attributes": {},
+ "entityGuid": "df122fc3-5555-40f8-a30f-3090b8a622f8",
+ "entityStatus": "ACTIVE",
+ "propagate": false,
+ "validityPeriods": [],
+ "removePropagationsOnEntityDelete": false
+ },
+ {
+ "typeName": "Tn",
+ "attributes": {},
+ "entityGuid": "22222222-5555-40f8-a30f-3090b8a622f8",
+ "entityStatus": "ACTIVE",
+ "propagate": false,
+ "validityPeriods": [],
+ "removePropagationsOnEntityDelete": false
+ }
+ ],
+ "meaningNames": [],
+ "meanings": []
+}
diff --git a/repository/src/test/resources/json/classification-association/header-FIN_PII.json b/repository/src/test/resources/json/classification-association/header-FIN_PII.json
new file mode 100644
index 0000000..35bb6db
--- /dev/null
+++ b/repository/src/test/resources/json/classification-association/header-FIN_PII.json
@@ -0,0 +1,32 @@
+{
+ "guidHeaderMap": {
+ "0ce68113-77fe-4ed1-9585-69371202bd74": {
+ "typeName": "hive_column",
+ "attributes": {
+ "owner": "hive",
+ "qualifiedName": "hortoniabank.us_customers.nationalid@cl1",
+ "name": "nationalid"
+ },
+ "guid": "0ce68113-77fe-4ed1-9585-69371202bd74",
+ "status": "ACTIVE",
+ "displayText": "nationalid",
+ "classificationNames": [
+ "FIN_PII"
+ ],
+ "classifications": [
+ {
+ "typeName": "FIN_PII",
+ "attributes": {
+ "type": "ssn"
+ },
+ "entityGuid": "0ce68113-77fe-4ed1-9585-69371202bd74",
+ "entityStatus": "ACTIVE",
+ "propagate": true,
+ "removePropagationsOnEntityDelete": false
+ }
+ ],
+ "meaningNames": [],
+ "meanings": []
+ }
+ }
+}
diff --git a/repository/src/test/resources/json/classification-association/header-None.json b/repository/src/test/resources/json/classification-association/header-None.json
new file mode 100644
index 0000000..a858990
--- /dev/null
+++ b/repository/src/test/resources/json/classification-association/header-None.json
@@ -0,0 +1,21 @@
+{
+ "guidHeaderMap": {
+ "0ce68113-77fe-4ed1-9585-69371202bd74": {
+ "typeName": "hive_column",
+ "attributes": {
+ "owner": "hive",
+ "qualifiedName": "hortoniabank.us_customers.nationalid@cl1",
+ "name": "nationalid"
+ },
+ "guid": "0ce68113-77fe-4ed1-9585-69371202bd74",
+ "status": "ACTIVE",
+ "displayText": "nationalid",
+ "classificationNames": [
+ ],
+ "classifications": [
+ ],
+ "meaningNames": [],
+ "meanings": []
+ }
+ }
+}
diff --git a/repository/src/test/resources/json/classification-association/header-PII-VENDOR_PII.json b/repository/src/test/resources/json/classification-association/header-PII-VENDOR_PII.json
new file mode 100644
index 0000000..58638f7
--- /dev/null
+++ b/repository/src/test/resources/json/classification-association/header-PII-VENDOR_PII.json
@@ -0,0 +1,42 @@
+{
+ "guidHeaderMap": {
+ "0ce68113-77fe-4ed1-9585-69371202bd74": {
+ "typeName": "hive_column",
+ "attributes": {
+ "owner": "hive",
+ "qualifiedName": "hortoniabank.us_customers.nationalid@cl1",
+ "name": "nationalid"
+ },
+ "guid": "0ce68113-77fe-4ed1-9585-69371202bd74",
+ "status": "ACTIVE",
+ "displayText": "nationalid",
+ "classificationNames": [
+ "PII"
+ ],
+ "classifications": [
+ {
+ "typeName": "PII",
+ "attributes": {
+ "type": "ssn"
+ },
+ "entityGuid": "0ce68113-77fe-4ed1-9585-69371202bd74",
+ "entityStatus": "ACTIVE",
+ "propagate": true,
+ "removePropagationsOnEntityDelete": false
+ },
+ {
+ "typeName": "VENDOR_PII",
+ "attributes": {
+ "type": "ssn"
+ },
+ "entityGuid": "0ce68113-77fe-4ed1-9585-69371202bd74",
+ "entityStatus": "ACTIVE",
+ "propagate": true,
+ "removePropagationsOnEntityDelete": false
+ }
+ ],
+ "meaningNames": [],
+ "meanings": []
+ }
+ }
+}
diff --git a/repository/src/test/resources/json/classification-association/header-PII.json b/repository/src/test/resources/json/classification-association/header-PII.json
new file mode 100644
index 0000000..bfc6d2e
--- /dev/null
+++ b/repository/src/test/resources/json/classification-association/header-PII.json
@@ -0,0 +1,32 @@
+{
+ "guidHeaderMap": {
+ "0ce68113-77fe-4ed1-9585-69371202bd74": {
+ "typeName": "hive_column",
+ "attributes": {
+ "owner": "hive",
+ "qualifiedName": "hortoniabank.us_customers.nationalid@cl1",
+ "name": "nationalid"
+ },
+ "guid": "0ce68113-77fe-4ed1-9585-69371202bd74",
+ "status": "ACTIVE",
+ "displayText": "nationalid",
+ "classificationNames": [
+ "PII"
+ ],
+ "classifications": [
+ {
+ "typeName": "PII",
+ "attributes": {
+ "type": "ssn"
+ },
+ "entityGuid": "0ce68113-77fe-4ed1-9585-69371202bd74",
+ "entityStatus": "ACTIVE",
+ "propagate": true,
+ "removePropagationsOnEntityDelete": false
+ }
+ ],
+ "meaningNames": [],
+ "meanings": []
+ }
+ }
+}
diff --git a/repository/src/test/resources/json/classification-association/header-Tx-prop-T1.json b/repository/src/test/resources/json/classification-association/header-Tx-prop-T1.json
new file mode 100644
index 0000000..8f2f26f
--- /dev/null
+++ b/repository/src/test/resources/json/classification-association/header-Tx-prop-T1.json
@@ -0,0 +1,42 @@
+{
+ "guidHeaderMap": {
+ "0ce68113-77fe-4ed1-9585-69371202bd74": {
+ "typeName": "hive_column",
+ "attributes": {
+ "owner": "hive",
+ "qualifiedName": "hortoniabank.us_customers.nationalid@cl1",
+ "name": "nationalid"
+ },
+ "guid": "0ce68113-77fe-4ed1-9585-69371202bd74",
+ "status": "ACTIVE",
+ "displayText": "nationalid",
+ "classificationNames": [
+ "T1", "Tx"
+ ],
+ "classifications": [
+ {
+ "typeName": "Tx",
+ "attributes": {
+ "type": "ssn"
+ },
+ "entityGuid": "0ce68113-77fe-4ed1-9585-69371202bd74",
+ "entityStatus": "ACTIVE",
+ "propagate": true,
+ "removePropagationsOnEntityDelete": false
+ },
+ {
+ "typeName": "T1",
+ "attributes": {
+ "type": "ssn"
+ },
+ "entityGuid": "22222222-77fe-4ed1-9585-69371202bd74",
+ "entityStatus": "ACTIVE",
+ "propagate": true,
+ "removePropagationsOnEntityDelete": false
+ }
+ ],
+ "meaningNames": [],
+ "meanings": []
+ }
+ }
+}
diff --git a/repository/src/test/resources/json/classification-association/header-Tx.json b/repository/src/test/resources/json/classification-association/header-Tx.json
new file mode 100644
index 0000000..bab1eaa
--- /dev/null
+++ b/repository/src/test/resources/json/classification-association/header-Tx.json
@@ -0,0 +1,26 @@
+{
+ "typeName": "hive_table",
+ "attributes": {
+ "owner": "hive",
+ "createTime": 1547071410000,
+ "qualifiedName": "stocks.daily@cl1",
+ "name": "daily"
+ },
+ "guid": "df122fc3-5555-40f8-a30f-3090b8a622f8",
+ "status": "ACTIVE",
+ "displayText": "daily",
+ "classificationNames": [],
+ "classifications": [
+ {
+ "typeName": "Tx",
+ "attributes": {},
+ "entityGuid": "df122fc3-5555-40f8-a30f-3090b8a622f8",
+ "entityStatus": "ACTIVE",
+ "propagate": false,
+ "validityPeriods": [],
+ "removePropagationsOnEntityDelete": false
+ }
+ ],
+ "meaningNames": [],
+ "meanings": []
+}
\ No newline at end of file
diff --git a/repository/src/test/resources/json/classification-association/header-empty.json b/repository/src/test/resources/json/classification-association/header-empty.json
new file mode 100644
index 0000000..9c8f417
--- /dev/null
+++ b/repository/src/test/resources/json/classification-association/header-empty.json
@@ -0,0 +1,3 @@
+{
+ "guidHeaderMap": {}
+}
\ No newline at end of file
diff --git a/webapp/src/main/java/org/apache/atlas/web/rest/EntityREST.java b/webapp/src/main/java/org/apache/atlas/web/rest/EntityREST.java
index 68c132c..713338e 100644
--- a/webapp/src/main/java/org/apache/atlas/web/rest/EntityREST.java
+++ b/webapp/src/main/java/org/apache/atlas/web/rest/EntityREST.java
@@ -26,10 +26,12 @@ import org.apache.atlas.model.instance.AtlasClassification;
import org.apache.atlas.model.instance.AtlasEntity.AtlasEntitiesWithExtInfo;
import org.apache.atlas.model.instance.AtlasEntity.AtlasEntityWithExtInfo;
import org.apache.atlas.model.instance.AtlasEntityHeader;
+import org.apache.atlas.model.instance.AtlasEntityHeaders;
import org.apache.atlas.model.instance.ClassificationAssociateRequest;
import org.apache.atlas.model.instance.EntityMutationResponse;
import org.apache.atlas.model.typedef.AtlasStructDef.AtlasAttributeDef;
import org.apache.atlas.repository.audit.EntityAuditRepository;
+import org.apache.atlas.repository.store.graph.v2.ClassificationAssociator;
import org.apache.atlas.repository.converters.AtlasInstanceConverter;
import org.apache.atlas.repository.store.graph.AtlasEntityStore;
import org.apache.atlas.repository.store.graph.v2.AtlasEntityStream;
@@ -49,7 +51,16 @@ import org.springframework.stereotype.Service;
import javax.inject.Inject;
import javax.inject.Singleton;
import javax.servlet.http.HttpServletRequest;
-import javax.ws.rs.*;
+import javax.ws.rs.Consumes;
+import javax.ws.rs.DELETE;
+import javax.ws.rs.DefaultValue;
+import javax.ws.rs.GET;
+import javax.ws.rs.POST;
+import javax.ws.rs.PUT;
+import javax.ws.rs.Path;
+import javax.ws.rs.PathParam;
+import javax.ws.rs.Produces;
+import javax.ws.rs.QueryParam;
import javax.ws.rs.core.Context;
import javax.ws.rs.core.MediaType;
import java.util.ArrayList;
@@ -77,7 +88,6 @@ public class EntityREST {
private final EntityAuditRepository auditRepository;
private final AtlasInstanceConverter instanceConverter;
-
@Inject
public EntityREST(AtlasTypeRegistry typeRegistry, AtlasEntityStore entitiesStore,
EntityAuditRepository auditRepository, AtlasInstanceConverter instanceConverter) {
@@ -601,7 +611,7 @@ public class EntityREST {
}
/**
- * Bulk API to create new entities or update existing entities in Atlas.
+ * Bulk API to create new entities or updates existing entities in Atlas.
* Existing entity is matched using its unique guid if supplied or by its unique attributes eg: qualifiedName
*/
@POST
@@ -708,6 +718,49 @@ public class EntityREST {
}
}
+ @GET
+ @Path("bulk/headers")
+ @Produces(Servlets.JSON_MEDIA_TYPE)
+ public AtlasEntityHeaders getEntityHeaders(@QueryParam("tagUpdateStartTime") long tagUpdateStartTime) throws AtlasBaseException {
+ AtlasPerfTracer perf = null;
+
+ try {
+ long tagUpdateEndTime = System.currentTimeMillis();
+
+ if (tagUpdateStartTime > tagUpdateEndTime) {
+ throw new AtlasBaseException(AtlasErrorCode.BAD_REQUEST, "fromTimestamp should be less than toTimestamp");
+ }
+
+ if (AtlasPerfTracer.isPerfTraceEnabled(PERF_LOG)) {
+ perf = AtlasPerfTracer.getPerfTracer(PERF_LOG, "EntityREST.getEntityHeaders(" + tagUpdateStartTime + ", " + tagUpdateEndTime + ")");
+ }
+
+ ClassificationAssociator.Retriever associator = new ClassificationAssociator.Retriever(typeRegistry, auditRepository);
+ return associator.get(tagUpdateStartTime, tagUpdateEndTime);
+ } finally {
+ AtlasPerfTracer.log(perf);
+ }
+ }
+
+ @POST
+ @Path("bulk/setClassifications")
+ @Produces(Servlets.JSON_MEDIA_TYPE)
+ @Consumes(Servlets.JSON_MEDIA_TYPE)
+ public String setClassifications(AtlasEntityHeaders entityHeaders) throws AtlasBaseException {
+ AtlasPerfTracer perf = null;
+
+ try {
+ if (AtlasPerfTracer.isPerfTraceEnabled(PERF_LOG)) {
+ perf = AtlasPerfTracer.getPerfTracer(PERF_LOG, "EntityREST.setClassifications()");
+ }
+
+ ClassificationAssociator.Updater associator = new ClassificationAssociator.Updater(typeRegistry, entitiesStore);
+ return associator.setClassifications(entityHeaders.getGuidHeaderMap());
+ } finally {
+ AtlasPerfTracer.log(perf);
+ }
+ }
+
private AtlasEntityType ensureEntityType(String typeName) throws AtlasBaseException {
AtlasEntityType ret = typeRegistry.getEntityTypeByName(typeName);