You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@atlas.apache.org by sh...@apache.org on 2015/12/15 11:20:47 UTC
[1/2] incubator-atlas git commit: ATLAS-386 Handle hive rename Table
(shwethags)
Repository: incubator-atlas
Updated Branches:
refs/heads/master 90692af5b -> 8cbb2c13b
ATLAS-386 Handle hive rename Table (shwethags)
Project: http://git-wip-us.apache.org/repos/asf/incubator-atlas/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-atlas/commit/dce31ab8
Tree: http://git-wip-us.apache.org/repos/asf/incubator-atlas/tree/dce31ab8
Diff: http://git-wip-us.apache.org/repos/asf/incubator-atlas/diff/dce31ab8
Branch: refs/heads/master
Commit: dce31ab89be353e8e1e23a2ce1e1e903745db084
Parents: 90692af
Author: Shwetha GS <ss...@hortonworks.com>
Authored: Tue Dec 15 15:08:44 2015 +0530
Committer: Shwetha GS <ss...@hortonworks.com>
Committed: Tue Dec 15 15:08:44 2015 +0530
----------------------------------------------------------------------
addons/hive-bridge/pom.xml | 4 +
.../atlas/hive/bridge/HiveMetaStoreBridge.java | 4 +
.../org/apache/atlas/hive/hook/HiveHook.java | 112 ++++++------
.../org/apache/atlas/hive/hook/HiveHookIT.java | 15 +-
.../main/java/org/apache/atlas/AtlasClient.java | 13 +-
docs/src/site/twiki/Bridge-Hive.twiki | 2 +-
.../apache/atlas/kafka/KafkaNotification.java | 2 +-
.../notification/AbstractNotification.java | 19 ++
.../AbstractNotificationConsumer.java | 42 ++++-
.../notification/NotificationHookConsumer.java | 41 +++--
.../notification/NotificationInterface.java | 8 +-
.../NotificationEntityChangeListener.java | 32 +---
.../notification/hook/HookNotification.java | 174 +++++++++++++++++++
.../atlas/kafka/KafkaNotificationTest.java | 20 ---
.../notification/hook/HookNotificationTest.java | 68 ++++++++
pom.xml | 9 +-
release-log.txt | 1 +
.../repository/graph/TitanGraphProvider.java | 6 -
.../atlas/services/DefaultMetadataService.java | 7 +
.../test/java/org/apache/atlas/TestUtils.java | 4 +-
.../service/DefaultMetadataServiceTest.java | 46 +++--
.../apache/atlas/typesystem/Referenceable.java | 6 +-
.../atlas/typesystem/types/ClassType.java | 6 +-
.../atlas/typesystem/types/Multiplicity.java | 4 +-
.../src/main/resources/application.properties | 6 +-
webapp/pom.xml | 4 +
.../src/main/java/org/apache/atlas/Atlas.java | 4 +
.../NotificationHookConsumerIT.java | 111 ++++++++----
.../web/resources/EntityJerseyResourceIT.java | 31 ++++
29 files changed, 600 insertions(+), 201 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/dce31ab8/addons/hive-bridge/pom.xml
----------------------------------------------------------------------
diff --git a/addons/hive-bridge/pom.xml b/addons/hive-bridge/pom.xml
index a61d11b..71d6d4d 100755
--- a/addons/hive-bridge/pom.xml
+++ b/addons/hive-bridge/pom.xml
@@ -290,6 +290,10 @@
<name>atlas.log.dir</name>
<value>${project.build.directory}/logs</value>
</systemProperty>
+ <systemProperty>
+ <name>atlas.data</name>
+ <value>${project.build.directory}/data</value>
+ </systemProperty>
</systemProperties>
<stopKey>atlas-stop</stopKey>
<stopPort>31001</stopPort>
http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/dce31ab8/addons/hive-bridge/src/main/java/org/apache/atlas/hive/bridge/HiveMetaStoreBridge.java
----------------------------------------------------------------------
diff --git a/addons/hive-bridge/src/main/java/org/apache/atlas/hive/bridge/HiveMetaStoreBridge.java b/addons/hive-bridge/src/main/java/org/apache/atlas/hive/bridge/HiveMetaStoreBridge.java
index f367317..ee5ae10 100755
--- a/addons/hive-bridge/src/main/java/org/apache/atlas/hive/bridge/HiveMetaStoreBridge.java
+++ b/addons/hive-bridge/src/main/java/org/apache/atlas/hive/bridge/HiveMetaStoreBridge.java
@@ -70,6 +70,10 @@ public class HiveMetaStoreBridge {
this(hiveConf, atlasConf, null, null);
}
+ public String getClusterName() {
+ return clusterName;
+ }
+
/**
* Construct a HiveMetaStoreBridge.
* @param hiveConf hive conf
http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/dce31ab8/addons/hive-bridge/src/main/java/org/apache/atlas/hive/hook/HiveHook.java
----------------------------------------------------------------------
diff --git a/addons/hive-bridge/src/main/java/org/apache/atlas/hive/hook/HiveHook.java b/addons/hive-bridge/src/main/java/org/apache/atlas/hive/hook/HiveHook.java
index 2f88446..37a3169 100755
--- a/addons/hive-bridge/src/main/java/org/apache/atlas/hive/hook/HiveHook.java
+++ b/addons/hive-bridge/src/main/java/org/apache/atlas/hive/hook/HiveHook.java
@@ -25,11 +25,12 @@ import com.google.inject.Inject;
import com.google.inject.Injector;
import org.apache.atlas.ApplicationProperties;
import org.apache.atlas.hive.bridge.HiveMetaStoreBridge;
+import org.apache.atlas.hive.model.HiveDataModelGenerator;
import org.apache.atlas.hive.model.HiveDataTypes;
import org.apache.atlas.notification.NotificationInterface;
import org.apache.atlas.notification.NotificationModule;
+import org.apache.atlas.notification.hook.HookNotification;
import org.apache.atlas.typesystem.Referenceable;
-import org.apache.atlas.typesystem.json.InstanceSerialization;
import org.apache.commons.configuration.Configuration;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.hive.conf.HiveConf;
@@ -47,16 +48,12 @@ import org.apache.hadoop.hive.ql.metadata.Partition;
import org.apache.hadoop.hive.ql.metadata.Table;
import org.apache.hadoop.hive.ql.plan.HiveOperation;
import org.apache.hadoop.security.UserGroupInformation;
-import org.codehaus.jettison.json.JSONArray;
import org.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
-import java.util.Collection;
-import java.util.LinkedHashMap;
import java.util.List;
-import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
@@ -109,6 +106,8 @@ public class HiveHook implements ExecuteWithHookContext {
@Inject
private static NotificationInterface notifInterface;
+ private List<HookNotification.HookNotificationMessage> messages = new ArrayList<>();
+
private static final HiveConf hiveConf;
static {
@@ -233,37 +232,51 @@ public class HiveHook implements ExecuteWithHookContext {
default:
}
+
+ notifyAtlas();
}
- //todo re-write with notification
private void renameTable(HiveMetaStoreBridge dgiBridge, HiveEvent event) throws Exception {
//crappy, no easy of getting new name
assert event.inputs != null && event.inputs.size() == 1;
assert event.outputs != null && event.outputs.size() > 0;
- Table oldTable = event.inputs.iterator().next().getTable();
- Table newTable = null;
+ //Update entity if not exists
+ ReadEntity oldEntity = event.inputs.iterator().next();
+ Table oldTable = oldEntity.getTable();
+
for (WriteEntity writeEntity : event.outputs) {
if (writeEntity.getType() == Entity.Type.TABLE) {
- Table table = writeEntity.getTable();
- if (table.getDbName().equals(oldTable.getDbName()) && !table.getTableName()
+ Table newTable = writeEntity.getTable();
+ if (newTable.getDbName().equals(oldTable.getDbName()) && !newTable.getTableName()
.equals(oldTable.getTableName())) {
- newTable = table;
- break;
+
+ //Create/update old table entity - create new entity and replace id
+ Referenceable tableEntity = createEntities(dgiBridge, writeEntity);
+ String oldQualifiedName = dgiBridge.getTableQualifiedName(dgiBridge.getClusterName(),
+ oldTable.getDbName(), oldTable.getTableName());
+ tableEntity.set(HiveDataModelGenerator.NAME, oldQualifiedName);
+ tableEntity.set(HiveDataModelGenerator.TABLE_NAME, oldTable.getTableName().toLowerCase());
+
+
+ String newQualifiedName = dgiBridge.getTableQualifiedName(dgiBridge.getClusterName(),
+ newTable.getDbName(), newTable.getTableName());
+
+ Referenceable newEntity = new Referenceable(HiveDataTypes.HIVE_TABLE.getName());
+ newEntity.set(HiveDataModelGenerator.NAME, newQualifiedName);
+ newEntity.set(HiveDataModelGenerator.TABLE_NAME, newTable.getTableName().toLowerCase());
+ messages.add(new HookNotification.EntityPartialUpdateRequest(HiveDataTypes.HIVE_TABLE.getName(),
+ HiveDataModelGenerator.NAME, oldQualifiedName, newEntity));
}
}
}
- if (newTable == null) {
- LOG.warn("Failed to deduct new name for " + event.queryStr);
- return;
- }
}
- private Map<Type, Referenceable> createEntities(HiveMetaStoreBridge dgiBridge, Entity entity) throws Exception {
- Map<Type, Referenceable> entities = new LinkedHashMap<>();
+ private Referenceable createEntities(HiveMetaStoreBridge dgiBridge, Entity entity) throws Exception {
Database db = null;
Table table = null;
Partition partition = null;
+ List<Referenceable> entities = new ArrayList<>();
switch (entity.getType()) {
case DATABASE:
@@ -283,64 +296,54 @@ public class HiveHook implements ExecuteWithHookContext {
}
db = dgiBridge.hiveClient.getDatabase(db.getName());
- Referenceable dbReferenceable = dgiBridge.createDBInstance(db);
- entities.put(Type.DATABASE, dbReferenceable);
+ Referenceable dbEntity = dgiBridge.createDBInstance(db);
+ entities.add(dbEntity);
- Referenceable tableReferenceable = null;
+ Referenceable tableEntity = null;
if (table != null) {
table = dgiBridge.hiveClient.getTable(table.getDbName(), table.getTableName());
- tableReferenceable = dgiBridge.createTableInstance(dbReferenceable, table);
- entities.put(Type.TABLE, tableReferenceable);
+ tableEntity = dgiBridge.createTableInstance(dbEntity, table);
+ entities.add(tableEntity);
}
if (partition != null) {
- Referenceable partitionReferenceable = dgiBridge.createPartitionReferenceable(tableReferenceable,
- (Referenceable) tableReferenceable.get("sd"), partition);
- entities.put(Type.PARTITION, partitionReferenceable);
+ Referenceable partitionEntity = dgiBridge.createPartitionReferenceable(tableEntity,
+ (Referenceable) tableEntity.get("sd"), partition);
+ entities.add(partitionEntity);
}
- return entities;
+
+ messages.add(new HookNotification.EntityUpdateRequest(entities));
+ return tableEntity;
}
private void handleEventOutputs(HiveMetaStoreBridge dgiBridge, HiveEvent event, Type entityType) throws Exception {
- List<Referenceable> entities = new ArrayList<>();
for (WriteEntity entity : event.outputs) {
if (entity.getType() == entityType) {
- entities.addAll(createEntities(dgiBridge, entity).values());
+ createEntities(dgiBridge, entity);
}
}
- notifyEntity(entities);
- }
-
- private void notifyEntity(Collection<Referenceable> entities) {
- JSONArray entitiesArray = new JSONArray();
- for (Referenceable entity : entities) {
- String entityJson = InstanceSerialization.toJson(entity, true);
- entitiesArray.put(entityJson);
- }
- notifyEntity(entitiesArray);
}
/**
* Notify atlas of the entity through message. The entity can be a complex entity with reference to other entities.
* De-duping of entities is done on server side depending on the unique attribute on the
- * @param entities
*/
- private void notifyEntity(JSONArray entities) {
+ private void notifyAtlas() {
int maxRetries = atlasProperties.getInt(HOOK_NUM_RETRIES, 3);
- String message = entities.toString();
+ LOG.debug("Notifying atlas with messages {}", messages);
int numRetries = 0;
while (true) {
try {
- notifInterface.send(NotificationInterface.NotificationType.HOOK, message);
- return;
+ notifInterface.send(NotificationInterface.NotificationType.HOOK, messages);
+ break;
} catch(Exception e) {
numRetries++;
if(numRetries < maxRetries) {
- LOG.debug("Failed to notify atlas for entity {}. Retrying", message, e);
+ LOG.debug("Failed to notify atlas. Retrying", e);
} else {
- LOG.error("Failed to notify atlas for entity {} after {} retries. Quitting", message,
- maxRetries, e);
+ LOG.error("Failed to notify atlas after {} retries. Quitting", maxRetries, e);
+ break;
}
}
}
@@ -369,7 +372,7 @@ public class HiveHook implements ExecuteWithHookContext {
String queryStr = normalize(event.queryStr);
LOG.debug("Registering query: {}", queryStr);
- List<Referenceable> entities = new ArrayList<>();
+
Referenceable processReferenceable = new Referenceable(HiveDataTypes.HIVE_PROCESS.getName());
processReferenceable.set("name", queryStr);
processReferenceable.set("operationType", event.operation.getOperationName());
@@ -379,9 +382,8 @@ public class HiveHook implements ExecuteWithHookContext {
List<Referenceable> source = new ArrayList<>();
for (ReadEntity readEntity : inputs) {
if (readEntity.getType() == Type.TABLE || readEntity.getType() == Type.PARTITION) {
- Map<Type, Referenceable> localEntities = createEntities(dgiBridge, readEntity);
- source.add(localEntities.get(Type.TABLE));
- entities.addAll(localEntities.values());
+ Referenceable inTable = createEntities(dgiBridge, readEntity);
+ source.add(inTable);
}
}
processReferenceable.set("inputs", source);
@@ -389,9 +391,8 @@ public class HiveHook implements ExecuteWithHookContext {
List<Referenceable> target = new ArrayList<>();
for (WriteEntity writeEntity : outputs) {
if (writeEntity.getType() == Type.TABLE || writeEntity.getType() == Type.PARTITION) {
- Map<Type, Referenceable> localEntities = createEntities(dgiBridge, writeEntity);
- target.add(localEntities.get(Type.TABLE));
- entities.addAll(localEntities.values());
+ Referenceable outTable = createEntities(dgiBridge, writeEntity);
+ target.add(outTable);
}
}
processReferenceable.set("outputs", target);
@@ -402,8 +403,7 @@ public class HiveHook implements ExecuteWithHookContext {
//TODO set
processReferenceable.set("queryGraph", "queryGraph");
- entities.add(processReferenceable);
- notifyEntity(entities);
+ messages.add(new HookNotification.EntityCreateRequest(processReferenceable));
}
http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/dce31ab8/addons/hive-bridge/src/test/java/org/apache/atlas/hive/hook/HiveHookIT.java
----------------------------------------------------------------------
diff --git a/addons/hive-bridge/src/test/java/org/apache/atlas/hive/hook/HiveHookIT.java b/addons/hive-bridge/src/test/java/org/apache/atlas/hive/hook/HiveHookIT.java
index 5447de5..1c3d9a4 100755
--- a/addons/hive-bridge/src/test/java/org/apache/atlas/hive/hook/HiveHookIT.java
+++ b/addons/hive-bridge/src/test/java/org/apache/atlas/hive/hook/HiveHookIT.java
@@ -149,6 +149,17 @@ public class HiveHookIT {
assertDatabaseIsRegistered(DEFAULT_DB);
}
+ @Test
+ public void testRenameTable() throws Exception {
+ String tableName = createTable();
+ String newTableName = tableName();
+ runCommand(String.format("alter table %s rename to %s", tableName, newTableName));
+
+ assertTableIsRegistered(DEFAULT_DB, newTableName);
+ assertTableIsNotRegistered(DEFAULT_DB, tableName);
+ }
+
+
private String assertColumnIsRegistered(String colName) throws Exception {
LOG.debug("Searching for column {}", colName);
String query =
@@ -327,8 +338,8 @@ public class HiveHookIT {
LOG.debug("Searching for partition of {}.{} with values {}", dbName, tableName, value);
String dslQuery = String.format("%s as p where values = ['%s'], table where tableName = '%s', "
- + "db where name = '%s' and clusterName = '%s' select p", typeName, value,
- tableName.toLowerCase(), dbName.toLowerCase(), CLUSTER_NAME);
+ + "db where name = '%s' and clusterName = '%s' select p", typeName, value,
+ tableName.toLowerCase(), dbName.toLowerCase(), CLUSTER_NAME);
return assertEntityIsRegistered(dslQuery, "p");
}
http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/dce31ab8/client/src/main/java/org/apache/atlas/AtlasClient.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/atlas/AtlasClient.java b/client/src/main/java/org/apache/atlas/AtlasClient.java
index b108b25..0a730af 100755
--- a/client/src/main/java/org/apache/atlas/AtlasClient.java
+++ b/client/src/main/java/org/apache/atlas/AtlasClient.java
@@ -44,6 +44,7 @@ import javax.ws.rs.core.Response;
import javax.ws.rs.core.UriBuilder;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collection;
import java.util.List;
import static org.apache.atlas.security.SecurityProperties.TLS_ENABLED;
@@ -291,12 +292,16 @@ public class AtlasClient {
}
public JSONArray createEntity(Referenceable... entities) throws AtlasServiceException {
+ return createEntity(Arrays.asList(entities));
+ }
+
+ public JSONArray createEntity(Collection<Referenceable> entities) throws AtlasServiceException {
JSONArray entityArray = getEntitiesArray(entities);
return createEntity(entityArray);
}
- private JSONArray getEntitiesArray(Referenceable[] entities) {
- JSONArray entityArray = new JSONArray(entities.length);
+ private JSONArray getEntitiesArray(Collection<Referenceable> entities) {
+ JSONArray entityArray = new JSONArray(entities.size());
for (Referenceable entity : entities) {
entityArray.put(InstanceSerialization.toJson(entity, true));
}
@@ -311,6 +316,10 @@ public class AtlasClient {
* @throws AtlasServiceException
*/
public JSONArray updateEntities(Referenceable... entities) throws AtlasServiceException {
+ return updateEntities(Arrays.asList(entities));
+ }
+
+ public JSONArray updateEntities(Collection<Referenceable> entities) throws AtlasServiceException {
JSONArray entitiesArray = getEntitiesArray(entities);
JSONObject response = callAPI(API.UPDATE_ENTITY, entitiesArray.toString());
try {
http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/dce31ab8/docs/src/site/twiki/Bridge-Hive.twiki
----------------------------------------------------------------------
diff --git a/docs/src/site/twiki/Bridge-Hive.twiki b/docs/src/site/twiki/Bridge-Hive.twiki
index 81b095e..c8e9009 100644
--- a/docs/src/site/twiki/Bridge-Hive.twiki
+++ b/docs/src/site/twiki/Bridge-Hive.twiki
@@ -29,7 +29,7 @@ The entities are created and de-duped using unique qualified name. They provide
---++ Importing Hive Metadata
org.apache.atlas.hive.bridge.HiveMetaStoreBridge imports the hive metadata into Atlas using the model defined in org.apache.atlas.hive.model.HiveDataModelGenerator. import-hive.sh command can be used to facilitate this.
-Set the following configuration in <atlas-conf>/client.properties and set environment variable HIVE_CONFIG to the hive conf directory:
+Set the following configuration in <atlas-conf>/client.properties and set environment variable $HIVE_CONF_DIR to the hive conf directory:
<verbatim>
<property>
<name>atlas.cluster.name</name>
http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/dce31ab8/notification/src/main/java/org/apache/atlas/kafka/KafkaNotification.java
----------------------------------------------------------------------
diff --git a/notification/src/main/java/org/apache/atlas/kafka/KafkaNotification.java b/notification/src/main/java/org/apache/atlas/kafka/KafkaNotification.java
index bacabeb..37467b3 100644
--- a/notification/src/main/java/org/apache/atlas/kafka/KafkaNotification.java
+++ b/notification/src/main/java/org/apache/atlas/kafka/KafkaNotification.java
@@ -174,7 +174,7 @@ public class KafkaNotification extends AbstractNotification implements Service {
}
@Override
- public void send(NotificationType type, String... messages) throws NotificationException {
+ public void _send(NotificationType type, String... messages) throws NotificationException {
if (producer == null) {
createProducer();
}
http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/dce31ab8/notification/src/main/java/org/apache/atlas/notification/AbstractNotification.java
----------------------------------------------------------------------
diff --git a/notification/src/main/java/org/apache/atlas/notification/AbstractNotification.java b/notification/src/main/java/org/apache/atlas/notification/AbstractNotification.java
index f7bb7b1..72b5a0a 100644
--- a/notification/src/main/java/org/apache/atlas/notification/AbstractNotification.java
+++ b/notification/src/main/java/org/apache/atlas/notification/AbstractNotification.java
@@ -20,6 +20,9 @@ package org.apache.atlas.notification;
import org.apache.atlas.AtlasException;
import org.apache.commons.configuration.Configuration;
+import java.util.Arrays;
+import java.util.List;
+
/**
* Abstract notification interface implementation.
*/
@@ -46,4 +49,20 @@ public abstract class AbstractNotification implements NotificationInterface {
protected final boolean isEmbedded() {
return embedded;
}
+
+ @Override
+ public <T> void send(NotificationType type, List<T> messages) throws NotificationException {
+ String[] strMessages = new String[messages.size()];
+ for (int index = 0; index < messages.size(); index++) {
+ strMessages[index] = AbstractNotificationConsumer.GSON.toJson(messages.get(index));
+ }
+ _send(type, strMessages);
+ }
+
+ @Override
+ public <T> void send(NotificationType type, T... messages) throws NotificationException {
+ send(type, Arrays.asList(messages));
+ }
+
+ protected abstract void _send(NotificationType type, String[] messages) throws NotificationException;
}
http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/dce31ab8/notification/src/main/java/org/apache/atlas/notification/AbstractNotificationConsumer.java
----------------------------------------------------------------------
diff --git a/notification/src/main/java/org/apache/atlas/notification/AbstractNotificationConsumer.java b/notification/src/main/java/org/apache/atlas/notification/AbstractNotificationConsumer.java
index 8c49d4a..42a4e7f 100644
--- a/notification/src/main/java/org/apache/atlas/notification/AbstractNotificationConsumer.java
+++ b/notification/src/main/java/org/apache/atlas/notification/AbstractNotificationConsumer.java
@@ -26,11 +26,16 @@ import com.google.gson.JsonDeserializationContext;
import com.google.gson.JsonDeserializer;
import com.google.gson.JsonElement;
import com.google.gson.JsonParseException;
+import com.google.gson.JsonParser;
+import com.google.gson.JsonSerializationContext;
+import com.google.gson.JsonSerializer;
import com.google.gson.reflect.TypeToken;
import org.apache.atlas.notification.entity.EntityNotification;
import org.apache.atlas.notification.entity.EntityNotificationImpl;
+import org.apache.atlas.notification.hook.HookNotification;
import org.apache.atlas.typesystem.IReferenceableInstance;
import org.apache.atlas.typesystem.IStruct;
+import org.apache.atlas.typesystem.Referenceable;
import org.apache.atlas.typesystem.Struct;
import org.apache.atlas.typesystem.json.InstanceSerialization;
import org.codehaus.jettison.json.JSONArray;
@@ -45,13 +50,15 @@ import java.util.Map;
*/
public abstract class AbstractNotificationConsumer<T> implements NotificationConsumer<T> {
- private static final Gson GSON = new GsonBuilder().
+ public static final Gson GSON = new GsonBuilder().
registerTypeAdapter(ImmutableList.class, new ImmutableListDeserializer()).
registerTypeAdapter(ImmutableMap.class, new ImmutableMapDeserializer()).
registerTypeAdapter(EntityNotification.class, new EntityNotificationDeserializer()).
registerTypeAdapter(IStruct.class, new StructDeserializer()).
- registerTypeAdapter(IReferenceableInstance.class, new ReferenceableDeserializer()).
- registerTypeAdapter(JSONArray.class, new JSONArrayDeserializer()).
+ registerTypeAdapter(IReferenceableInstance.class, new ReferenceableSerializerDeserializer()).
+ registerTypeAdapter(Referenceable.class, new ReferenceableSerializerDeserializer()).
+ registerTypeAdapter(JSONArray.class, new JSONArraySerializerDeserializer()).
+ registerTypeAdapter(HookNotification.HookNotificationMessage.class, new HookNotification()).
create();
private final Class<T> type;
@@ -136,30 +143,44 @@ public abstract class AbstractNotificationConsumer<T> implements NotificationCon
// ----- inner class : StructDeserializer -------------------------------
- public final static class StructDeserializer implements JsonDeserializer<IStruct> {
+ public final static class StructDeserializer implements JsonDeserializer<IStruct>, JsonSerializer<IStruct> {
@Override
public IStruct deserialize(final JsonElement json, final Type type,
final JsonDeserializationContext context) throws JsonParseException {
return context.deserialize(json, Struct.class);
}
+
+ @Override
+ public JsonElement serialize(IStruct src, Type typeOfSrc, JsonSerializationContext context) {
+ String instanceJson = InstanceSerialization.toJson(src, true);
+ return new JsonParser().parse(instanceJson).getAsJsonObject();
+ }
}
- // ----- inner class : ReferenceableDeserializer ------------------------
+ // ----- inner class : ReferenceableSerializerDeserializer ------------------------
- public final static class ReferenceableDeserializer implements JsonDeserializer<IStruct> {
+ public final static class ReferenceableSerializerDeserializer implements JsonDeserializer<IStruct>,
+ JsonSerializer<IReferenceableInstance> {
@Override
public IReferenceableInstance deserialize(final JsonElement json, final Type type,
final JsonDeserializationContext context) throws JsonParseException {
return InstanceSerialization.fromJsonReferenceable(json.toString(), true);
}
+
+ @Override
+ public JsonElement serialize(IReferenceableInstance src, Type typeOfSrc, JsonSerializationContext context) {
+ String instanceJson = InstanceSerialization.toJson(src, true);
+ return new JsonParser().parse(instanceJson).getAsJsonObject();
+ }
}
- // ----- inner class : JSONArrayDeserializer ----------------------------
+ // ----- inner class : JSONArraySerializerDeserializer ----------------------------
- public final static class JSONArrayDeserializer implements JsonDeserializer<JSONArray> {
+ public final static class JSONArraySerializerDeserializer implements JsonDeserializer<JSONArray>,
+ JsonSerializer<JSONArray> {
@Override
public JSONArray deserialize(final JsonElement json, final Type type,
final JsonDeserializationContext context) throws JsonParseException {
@@ -170,5 +191,10 @@ public abstract class AbstractNotificationConsumer<T> implements NotificationCon
throw new JsonParseException(e.getMessage(), e);
}
}
+
+ @Override
+ public JsonElement serialize(JSONArray src, Type typeOfSrc, JsonSerializationContext context) {
+ return new JsonParser().parse(src.toString()).getAsJsonArray();
+ }
}
}
http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/dce31ab8/notification/src/main/java/org/apache/atlas/notification/NotificationHookConsumer.java
----------------------------------------------------------------------
diff --git a/notification/src/main/java/org/apache/atlas/notification/NotificationHookConsumer.java b/notification/src/main/java/org/apache/atlas/notification/NotificationHookConsumer.java
index 1bee26f..e02aafa 100644
--- a/notification/src/main/java/org/apache/atlas/notification/NotificationHookConsumer.java
+++ b/notification/src/main/java/org/apache/atlas/notification/NotificationHookConsumer.java
@@ -23,9 +23,9 @@ import org.apache.atlas.ApplicationProperties;
import org.apache.atlas.AtlasClient;
import org.apache.atlas.AtlasException;
import org.apache.atlas.AtlasServiceException;
+import org.apache.atlas.notification.hook.HookNotification;
import org.apache.atlas.service.Service;
import org.apache.commons.configuration.Configuration;
-import org.codehaus.jettison.json.JSONArray;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -57,11 +57,11 @@ public class NotificationHookConsumer implements Service {
String atlasEndpoint = applicationProperties.getString(ATLAS_ENDPOINT_PROPERTY, "http://localhost:21000");
atlasClient = new AtlasClient(atlasEndpoint);
int numThreads = applicationProperties.getInt(CONSUMER_THREADS_PROPERTY, 1);
- List<NotificationConsumer<JSONArray>> consumers =
+ List<NotificationConsumer<HookNotification.HookNotificationMessage>> consumers =
notificationInterface.createConsumers(NotificationInterface.NotificationType.HOOK, numThreads);
executors = Executors.newFixedThreadPool(consumers.size());
- for (final NotificationConsumer<JSONArray> consumer : consumers) {
+ for (final NotificationConsumer<HookNotification.HookNotificationMessage> consumer : consumers) {
executors.submit(new HookConsumer(consumer));
}
}
@@ -86,14 +86,14 @@ public class NotificationHookConsumer implements Service {
}
class HookConsumer implements Runnable {
- private final NotificationConsumer<JSONArray> consumer;
+ private final NotificationConsumer<HookNotification.HookNotificationMessage> consumer;
private final AtlasClient client;
- public HookConsumer(NotificationConsumer<JSONArray> consumer) {
+ public HookConsumer(NotificationConsumer<HookNotification.HookNotificationMessage> consumer) {
this(atlasClient, consumer);
}
- public HookConsumer(AtlasClient client, NotificationConsumer<JSONArray> consumer) {
+ public HookConsumer(AtlasClient client, NotificationConsumer<HookNotification.HookNotificationMessage> consumer) {
this.client = client;
this.consumer = consumer;
}
@@ -106,14 +106,33 @@ public class NotificationHookConsumer implements Service {
}
while(consumer.hasNext()) {
- JSONArray entityJson = consumer.next();
- LOG.info("Processing message {}", entityJson);
+ HookNotification.HookNotificationMessage message = consumer.next();
+
try {
- JSONArray guids = atlasClient.createEntity(entityJson);
- LOG.info("Create entities with guid {}", guids);
+ switch (message.getType()) {
+ case ENTITY_CREATE:
+ HookNotification.EntityCreateRequest createRequest =
+ (HookNotification.EntityCreateRequest) message;
+ atlasClient.createEntity(createRequest.getEntities());
+ break;
+
+ case ENTITY_PARTIAL_UPDATE:
+ HookNotification.EntityPartialUpdateRequest partialUpdateRequest =
+ (HookNotification.EntityPartialUpdateRequest) message;
+ atlasClient.updateEntity(partialUpdateRequest.getTypeName(),
+ partialUpdateRequest.getAttribute(), partialUpdateRequest.getAttributeValue(),
+ partialUpdateRequest.getEntity());
+ break;
+
+ case ENTITY_FULL_UPDATE:
+ HookNotification.EntityUpdateRequest updateRequest =
+ (HookNotification.EntityUpdateRequest) message;
+ atlasClient.updateEntities(updateRequest.getEntities());
+ break;
+ }
} catch (Exception e) {
//todo handle failures
- LOG.warn("Error handling message {}", entityJson, e);
+ LOG.debug("Error handling message {}", message, e);
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/dce31ab8/notification/src/main/java/org/apache/atlas/notification/NotificationInterface.java
----------------------------------------------------------------------
diff --git a/notification/src/main/java/org/apache/atlas/notification/NotificationInterface.java b/notification/src/main/java/org/apache/atlas/notification/NotificationInterface.java
index 3e68998..e4c4fd6 100644
--- a/notification/src/main/java/org/apache/atlas/notification/NotificationInterface.java
+++ b/notification/src/main/java/org/apache/atlas/notification/NotificationInterface.java
@@ -18,7 +18,7 @@
package org.apache.atlas.notification;
import org.apache.atlas.notification.entity.EntityNotification;
-import org.codehaus.jettison.json.JSONArray;
+import org.apache.atlas.notification.hook.HookNotification;
import java.util.List;
@@ -28,7 +28,7 @@ public interface NotificationInterface {
String PROPERTY_PREFIX = "atlas.notification";
enum NotificationType {
- HOOK(JSONArray.class), ENTITIES(EntityNotification.class);
+ HOOK(HookNotification.HookNotificationMessage.class), ENTITIES(EntityNotification.class);
private final Class classType;
@@ -52,7 +52,9 @@ public interface NotificationInterface {
*/
<T> List<NotificationConsumer<T>> createConsumers(NotificationType notificationType, int numConsumers);
- void send(NotificationType type, String... messages) throws NotificationException;
+ <T> void send(NotificationType type, T... messages) throws NotificationException;
+
+ <T> void send(NotificationType type, List<T> messages) throws NotificationException;
void close();
}
http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/dce31ab8/notification/src/main/java/org/apache/atlas/notification/entity/NotificationEntityChangeListener.java
----------------------------------------------------------------------
diff --git a/notification/src/main/java/org/apache/atlas/notification/entity/NotificationEntityChangeListener.java b/notification/src/main/java/org/apache/atlas/notification/entity/NotificationEntityChangeListener.java
index e2d16df..243f93e 100644
--- a/notification/src/main/java/org/apache/atlas/notification/entity/NotificationEntityChangeListener.java
+++ b/notification/src/main/java/org/apache/atlas/notification/entity/NotificationEntityChangeListener.java
@@ -17,12 +17,6 @@
package org.apache.atlas.notification.entity;
-import com.google.gson.Gson;
-import com.google.gson.GsonBuilder;
-import com.google.gson.JsonElement;
-import com.google.gson.JsonParser;
-import com.google.gson.JsonSerializationContext;
-import com.google.gson.JsonSerializer;
import org.apache.atlas.AtlasException;
import org.apache.atlas.listener.EntityChangeListener;
import org.apache.atlas.notification.NotificationInterface;
@@ -30,10 +24,8 @@ import org.apache.atlas.typesystem.IReferenceableInstance;
import org.apache.atlas.typesystem.IStruct;
import org.apache.atlas.typesystem.ITypedReferenceableInstance;
import org.apache.atlas.typesystem.Referenceable;
-import org.apache.atlas.typesystem.json.InstanceSerialization;
import org.apache.atlas.typesystem.types.TypeSystem;
-import java.lang.reflect.Type;
import java.util.Collection;
import java.util.Collections;
import java.util.LinkedList;
@@ -44,9 +36,6 @@ import java.util.List;
*/
public class NotificationEntityChangeListener implements EntityChangeListener {
- private static final Gson GSON = new GsonBuilder().
- registerTypeAdapter(Referenceable.class, new ReferencableSerializer()).create();
-
private final NotificationInterface notificationInterface;
private final TypeSystem typeSystem;
@@ -93,7 +82,7 @@ public class NotificationEntityChangeListener implements EntityChangeListener {
// send notification of entity change
private void notifyOfEntityEvent(Collection<ITypedReferenceableInstance> entityDefinitions,
EntityNotification.OperationType operationType) throws AtlasException {
- List<String> messages = new LinkedList<>();
+ List<EntityNotification> messages = new LinkedList<>();
for (IReferenceableInstance entityDefinition : entityDefinitions) {
Referenceable entity = new Referenceable(entityDefinition);
@@ -101,24 +90,9 @@ public class NotificationEntityChangeListener implements EntityChangeListener {
EntityNotificationImpl notification =
new EntityNotificationImpl(entity, operationType, typeSystem);
- messages.add(GSON.toJson(notification));
+ messages.add(notification);
}
- notificationInterface.send(NotificationInterface.NotificationType.ENTITIES,
- messages.toArray(new String[messages.size()]));
- }
-
-
- // ----- inner class : ReferencableSerializer ---------------------------
-
- private static class ReferencableSerializer implements JsonSerializer<Referenceable> {
-
- public static final JsonParser JSON_PARSER = new JsonParser();
-
- @Override
- public JsonElement serialize(Referenceable referenceable, Type type,
- JsonSerializationContext jsonSerializationContext) {
- return JSON_PARSER.parse(InstanceSerialization.toJson(referenceable, true)).getAsJsonObject();
- }
+ notificationInterface.send(NotificationInterface.NotificationType.ENTITIES, messages);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/dce31ab8/notification/src/main/java/org/apache/atlas/notification/hook/HookNotification.java
----------------------------------------------------------------------
diff --git a/notification/src/main/java/org/apache/atlas/notification/hook/HookNotification.java b/notification/src/main/java/org/apache/atlas/notification/hook/HookNotification.java
new file mode 100644
index 0000000..568f58b
--- /dev/null
+++ b/notification/src/main/java/org/apache/atlas/notification/hook/HookNotification.java
@@ -0,0 +1,174 @@
+/**
+ * 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.notification.hook;
+
+import com.google.gson.JsonDeserializationContext;
+import com.google.gson.JsonDeserializer;
+import com.google.gson.JsonElement;
+import com.google.gson.JsonObject;
+import com.google.gson.JsonParseException;
+import org.apache.atlas.typesystem.Referenceable;
+import org.apache.atlas.typesystem.TypesDef;
+import org.apache.atlas.typesystem.json.InstanceSerialization;
+import org.codehaus.jettison.json.JSONArray;
+import org.codehaus.jettison.json.JSONException;
+
+import java.lang.reflect.Type;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+public class HookNotification implements JsonDeserializer<HookNotification.HookNotificationMessage> {
+
+ @Override
+ public HookNotificationMessage deserialize(JsonElement json, Type typeOfT,
+ JsonDeserializationContext context) throws JsonParseException {
+ if (json.isJsonArray()) {
+ JSONArray jsonArray = context.deserialize(json, JSONArray.class);
+ return new EntityCreateRequest(jsonArray);
+ } else {
+ HookNotificationType type =
+ context.deserialize(((JsonObject) json).get("type"), HookNotificationType.class);
+ switch (type) {
+ case ENTITY_CREATE:
+ return context.deserialize(json, EntityCreateRequest.class);
+
+ case ENTITY_FULL_UPDATE:
+ return context.deserialize(json, EntityUpdateRequest.class);
+
+ case ENTITY_PARTIAL_UPDATE:
+ return context.deserialize(json, EntityPartialUpdateRequest.class);
+
+ case TYPE_CREATE:
+ case TYPE_UPDATE:
+ return context.deserialize(json, TypeRequest.class);
+ }
+ throw new IllegalStateException("Unhandled type " + type);
+ }
+ }
+
+ public enum HookNotificationType {
+ TYPE_CREATE, TYPE_UPDATE, ENTITY_CREATE, ENTITY_PARTIAL_UPDATE, ENTITY_FULL_UPDATE
+ }
+
+ public static class HookNotificationMessage {
+ protected HookNotificationType type;
+
+ private HookNotificationMessage() { }
+
+ public HookNotificationMessage(HookNotificationType type) {
+ this.type = type;
+ }
+
+ public HookNotificationType getType() {
+ return type;
+ }
+ }
+
+ public static class TypeRequest extends HookNotificationMessage {
+ private TypesDef typesDef;
+
+ private TypeRequest() { }
+
+ public TypeRequest(HookNotificationType type, TypesDef typesDef) {
+ super(type);
+ this.typesDef = typesDef;
+ }
+
+ public TypesDef getTypesDef() {
+ return typesDef;
+ }
+ }
+
+ public static class EntityCreateRequest extends HookNotificationMessage {
+ private List<Referenceable> entities;
+
+ private EntityCreateRequest() { }
+
+ public EntityCreateRequest(Referenceable... entities) {
+ super(HookNotificationType.ENTITY_CREATE);
+ this.entities = Arrays.asList(entities);
+ }
+
+ protected EntityCreateRequest(HookNotificationType type, List<Referenceable> entities) {
+ super(type);
+ this.entities = entities;
+ }
+
+ public EntityCreateRequest(JSONArray jsonArray) {
+ super(HookNotificationType.ENTITY_CREATE);
+ entities = new ArrayList<>();
+ for (int index = 0; index < jsonArray.length(); index++) {
+ try {
+ entities.add(InstanceSerialization.fromJsonReferenceable(jsonArray.getString(index), true));
+ } catch (JSONException e) {
+ throw new JsonParseException(e);
+ }
+ }
+ }
+
+ public List<Referenceable> getEntities() throws JSONException {
+ return entities;
+ }
+ }
+
+ public static class EntityUpdateRequest extends EntityCreateRequest {
+ public EntityUpdateRequest(Referenceable... entities) {
+ this(Arrays.asList(entities));
+ }
+
+ public EntityUpdateRequest(List<Referenceable> entities) {
+ super(HookNotificationType.ENTITY_FULL_UPDATE, entities);
+ }
+ }
+
+ public static class EntityPartialUpdateRequest extends HookNotificationMessage {
+ private String typeName;
+ private String attribute;
+ private Referenceable entity;
+ private String attributeValue;
+
+ private EntityPartialUpdateRequest() { }
+
+ public EntityPartialUpdateRequest(String typeName, String attribute, String attributeValue,
+ Referenceable entity) {
+ super(HookNotificationType.ENTITY_PARTIAL_UPDATE);
+ this.typeName = typeName;
+ this.attribute = attribute;
+ this.attributeValue = attributeValue;
+ this.entity = entity;
+ }
+
+ public String getTypeName() {
+ return typeName;
+ }
+
+ public String getAttribute() {
+ return attribute;
+ }
+
+ public Referenceable getEntity() {
+ return entity;
+ }
+
+ public String getAttributeValue() {
+ return attributeValue;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/dce31ab8/notification/src/test/java/org/apache/atlas/kafka/KafkaNotificationTest.java
----------------------------------------------------------------------
diff --git a/notification/src/test/java/org/apache/atlas/kafka/KafkaNotificationTest.java b/notification/src/test/java/org/apache/atlas/kafka/KafkaNotificationTest.java
index 625a0b0..eb90f52 100644
--- a/notification/src/test/java/org/apache/atlas/kafka/KafkaNotificationTest.java
+++ b/notification/src/test/java/org/apache/atlas/kafka/KafkaNotificationTest.java
@@ -26,9 +26,7 @@ import org.apache.atlas.notification.NotificationConsumer;
import org.apache.atlas.notification.NotificationInterface;
import org.apache.atlas.notification.NotificationModule;
import org.apache.commons.configuration.Configuration;
-import org.apache.commons.lang.RandomStringUtils;
import org.apache.kafka.clients.consumer.ConsumerConfig;
-import org.codehaus.jettison.json.JSONArray;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Guice;
@@ -61,20 +59,6 @@ public class KafkaNotificationTest {
}
@Test
- public void testSendReceiveMessage() throws Exception {
- String msg1 = "[{\"message\": " + 123 + "}]";
- String msg2 = "[{\"message\": " + 456 + "}]";
- kafka.send(NotificationInterface.NotificationType.HOOK, msg1, msg2);
- List<NotificationConsumer<JSONArray>> consumers =
- kafka.createConsumers(NotificationInterface.NotificationType.HOOK, 1);
- NotificationConsumer<JSONArray> consumer = consumers.get(0);
- assertTrue(consumer.hasNext());
- assertEquals(new JSONArray(msg1), consumer.next());
- assertTrue(consumer.hasNext());
- assertEquals(new JSONArray(msg2), consumer.next());
- }
-
- @Test
@SuppressWarnings("unchecked")
public void testCreateConsumers() throws Exception {
Configuration configuration = mock(Configuration.class);
@@ -119,10 +103,6 @@ public class KafkaNotificationTest {
assertEquals(groupId, properties.getProperty(ConsumerConfig.GROUP_ID_CONFIG));
}
- private String random() {
- return RandomStringUtils.randomAlphanumeric(5);
- }
-
@AfterClass
public void teardown() throws Exception {
kafka.stop();
http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/dce31ab8/notification/src/test/java/org/apache/atlas/notification/hook/HookNotificationTest.java
----------------------------------------------------------------------
diff --git a/notification/src/test/java/org/apache/atlas/notification/hook/HookNotificationTest.java b/notification/src/test/java/org/apache/atlas/notification/hook/HookNotificationTest.java
new file mode 100644
index 0000000..4b9f81f
--- /dev/null
+++ b/notification/src/test/java/org/apache/atlas/notification/hook/HookNotificationTest.java
@@ -0,0 +1,68 @@
+/**
+ * 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.notification.hook;
+
+import org.apache.atlas.notification.AbstractNotificationConsumer;
+import org.apache.atlas.typesystem.Referenceable;
+import org.apache.atlas.typesystem.json.InstanceSerialization;
+import org.codehaus.jettison.json.JSONArray;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotNull;
+
+public class HookNotificationTest {
+
+ @Test
+ public void testMessageBackwardCompatibility() throws Exception {
+ JSONArray jsonArray = new JSONArray();
+ Referenceable entity = new Referenceable("sometype");
+ entity.set("name", "somename");
+ String entityJson = InstanceSerialization.toJson(entity, true);
+ jsonArray.put(entityJson);
+
+ HookNotification.HookNotificationMessage notification = AbstractNotificationConsumer.GSON.fromJson(
+ jsonArray.toString(), HookNotification.HookNotificationMessage.class);
+ assertNotNull(notification);
+ assertEquals(notification.getType(), HookNotification.HookNotificationType.ENTITY_CREATE);
+ HookNotification.EntityCreateRequest createRequest = (HookNotification.EntityCreateRequest) notification;
+ assertEquals(createRequest.getEntities().size(), 1);
+ assertEquals(createRequest.getEntities().get(0).getTypeName(), entity.getTypeName());
+ }
+
+ @Test
+ public void testNewMessageSerDe() throws Exception {
+ Referenceable entity1 = new Referenceable("sometype");
+ entity1.set("attr", "value");
+ entity1.set("complex", new Referenceable("othertype"));
+ Referenceable entity2 = new Referenceable("newtype");
+ HookNotification.EntityCreateRequest request = new HookNotification.EntityCreateRequest(entity1, entity2);
+
+ String notificationJson = AbstractNotificationConsumer.GSON.toJson(request);
+ HookNotification.HookNotificationMessage actualNotification = AbstractNotificationConsumer.GSON.fromJson(
+ notificationJson, HookNotification.HookNotificationMessage.class);
+ assertEquals(actualNotification.getType(), HookNotification.HookNotificationType.ENTITY_CREATE);
+ HookNotification.EntityCreateRequest createRequest = (HookNotification.EntityCreateRequest) actualNotification;
+ assertEquals(createRequest.getEntities().size(), 2);
+ Referenceable actualEntity1 = createRequest.getEntities().get(0);
+ assertEquals(actualEntity1.getTypeName(), "sometype");
+ assertEquals(((Referenceable)actualEntity1.get("complex")).getTypeName(), "othertype");
+ assertEquals(createRequest.getEntities().get(1).getTypeName(), "newtype");
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/dce31ab8/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index d9c3df1..0c3ab4d 100755
--- a/pom.xml
+++ b/pom.xml
@@ -354,7 +354,8 @@
<!-- skips checkstyle and find bugs -->
<skipCheck>false</skipCheck>
- <skipTests>false</skipTests>
+ <skipUTs>false</skipUTs>
+ <skipITs>false</skipITs>
<skipDocs>true</skipDocs>
<skipSite>true</skipSite>
<projectBaseDir>${project.basedir}</projectBaseDir>
@@ -1087,7 +1088,7 @@
<dependency>
<groupId>com.google.code.gson</groupId>
<artifactId>gson</artifactId>
- <version>2.3.1</version>
+ <version>2.5</version>
</dependency>
<dependency>
@@ -1394,6 +1395,7 @@
<configuration>
<systemProperties>
<user.dir>${project.basedir}</user.dir>
+ <atlas.data>${project.build.directory}/data</atlas.data>
</systemProperties>
<!--<skipTests>true</skipTests>-->
<forkMode>always</forkMode>
@@ -1403,6 +1405,7 @@
-Xmx1024m -XX:MaxPermSize=512m -Dlog4j.configuration=atlas-log4j.xml
-Djava.net.preferIPv4Stack=true
</argLine>
+ <skip>${skipUTs}</skip>
<excludes>
<exclude>**/*Base*</exclude>
</excludes>
@@ -1423,12 +1426,14 @@
<configuration>
<systemPropertyVariables>
<projectBaseDir>${projectBaseDir}</projectBaseDir>
+ <atlas.data>${project.build.directory}/data</atlas.data>
</systemPropertyVariables>
<redirectTestOutputToFile>true</redirectTestOutputToFile>
<argLine>-Djava.awt.headless=true -Dproject.version=${project.version}
-Dhadoop.tmp.dir="${project.build.directory}/tmp-hadoop-${user.name}"
-Xmx1024m -XX:MaxPermSize=512m -Dlog4j.configuration=atlas-log4j.xml
</argLine>
+ <skip>${skipITs}</skip>
<parallel>none</parallel>
<reuseForks>false</reuseForks>
<forkCount>1</forkCount>
http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/dce31ab8/release-log.txt
----------------------------------------------------------------------
diff --git a/release-log.txt b/release-log.txt
index d2ee389..1ad4977 100644
--- a/release-log.txt
+++ b/release-log.txt
@@ -14,6 +14,7 @@ ATLAS-54 Rename configs in hive hook (shwethags)
ATLAS-3 Mixed Index creation fails with Date types (sumasai via shwethags)
ALL CHANGES:
+ATLAS-386 Handle hive rename Table (shwethags)
ATLAS-374 Doc: Create a wiki for documenting fault tolerance and HA options for Atlas data (yhemanth via sumasai)
ATLAS-346 Atlas server loses messages sent from Hive hook if restarted after unclean shutdown (yhemanth via sumasai)
ATLAS-382 Fixed Hive Bridge doc for ATLAS cluster name (sumasai)
http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/dce31ab8/repository/src/main/java/org/apache/atlas/repository/graph/TitanGraphProvider.java
----------------------------------------------------------------------
diff --git a/repository/src/main/java/org/apache/atlas/repository/graph/TitanGraphProvider.java b/repository/src/main/java/org/apache/atlas/repository/graph/TitanGraphProvider.java
index 26d5e2b..2cc1a50 100755
--- a/repository/src/main/java/org/apache/atlas/repository/graph/TitanGraphProvider.java
+++ b/repository/src/main/java/org/apache/atlas/repository/graph/TitanGraphProvider.java
@@ -22,14 +22,9 @@ import com.google.common.collect.ImmutableMap;
import com.google.inject.Provides;
import com.thinkaurelius.titan.core.TitanFactory;
import com.thinkaurelius.titan.core.TitanGraph;
-import com.thinkaurelius.titan.core.TitanTransaction;
import com.thinkaurelius.titan.core.schema.TitanManagement;
-import com.thinkaurelius.titan.diskstorage.Backend;
import com.thinkaurelius.titan.diskstorage.StandardIndexProvider;
-import com.thinkaurelius.titan.diskstorage.indexing.IndexInformation;
import com.thinkaurelius.titan.diskstorage.solr.Solr5Index;
-import com.thinkaurelius.titan.graphdb.configuration.GraphDatabaseConfiguration;
-import com.thinkaurelius.titan.graphdb.database.StandardTitanGraph;
import org.apache.atlas.ApplicationProperties;
import org.apache.atlas.AtlasException;
import org.apache.commons.configuration.Configuration;
@@ -41,7 +36,6 @@ import java.lang.reflect.Field;
import java.lang.reflect.Modifier;
import java.util.HashMap;
import java.util.Map;
-import java.util.Set;
/**
* Default implementation for Graph Provider that doles out Titan Graph.
http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/dce31ab8/repository/src/main/java/org/apache/atlas/services/DefaultMetadataService.java
----------------------------------------------------------------------
diff --git a/repository/src/main/java/org/apache/atlas/services/DefaultMetadataService.java b/repository/src/main/java/org/apache/atlas/services/DefaultMetadataService.java
index fb782a2..f605c26 100755
--- a/repository/src/main/java/org/apache/atlas/services/DefaultMetadataService.java
+++ b/repository/src/main/java/org/apache/atlas/services/DefaultMetadataService.java
@@ -304,7 +304,14 @@ public class DefaultMetadataService implements MetadataService {
ParamChecker.notEmpty(entityTypeName, "Entity type cannot be null");
ClassType entityType = typeSystem.getDataType(ClassType.class, entityTypeName);
+
+ //Both assigned id and values are required for full update
+ //classtype.convert() will remove values if id is assigned. So, set temp id, convert and
+ // then replace with original id
+ Id origId = entityInstance.getId();
+ entityInstance.replaceWithNewId(new Id(entityInstance.getTypeName()));
ITypedReferenceableInstance typedInstrance = entityType.convert(entityInstance, Multiplicity.REQUIRED);
+ ((ReferenceableInstance)typedInstrance).replaceWithNewId(origId);
instances[index] = typedInstrance;
}
return instances;
http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/dce31ab8/repository/src/test/java/org/apache/atlas/TestUtils.java
----------------------------------------------------------------------
diff --git a/repository/src/test/java/org/apache/atlas/TestUtils.java b/repository/src/test/java/org/apache/atlas/TestUtils.java
index 12c47d4..2a45bc8 100755
--- a/repository/src/test/java/org/apache/atlas/TestUtils.java
+++ b/repository/src/test/java/org/apache/atlas/TestUtils.java
@@ -251,12 +251,12 @@ public final class TestUtils {
new AttributeDefinition("columnsMap",
DataTypes.mapTypeName(DataTypes.STRING_TYPE.getName(),
"column_type"),
- Multiplicity.COLLECTION, true, null),
+ Multiplicity.OPTIONAL, true, null),
//map of structs
new AttributeDefinition("partitionsMap",
DataTypes.mapTypeName(DataTypes.STRING_TYPE.getName(),
"partition_type"),
- Multiplicity.COLLECTION, true, null),
+ Multiplicity.OPTIONAL, true, null),
// struct reference
new AttributeDefinition("serde1", "serdeType", Multiplicity.OPTIONAL, false, null),
new AttributeDefinition("serde2", "serdeType", Multiplicity.OPTIONAL, false, null),
http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/dce31ab8/repository/src/test/java/org/apache/atlas/service/DefaultMetadataServiceTest.java
----------------------------------------------------------------------
diff --git a/repository/src/test/java/org/apache/atlas/service/DefaultMetadataServiceTest.java b/repository/src/test/java/org/apache/atlas/service/DefaultMetadataServiceTest.java
index 0352ef3..0307fd4 100644
--- a/repository/src/test/java/org/apache/atlas/service/DefaultMetadataServiceTest.java
+++ b/repository/src/test/java/org/apache/atlas/service/DefaultMetadataServiceTest.java
@@ -22,9 +22,6 @@ import com.google.common.collect.ImmutableList;
import com.google.inject.Inject;
import com.thinkaurelius.titan.core.TitanGraph;
import com.thinkaurelius.titan.core.util.TitanCleanup;
-import org.apache.atlas.typesystem.exception.TypeNotFoundException;
-import org.apache.atlas.typesystem.exception.EntityNotFoundException;
-import org.apache.atlas.utils.ParamChecker;
import org.apache.atlas.RepositoryMetadataModule;
import org.apache.atlas.TestUtils;
import org.apache.atlas.repository.graph.GraphProvider;
@@ -32,12 +29,15 @@ import org.apache.atlas.services.MetadataService;
import org.apache.atlas.typesystem.Referenceable;
import org.apache.atlas.typesystem.Struct;
import org.apache.atlas.typesystem.TypesDef;
+import org.apache.atlas.typesystem.exception.EntityNotFoundException;
+import org.apache.atlas.typesystem.exception.TypeNotFoundException;
import org.apache.atlas.typesystem.json.InstanceSerialization;
import org.apache.atlas.typesystem.json.TypesSerialization;
import org.apache.atlas.typesystem.persistence.Id;
import org.apache.atlas.typesystem.types.EnumValue;
import org.apache.atlas.typesystem.types.TypeSystem;
import org.apache.atlas.typesystem.types.ValueConversionException;
+import org.apache.atlas.utils.ParamChecker;
import org.apache.commons.lang.RandomStringUtils;
import org.codehaus.jettison.json.JSONArray;
import org.testng.Assert;
@@ -47,11 +47,15 @@ import org.testng.annotations.Guice;
import org.testng.annotations.Test;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+
@Guice(modules = RepositoryMetadataModule.class)
public class DefaultMetadataServiceTest {
@Inject
@@ -296,8 +300,8 @@ public class DefaultMetadataServiceTest {
Map<String, Object> values = new HashMap<>();
values.put("name", "col1");
values.put("type", "type");
- Referenceable ref = new Referenceable("column_type", values);
- columns.add(ref);
+ Referenceable col1 = new Referenceable("column_type", values);
+ columns.add(col1);
Referenceable tableUpdated = new Referenceable(TestUtils.TABLE_TYPE, new HashMap<String, Object>() {{
put("columns", columns);
}});
@@ -307,19 +311,18 @@ public class DefaultMetadataServiceTest {
metadataService.getEntityDefinition(TestUtils.TABLE_TYPE, "name", (String) table.get("name"));
Referenceable tableDefinition = InstanceSerialization.fromJsonReferenceable(tableDefinitionJson, true);
final List<Referenceable> arrClsColumns = (List) tableDefinition.get("columns");
- Assert.assertTrue(arrClsColumns.get(0).equalsContents(columns.get(0)));
+ assertReferenceables(arrClsColumns.get(0), columns.get(0));
//Partial update. Add col5 But also update col1
Map<String, Object> valuesCol5 = new HashMap<>();
valuesCol5.put("name", "col5");
valuesCol5.put("type", "type");
- ref = new Referenceable("column_type", valuesCol5);
+ Referenceable col2 = new Referenceable("column_type", valuesCol5);
//update col1
- arrClsColumns.get(0).set("type", "type1");
+ col1.set("type", "type1");
//add col5
- final List<Referenceable> updateColumns = new ArrayList<>(arrClsColumns);
- updateColumns.add(ref);
+ final List<Referenceable> updateColumns = Arrays.asList(col1, col2);
tableUpdated = new Referenceable(TestUtils.TABLE_TYPE, new HashMap<String, Object>() {{
put("columns", updateColumns);
@@ -331,8 +334,8 @@ public class DefaultMetadataServiceTest {
tableDefinition = InstanceSerialization.fromJsonReferenceable(tableDefinitionJson, true);
List<Referenceable> arrColumnsList = (List) tableDefinition.get("columns");
Assert.assertEquals(arrColumnsList.size(), 2);
- Assert.assertTrue(arrColumnsList.get(0).equalsContents(updateColumns.get(0)));
- Assert.assertTrue(arrColumnsList.get(1).equalsContents(updateColumns.get(1)));
+ assertReferenceables(arrColumnsList.get(0), updateColumns.get(0));
+ assertReferenceables(arrColumnsList.get(1), updateColumns.get(1));
//Complete update. Add array elements - col3,4
Map<String, Object> values1 = new HashMap<>();
@@ -355,9 +358,8 @@ public class DefaultMetadataServiceTest {
tableDefinition = InstanceSerialization.fromJsonReferenceable(tableDefinitionJson, true);
arrColumnsList = (List) tableDefinition.get("columns");
Assert.assertEquals(arrColumnsList.size(), columns.size());
- Assert.assertTrue(arrColumnsList.get(1).equalsContents(columns.get(1)));
- Assert.assertTrue(arrColumnsList.get(2).equalsContents(columns.get(2)));
-
+ assertReferenceables(arrColumnsList.get(1), columns.get(1));
+ assertReferenceables(arrColumnsList.get(2), columns.get(2));
//Remove a class reference/Id and insert another reference
//Also covers isComposite case since columns is a composite
@@ -366,8 +368,8 @@ public class DefaultMetadataServiceTest {
values.put("name", "col2");
values.put("type", "type");
- ref = new Referenceable("column_type", values);
- columns.add(ref);
+ col1 = new Referenceable("column_type", values);
+ columns.add(col1);
table.set("columns", columns);
updateInstance(table);
@@ -376,7 +378,7 @@ public class DefaultMetadataServiceTest {
tableDefinition = InstanceSerialization.fromJsonReferenceable(tableDefinitionJson, true);
arrColumnsList = (List) tableDefinition.get("columns");
Assert.assertEquals(arrColumnsList.size(), columns.size());
- Assert.assertTrue(arrColumnsList.get(0).equalsContents(columns.get(0)));
+ assertReferenceables(arrColumnsList.get(0), columns.get(0));
//Update array column to null
table.setNull("columns");
@@ -389,6 +391,14 @@ public class DefaultMetadataServiceTest {
Assert.assertNull(tableDefinition.get("columns"));
}
+ private void assertReferenceables(Referenceable r1, Referenceable r2) {
+ assertEquals(r1.getTypeName(), r2.getTypeName());
+ assertTrue(r1.getTraits().equals(r2.getTraits()));
+ for (String attr : r1.getValuesMap().keySet()) {
+ assertTrue(r1.getValuesMap().get(attr).equals(r2.getValuesMap().get(attr)));
+ }
+ //TODO assert trait instances and complex attributes
+ }
@Test
public void testStructs() throws Exception {
http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/dce31ab8/typesystem/src/main/java/org/apache/atlas/typesystem/Referenceable.java
----------------------------------------------------------------------
diff --git a/typesystem/src/main/java/org/apache/atlas/typesystem/Referenceable.java b/typesystem/src/main/java/org/apache/atlas/typesystem/Referenceable.java
index b8dcc7e..aa1736d 100755
--- a/typesystem/src/main/java/org/apache/atlas/typesystem/Referenceable.java
+++ b/typesystem/src/main/java/org/apache/atlas/typesystem/Referenceable.java
@@ -33,7 +33,7 @@ import java.util.Map;
*/
public class Referenceable extends Struct implements IReferenceableInstance {
- private final Id id;
+ private Id id;
private final ImmutableMap<String, IStruct> traits;
private final ImmutableList<String> traitNames;
@@ -151,6 +151,10 @@ public class Referenceable extends Struct implements IReferenceableInstance {
'}';
}
+ public void replaceWithNewId(Id id) {
+ this.id = id;
+ }
+
private static Map<String, IStruct> getTraits(IReferenceableInstance instance) throws AtlasException {
Map<String, IStruct> traits = new HashMap<>();
for (String traitName : instance.getTraits() ) {
http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/dce31ab8/typesystem/src/main/java/org/apache/atlas/typesystem/types/ClassType.java
----------------------------------------------------------------------
diff --git a/typesystem/src/main/java/org/apache/atlas/typesystem/types/ClassType.java b/typesystem/src/main/java/org/apache/atlas/typesystem/types/ClassType.java
index ac758fa..adf5f1c 100755
--- a/typesystem/src/main/java/org/apache/atlas/typesystem/types/ClassType.java
+++ b/typesystem/src/main/java/org/apache/atlas/typesystem/types/ClassType.java
@@ -125,9 +125,9 @@ public class ClassType extends HierarchicalType<ClassType, IReferenceableInstanc
r != null ? createInstanceWithTraits(id, r, r.getTraits().toArray(new String[0])) :
createInstance(id);
-// if (id != null && id.isAssigned()) {
-// return tr;
-// }
+ if (id != null && id.isAssigned()) {
+ return tr;
+ }
for (Map.Entry<String, AttributeInfo> e : fieldMapping.fields.entrySet()) {
String attrKey = e.getKey();
http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/dce31ab8/typesystem/src/main/java/org/apache/atlas/typesystem/types/Multiplicity.java
----------------------------------------------------------------------
diff --git a/typesystem/src/main/java/org/apache/atlas/typesystem/types/Multiplicity.java b/typesystem/src/main/java/org/apache/atlas/typesystem/types/Multiplicity.java
index 74d7f7c..a54dabc 100755
--- a/typesystem/src/main/java/org/apache/atlas/typesystem/types/Multiplicity.java
+++ b/typesystem/src/main/java/org/apache/atlas/typesystem/types/Multiplicity.java
@@ -25,8 +25,8 @@ public final class Multiplicity {
public static final Multiplicity OPTIONAL = new Multiplicity(0, 1, false);
public static final Multiplicity REQUIRED = new Multiplicity(1, 1, false);
- public static final Multiplicity COLLECTION = new Multiplicity(0, Integer.MAX_VALUE, false);
- public static final Multiplicity SET = new Multiplicity(0, Integer.MAX_VALUE, true);
+ public static final Multiplicity COLLECTION = new Multiplicity(1, Integer.MAX_VALUE, false);
+ public static final Multiplicity SET = new Multiplicity(1, Integer.MAX_VALUE, true);
public final int lower;
public final int upper;
http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/dce31ab8/typesystem/src/main/resources/application.properties
----------------------------------------------------------------------
diff --git a/typesystem/src/main/resources/application.properties b/typesystem/src/main/resources/application.properties
index 4a351e6..d475d7e 100644
--- a/typesystem/src/main/resources/application.properties
+++ b/typesystem/src/main/resources/application.properties
@@ -27,7 +27,7 @@ atlas.graph.storage.backend=${titan.storage.backend}
atlas.graph.index.search.backend=${titan.index.backend}
#Berkeley storage directory
-atlas.graph.storage.directory=target/data/berkley
+atlas.graph.storage.directory=${sys:atlas.data}/berkley
#hbase
#For standalone mode , specify localhost
@@ -38,7 +38,7 @@ atlas.graph.storage.hbase.regions-per-server=1
atlas.graph.storage.lock.wait-time=10000
#ElasticSearch
-atlas.graph.index.search.directory=target/data/es
+atlas.graph.index.search.directory=${sys:atlas.data}/es
atlas.graph.index.search.elasticsearch.client-only=false
atlas.graph.index.search.elasticsearch.local-mode=true
atlas.graph.index.search.elasticsearch.create.sleep=2000
@@ -63,7 +63,7 @@ atlas.notification.embedded=true
atlas.kafka.zookeeper.connect=localhost:19026
atlas.kafka.bootstrap.servers=localhost:19027
-atlas.kafka.data=target/data/kafka
+atlas.kafka.data=${sys:atlas.data}/kafka
atlas.kafka.zookeeper.session.timeout.ms=400
atlas.kafka.zookeeper.sync.time.ms=20
atlas.kafka.auto.commit.interval.ms=100
http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/dce31ab8/webapp/pom.xml
----------------------------------------------------------------------
diff --git a/webapp/pom.xml b/webapp/pom.xml
index e92795a..a765daa 100755
--- a/webapp/pom.xml
+++ b/webapp/pom.xml
@@ -372,6 +372,10 @@
<name>atlas.home</name>
<value>${project.build.directory}</value>
</systemProperty>
+ <systemProperty>
+ <name>atlas.data</name>
+ <value>${project.build.directory}/data</value>
+ </systemProperty>
</systemProperties>
<stopKey>atlas-stop</stopKey>
<stopPort>31001</stopPort>
http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/dce31ab8/webapp/src/main/java/org/apache/atlas/Atlas.java
----------------------------------------------------------------------
diff --git a/webapp/src/main/java/org/apache/atlas/Atlas.java b/webapp/src/main/java/org/apache/atlas/Atlas.java
index 5ae1e44..2d2d619 100755
--- a/webapp/src/main/java/org/apache/atlas/Atlas.java
+++ b/webapp/src/main/java/org/apache/atlas/Atlas.java
@@ -40,6 +40,7 @@ public final class Atlas {
private static final String APP_PATH = "app";
private static final String APP_PORT = "port";
private static final String ATLAS_HOME = "atlas.home";
+ private static final String ATLAS_DATA = "atlas.data";
private static final String ATLAS_LOG_DIR = "atlas.log.dir";
public static final String ATLAS_SERVER_HTTPS_PORT = "atlas.server.https.port";
public static final String ATLAS_SERVER_HTTP_PORT = "atlas.server.http.port";
@@ -110,6 +111,9 @@ public final class Atlas {
if (System.getProperty(ATLAS_HOME) == null) {
System.setProperty(ATLAS_HOME, "target");
}
+ if (System.getProperty(ATLAS_DATA) == null) {
+ System.setProperty(ATLAS_DATA, "target/data");
+ }
if (System.getProperty(ATLAS_LOG_DIR) == null) {
System.setProperty(ATLAS_LOG_DIR, "target/logs");
}
http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/dce31ab8/webapp/src/test/java/org/apache/atlas/notification/NotificationHookConsumerIT.java
----------------------------------------------------------------------
diff --git a/webapp/src/test/java/org/apache/atlas/notification/NotificationHookConsumerIT.java b/webapp/src/test/java/org/apache/atlas/notification/NotificationHookConsumerIT.java
index e5af26c..3a4661c 100644
--- a/webapp/src/test/java/org/apache/atlas/notification/NotificationHookConsumerIT.java
+++ b/webapp/src/test/java/org/apache/atlas/notification/NotificationHookConsumerIT.java
@@ -19,22 +19,22 @@
package org.apache.atlas.notification;
import com.google.inject.Inject;
+import org.apache.atlas.notification.hook.HookNotification;
import org.apache.atlas.typesystem.Referenceable;
-import org.apache.atlas.typesystem.json.InstanceSerialization;
import org.apache.atlas.web.resources.BaseResourceIT;
import org.codehaus.jettison.json.JSONArray;
-import org.testng.Assert;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Guice;
import org.testng.annotations.Test;
+import static org.testng.Assert.assertEquals;
+
@Guice(modules = NotificationModule.class)
public class NotificationHookConsumerIT extends BaseResourceIT {
@Inject
private NotificationInterface kafka;
- private String dbName;
@BeforeClass
public void setUp() throws Exception {
@@ -47,57 +47,106 @@ public class NotificationHookConsumerIT extends BaseResourceIT {
kafka.close();
}
- private void sendHookMessage(Referenceable entity) throws NotificationException {
- String entityJson = InstanceSerialization.toJson(entity, true);
- JSONArray jsonArray = new JSONArray();
- jsonArray.put(entityJson);
- kafka.send(NotificationInterface.NotificationType.HOOK, jsonArray.toString());
+ private void sendHookMessage(HookNotification.HookNotificationMessage message) throws NotificationException {
+ kafka.send(NotificationInterface.NotificationType.HOOK, message);
}
@Test
- public void testConsumeHookMessage() throws Exception {
- Referenceable entity = new Referenceable(DATABASE_TYPE);
- dbName = "db" + randomString();
- entity.set("name", dbName);
+ public void testCreateEntity() throws Exception {
+ final Referenceable entity = new Referenceable(DATABASE_TYPE);
+ entity.set("name", "db" + randomString());
entity.set("description", randomString());
- sendHookMessage(entity);
+ sendHookMessage(new HookNotification.EntityCreateRequest(entity));
waitFor(1000, new Predicate() {
@Override
public boolean evaluate() throws Exception {
- JSONArray results =
- serviceClient.searchByDSL(String.format("%s where name='%s'", DATABASE_TYPE, dbName));
+ JSONArray results = serviceClient.searchByDSL(String.format("%s where name='%s'", DATABASE_TYPE,
+ entity.get("name")));
return results.length() == 1;
}
});
}
- @Test (dependsOnMethods = "testConsumeHookMessage")
- public void testEnityDeduping() throws Exception {
-// Referenceable db = serviceClient.getEntity(DATABASE_TYPE, "name", dbName);
- Referenceable db = new Referenceable(DATABASE_TYPE);
- db.set("name", dbName);
- db.set("description", randomString());
+ @Test
+ public void testUpdateEntityPartial() throws Exception {
+ final Referenceable entity = new Referenceable(DATABASE_TYPE);
+ final String dbName = "db" + randomString();
+ entity.set("name", dbName);
+ entity.set("description", randomString());
+ serviceClient.createEntity(entity);
- Referenceable table = new Referenceable(HIVE_TABLE_TYPE);
- final String tableName = randomString();
- table.set("name", tableName);
- table.set("db", db);
+ final Referenceable newEntity = new Referenceable(DATABASE_TYPE);
+ newEntity.set("owner", randomString());
+ sendHookMessage(new HookNotification.EntityPartialUpdateRequest(DATABASE_TYPE, "name", dbName, newEntity));
+ waitFor(1000, new Predicate() {
+ @Override
+ public boolean evaluate() throws Exception {
+ Referenceable localEntity = serviceClient.getEntity(DATABASE_TYPE, "name", dbName);
+ return (localEntity.get("owner") != null && localEntity.get("owner").equals(newEntity.get("owner")));
+ }
+ });
+
+ //Its partial update and un-set fields are not updated
+ Referenceable actualEntity = serviceClient.getEntity(DATABASE_TYPE, "name", dbName);
+ assertEquals(actualEntity.get("description"), entity.get("description"));
+ }
+
+ @Test
+ public void testUpdatePartialUpdatingQualifiedName() throws Exception {
+ final Referenceable entity = new Referenceable(DATABASE_TYPE);
+ final String dbName = "db" + randomString();
+ entity.set("name", dbName);
+ entity.set("description", randomString());
+ serviceClient.createEntity(entity);
- sendHookMessage(table);
+ final Referenceable newEntity = new Referenceable(DATABASE_TYPE);
+ final String newName = "db" + randomString();
+ newEntity.set("name", newName);
+
+ sendHookMessage(new HookNotification.EntityPartialUpdateRequest(DATABASE_TYPE, "name", dbName, newEntity));
waitFor(1000, new Predicate() {
@Override
public boolean evaluate() throws Exception {
- JSONArray results =
- serviceClient.searchByDSL(String.format("%s where name='%s'", HIVE_TABLE_TYPE, tableName));
+ JSONArray results = serviceClient.searchByDSL(String.format("%s where name='%s'", DATABASE_TYPE,
+ newName));
return results.length() == 1;
}
});
- JSONArray results =
- serviceClient.searchByDSL(String.format("%s where name='%s'", DATABASE_TYPE, dbName));
- Assert.assertEquals(results.length(), 1);
+ //no entity with the old qualified name
+ JSONArray results = serviceClient.searchByDSL(String.format("%s where name='%s'", DATABASE_TYPE, dbName));
+ assertEquals(results.length(), 0);
+
}
+ @Test
+ public void testUpdateEntityFullUpdate() throws Exception {
+ Referenceable entity = new Referenceable(DATABASE_TYPE);
+ final String dbName = "db" + randomString();
+ entity.set("name", dbName);
+ entity.set("description", randomString());
+ serviceClient.createEntity(entity);
+
+ final Referenceable newEntity = new Referenceable(DATABASE_TYPE);
+ newEntity.set("name", dbName);
+ newEntity.set("description", randomString());
+ newEntity.set("owner", randomString());
+
+ //updating unique attribute
+ sendHookMessage(new HookNotification.EntityUpdateRequest(newEntity));
+ waitFor(1000, new Predicate() {
+ @Override
+ public boolean evaluate() throws Exception {
+ JSONArray results = serviceClient.searchByDSL(String.format("%s where name='%s'", DATABASE_TYPE,
+ dbName));
+ return results.length() == 1;
+ }
+ });
+
+ Referenceable actualEntity = serviceClient.getEntity(DATABASE_TYPE, "name", dbName);
+ assertEquals(actualEntity.get("description"), newEntity.get("description"));
+ assertEquals(actualEntity.get("owner"), newEntity.get("owner"));
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/dce31ab8/webapp/src/test/java/org/apache/atlas/web/resources/EntityJerseyResourceIT.java
----------------------------------------------------------------------
diff --git a/webapp/src/test/java/org/apache/atlas/web/resources/EntityJerseyResourceIT.java b/webapp/src/test/java/org/apache/atlas/web/resources/EntityJerseyResourceIT.java
index 7337eaf..f476af3 100755
--- a/webapp/src/test/java/org/apache/atlas/web/resources/EntityJerseyResourceIT.java
+++ b/webapp/src/test/java/org/apache/atlas/web/resources/EntityJerseyResourceIT.java
@@ -58,6 +58,8 @@ import java.util.List;
import java.util.Map;
import java.util.UUID;
+import static org.testng.Assert.assertEquals;
+
/**
* Integration tests for Entity Jersey Resource.
*/
@@ -95,6 +97,7 @@ public class EntityJerseyResourceIT extends BaseResourceIT {
}
@Test
+ //API should accept single entity (or jsonarray of entities)
public void testSubmitSingleEntity() throws Exception {
Referenceable databaseInstance = new Referenceable(DATABASE_TYPE);
databaseInstance.set("name", randomString());
@@ -115,6 +118,34 @@ public class EntityJerseyResourceIT extends BaseResourceIT {
}
@Test
+ public void testEntityDeduping() throws Exception {
+ Referenceable db = new Referenceable(DATABASE_TYPE);
+ String dbName = "db" + randomString();
+ db.set("name", dbName);
+ db.set("description", randomString());
+
+ serviceClient.createEntity(db);
+ JSONArray results =
+ serviceClient.searchByDSL(String.format("%s where name='%s'", DATABASE_TYPE, dbName));
+ assertEquals(results.length(), 1);
+
+ //create entity again shouldn't create another instance with same unique attribute value
+ serviceClient.createEntity(db);
+ results = serviceClient.searchByDSL(String.format("%s where name='%s'", DATABASE_TYPE, dbName));
+ assertEquals(results.length(), 1);
+
+ //Test the same across references
+ Referenceable table = new Referenceable(HIVE_TABLE_TYPE);
+ final String tableName = randomString();
+ table.set("name", tableName);
+ table.set("db", db);
+
+ serviceClient.createEntity(table);
+ results = serviceClient.searchByDSL(String.format("%s where name='%s'", DATABASE_TYPE, dbName));
+ assertEquals(results.length(), 1);
+ }
+
+ @Test
public void testEntityDefinitionAcrossTypeUpdate() throws Exception {
//create type
HierarchicalTypeDefinition<ClassType> typeDefinition = TypesUtil
[2/2] incubator-atlas git commit: ATLAS-342 Atlas is sending an
ENTITY_CREATE event to the ATLAS_ENTITIES topic even if the entity exists
already (shwethags)
Posted by sh...@apache.org.
ATLAS-342 Atlas is sending an ENTITY_CREATE event to the ATLAS_ENTITIES topic even if the entity exists already (shwethags)
Project: http://git-wip-us.apache.org/repos/asf/incubator-atlas/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-atlas/commit/8cbb2c13
Tree: http://git-wip-us.apache.org/repos/asf/incubator-atlas/tree/8cbb2c13
Diff: http://git-wip-us.apache.org/repos/asf/incubator-atlas/diff/8cbb2c13
Branch: refs/heads/master
Commit: 8cbb2c13bbe71957daf0de9222daaf4dd01199d3
Parents: dce31ab
Author: Shwetha GS <ss...@hortonworks.com>
Authored: Tue Dec 15 15:50:33 2015 +0530
Committer: Shwetha GS <ss...@hortonworks.com>
Committed: Tue Dec 15 15:50:33 2015 +0530
----------------------------------------------------------------------
release-log.txt | 1 +
.../atlas/repository/MetadataRepository.java | 11 +-
.../graph/GraphBackedMetadataRepository.java | 13 ++-
.../atlas/repository/graph/GraphHelper.java | 8 +-
.../graph/TypedInstanceToGraphMapper.java | 100 +++++++++++--------
.../atlas/services/DefaultMetadataService.java | 80 ++++++++-------
.../apache/atlas/BaseHiveRepositoryTest.java | 7 +-
.../GraphBackedMetadataRepositoryTest.java | 27 +++--
.../graph/GraphRepoMapperScaleTest.java | 2 +-
.../service/DefaultMetadataServiceTest.java | 9 +-
.../apache/atlas/services/MetadataService.java | 17 ++--
.../atlas/typesystem/types/TypeUtils.java | 10 +-
.../atlas/web/resources/EntityResource.java | 8 +-
.../notification/EntityNotificationIT.java | 58 +----------
.../atlas/web/resources/BaseResourceIT.java | 54 +++++++++-
.../web/resources/EntityJerseyResourceIT.java | 52 +++++++++-
16 files changed, 268 insertions(+), 189 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/8cbb2c13/release-log.txt
----------------------------------------------------------------------
diff --git a/release-log.txt b/release-log.txt
index 1ad4977..ce3337d 100644
--- a/release-log.txt
+++ b/release-log.txt
@@ -14,6 +14,7 @@ ATLAS-54 Rename configs in hive hook (shwethags)
ATLAS-3 Mixed Index creation fails with Date types (sumasai via shwethags)
ALL CHANGES:
+ATLAS-342 Atlas is sending an ENTITY_CREATE event to the ATLAS_ENTITIES topic even if the entity exists already (shwethags)
ATLAS-386 Handle hive rename Table (shwethags)
ATLAS-374 Doc: Create a wiki for documenting fault tolerance and HA options for Atlas data (yhemanth via sumasai)
ATLAS-346 Atlas server loses messages sent from Hive hook if restarted after unclean shutdown (yhemanth via sumasai)
http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/8cbb2c13/repository/src/main/java/org/apache/atlas/repository/MetadataRepository.java
----------------------------------------------------------------------
diff --git a/repository/src/main/java/org/apache/atlas/repository/MetadataRepository.java b/repository/src/main/java/org/apache/atlas/repository/MetadataRepository.java
index 2091e71..f66a4e5 100755
--- a/repository/src/main/java/org/apache/atlas/repository/MetadataRepository.java
+++ b/repository/src/main/java/org/apache/atlas/repository/MetadataRepository.java
@@ -19,12 +19,13 @@
package org.apache.atlas.repository;
import org.apache.atlas.AtlasException;
-import org.apache.atlas.typesystem.exception.EntityExistsException;
-import org.apache.atlas.typesystem.exception.EntityNotFoundException;
import org.apache.atlas.typesystem.ITypedReferenceableInstance;
import org.apache.atlas.typesystem.ITypedStruct;
+import org.apache.atlas.typesystem.exception.EntityExistsException;
+import org.apache.atlas.typesystem.exception.EntityNotFoundException;
import org.apache.atlas.typesystem.types.AttributeInfo;
import org.apache.atlas.typesystem.types.IDataType;
+import org.apache.atlas.typesystem.types.TypeUtils;
import java.util.List;
@@ -82,7 +83,7 @@ public interface MetadataRepository {
* @throws RepositoryException
* @throws EntityExistsException
*/
- String[] createEntities(ITypedReferenceableInstance... entities) throws RepositoryException, EntityExistsException;
+ List<String> createEntities(ITypedReferenceableInstance... entities) throws RepositoryException, EntityExistsException;
/**
* Fetch the complete definition of an entity given its GUID.
@@ -143,13 +144,13 @@ public interface MetadataRepository {
* Adds/Updates the property to the entity that corresponds to the GUID
* Supports only primitive attribute/Class Id updations.
*/
- void updatePartial(ITypedReferenceableInstance entity) throws RepositoryException;
+ TypeUtils.Pair<List<String>, List<String>> updatePartial(ITypedReferenceableInstance entity) throws RepositoryException;
/**
* Adds the property to the entity that corresponds to the GUID
* @param entitiesToBeUpdated The entities to be updated
*/
- String[] updateEntities(ITypedReferenceableInstance... entitiesToBeUpdated) throws RepositoryException;
+ TypeUtils.Pair<List<String>, List<String>> updateEntities(ITypedReferenceableInstance... entitiesToBeUpdated) throws RepositoryException;
/**
* Returns the entity for the given type and qualified name
http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/8cbb2c13/repository/src/main/java/org/apache/atlas/repository/graph/GraphBackedMetadataRepository.java
----------------------------------------------------------------------
diff --git a/repository/src/main/java/org/apache/atlas/repository/graph/GraphBackedMetadataRepository.java b/repository/src/main/java/org/apache/atlas/repository/graph/GraphBackedMetadataRepository.java
index fe1e576..dd64124 100755
--- a/repository/src/main/java/org/apache/atlas/repository/graph/GraphBackedMetadataRepository.java
+++ b/repository/src/main/java/org/apache/atlas/repository/graph/GraphBackedMetadataRepository.java
@@ -37,6 +37,7 @@ import org.apache.atlas.typesystem.types.AttributeInfo;
import org.apache.atlas.typesystem.types.ClassType;
import org.apache.atlas.typesystem.types.IDataType;
import org.apache.atlas.typesystem.types.TypeSystem;
+import org.apache.atlas.typesystem.types.TypeUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -114,12 +115,14 @@ public class GraphBackedMetadataRepository implements MetadataRepository {
@Override
@GraphTransaction
- public String[] createEntities(ITypedReferenceableInstance... entities) throws RepositoryException,
+ public List<String> createEntities(ITypedReferenceableInstance... entities) throws RepositoryException,
EntityExistsException {
LOG.info("adding entities={}", entities);
try {
TypedInstanceToGraphMapper instanceToGraphMapper = new TypedInstanceToGraphMapper(graphToInstanceMapper);
- return instanceToGraphMapper.mapTypedInstanceToGraph(TypedInstanceToGraphMapper.Operation.CREATE, entities);
+ TypeUtils.Pair<List<String>, List<String>> idPair =
+ instanceToGraphMapper.mapTypedInstanceToGraph(TypedInstanceToGraphMapper.Operation.CREATE, entities);
+ return idPair.left;
} catch (EntityExistsException e) {
throw e;
} catch (AtlasException e) {
@@ -279,7 +282,7 @@ public class GraphBackedMetadataRepository implements MetadataRepository {
@Override
@GraphTransaction
- public String[] updateEntities(ITypedReferenceableInstance... entitiesUpdated) throws RepositoryException {
+ public TypeUtils.Pair<List<String>, List<String>> updateEntities(ITypedReferenceableInstance... entitiesUpdated) throws RepositoryException {
LOG.info("updating entity {}", entitiesUpdated);
try {
TypedInstanceToGraphMapper instanceToGraphMapper = new TypedInstanceToGraphMapper(graphToInstanceMapper);
@@ -292,11 +295,11 @@ public class GraphBackedMetadataRepository implements MetadataRepository {
@Override
@GraphTransaction
- public void updatePartial(ITypedReferenceableInstance entity) throws RepositoryException {
+ public TypeUtils.Pair<List<String>, List<String>> updatePartial(ITypedReferenceableInstance entity) throws RepositoryException {
LOG.info("updating entity {}", entity);
try {
TypedInstanceToGraphMapper instanceToGraphMapper = new TypedInstanceToGraphMapper(graphToInstanceMapper);
- instanceToGraphMapper.mapTypedInstanceToGraph(TypedInstanceToGraphMapper.Operation.UPDATE_PARTIAL, entity);
+ return instanceToGraphMapper.mapTypedInstanceToGraph(TypedInstanceToGraphMapper.Operation.UPDATE_PARTIAL, entity);
} catch (AtlasException e) {
throw new RepositoryException(e);
}
http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/8cbb2c13/repository/src/main/java/org/apache/atlas/repository/graph/GraphHelper.java
----------------------------------------------------------------------
diff --git a/repository/src/main/java/org/apache/atlas/repository/graph/GraphHelper.java b/repository/src/main/java/org/apache/atlas/repository/graph/GraphHelper.java
index 9ac2819..6b2d5d1 100755
--- a/repository/src/main/java/org/apache/atlas/repository/graph/GraphHelper.java
+++ b/repository/src/main/java/org/apache/atlas/repository/graph/GraphHelper.java
@@ -71,13 +71,13 @@ public final class GraphHelper {
return INSTANCE;
}
- public Vertex createVertexWithIdentity(ITypedReferenceableInstance typedInstance,
- Set<String> superTypeNames) {
+ public Vertex createVertexWithIdentity(ITypedReferenceableInstance typedInstance, Set<String> superTypeNames) {
+ final String guid = UUID.randomUUID().toString();
+
final Vertex vertexWithIdentity = createVertexWithoutIdentity(typedInstance.getTypeName(),
- typedInstance.getId(), superTypeNames);
+ new Id(guid, 0 , typedInstance.getTypeName()), superTypeNames);
// add identity
- final String guid = UUID.randomUUID().toString();
setProperty(vertexWithIdentity, Constants.GUID_PROPERTY_KEY, guid);
// add version information
http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/8cbb2c13/repository/src/main/java/org/apache/atlas/repository/graph/TypedInstanceToGraphMapper.java
----------------------------------------------------------------------
diff --git a/repository/src/main/java/org/apache/atlas/repository/graph/TypedInstanceToGraphMapper.java b/repository/src/main/java/org/apache/atlas/repository/graph/TypedInstanceToGraphMapper.java
index 7ef5c50..996f31b 100644
--- a/repository/src/main/java/org/apache/atlas/repository/graph/TypedInstanceToGraphMapper.java
+++ b/repository/src/main/java/org/apache/atlas/repository/graph/TypedInstanceToGraphMapper.java
@@ -41,8 +41,8 @@ import org.apache.atlas.typesystem.types.Multiplicity;
import org.apache.atlas.typesystem.types.ObjectGraphWalker;
import org.apache.atlas.typesystem.types.TraitType;
import org.apache.atlas.typesystem.types.TypeSystem;
+import org.apache.atlas.typesystem.types.TypeUtils;
import org.apache.atlas.utils.MD5Utils;
-import org.apache.commons.lang3.tuple.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -79,36 +79,40 @@ public final class TypedInstanceToGraphMapper {
this.graphToTypedInstanceMapper = graphToTypedInstanceMapper;
}
- String[] mapTypedInstanceToGraph(Operation operation, ITypedReferenceableInstance... typedInstances)
+ TypeUtils.Pair<List<String>, List<String>> mapTypedInstanceToGraph(Operation operation, ITypedReferenceableInstance... typedInstances)
throws AtlasException {
- List<String> guids = new ArrayList<>();
+
+ List<String> createdIds = new ArrayList<>();
+ List<String> updatedIds = new ArrayList<>();
+
for (ITypedReferenceableInstance typedInstance : typedInstances) {
Collection<IReferenceableInstance> newInstances = walkClassInstances(typedInstance);
- Pair<List<ITypedReferenceableInstance>, List<ITypedReferenceableInstance>> instancesPair =
+ TypeUtils.Pair<List<ITypedReferenceableInstance>, List<ITypedReferenceableInstance>> instancesPair =
createVerticesAndDiscoverInstances(newInstances);
switch (operation) {
case CREATE:
- addOrUpdateAttributesAndTraits(operation, instancesPair.getLeft());
- addFullTextProperty(instancesPair.getLeft());
+ List<String> ids = addOrUpdateAttributesAndTraits(operation, instancesPair.left);
+ createdIds.addAll(ids);
+ addFullTextProperty(instancesPair.left);
break;
case UPDATE_FULL:
case UPDATE_PARTIAL:
- List<ITypedReferenceableInstance> instancesForUpdate = instancesPair.getLeft();
- instancesForUpdate.addAll(instancesPair.getRight());
- addOrUpdateAttributesAndTraits(operation, instancesForUpdate);
- addFullTextProperty(instancesForUpdate);
+ ids = addOrUpdateAttributesAndTraits(Operation.CREATE, instancesPair.left);
+ createdIds.addAll(ids);
+ ids = addOrUpdateAttributesAndTraits(operation, instancesPair.right);
+ updatedIds.addAll(ids);
+
+ addFullTextProperty(instancesPair.left);
+ addFullTextProperty(instancesPair.right);
break;
case DELETE:
throw new UnsupportedOperationException("Not handled - " + operation);
}
-
- //Return guid for
- addToGuids(typedInstance, guids);
}
- return guids.toArray(new String[guids.size()]);
+ return TypeUtils.Pair.of(createdIds, updatedIds);
}
private Collection<IReferenceableInstance> walkClassInstances(ITypedReferenceableInstance typedInstance)
@@ -126,18 +130,21 @@ public final class TypedInstanceToGraphMapper {
return entityProcessor.getInstances();
}
- private void addOrUpdateAttributesAndTraits(Operation operation, List<ITypedReferenceableInstance> instances) throws AtlasException {
+ private List<String> addOrUpdateAttributesAndTraits(Operation operation, List<ITypedReferenceableInstance> instances) throws AtlasException {
+ List<String> guids = new ArrayList<>();
for (ITypedReferenceableInstance instance : instances) {
try {
//new vertex, set all the properties
- addOrUpdateAttributesAndTraits(operation, instance);
+ String guid = addOrUpdateAttributesAndTraits(operation, instance);
+ guids.add(guid);
} catch (SchemaViolationException e) {
throw new EntityExistsException(instance, e);
}
}
+ return guids;
}
- private void addOrUpdateAttributesAndTraits(Operation operation, ITypedReferenceableInstance typedInstance)
+ private String addOrUpdateAttributesAndTraits(Operation operation, ITypedReferenceableInstance typedInstance)
throws AtlasException {
LOG.debug("Adding/Updating typed instance {}", typedInstance.getTypeName());
@@ -158,6 +165,8 @@ public final class TypedInstanceToGraphMapper {
//TODO - Handle Trait updates
addTraits(typedInstance, instanceVertex, classType);
}
+
+ return getId(typedInstance)._getId();
}
private void mapInstanceToVertex(ITypedInstance typedInstance, Vertex instanceVertex,
@@ -215,14 +224,16 @@ public final class TypedInstanceToGraphMapper {
}
}
- private Pair<List<ITypedReferenceableInstance>, List<ITypedReferenceableInstance>> createVerticesAndDiscoverInstances(
+ private TypeUtils.Pair<List<ITypedReferenceableInstance>, List<ITypedReferenceableInstance>> createVerticesAndDiscoverInstances(
Collection<IReferenceableInstance> instances) throws AtlasException {
List<ITypedReferenceableInstance> instancesToCreate = new ArrayList<>();
List<ITypedReferenceableInstance> instancesToUpdate = new ArrayList<>();
for (IReferenceableInstance instance : instances) {
+ ITypedReferenceableInstance newInstance;
Id id = instance.getId();
+
if (!idToVertexMap.containsKey(id)) {
Vertex instanceVertex;
if (id.isAssigned()) { // has a GUID
@@ -231,7 +242,9 @@ public final class TypedInstanceToGraphMapper {
throw new IllegalStateException(
String.format("%s is not of type ITypedReferenceableInstance", instance));
}
- instancesToUpdate.add((ITypedReferenceableInstance) instance);
+ newInstance = (ITypedReferenceableInstance) instance;
+ instancesToUpdate.add(newInstance);
+
} else {
//Check if there is already an instance with the same unique attribute value
ClassType classType = typeSystem.getDataType(ClassType.class, instance.getTypeName());
@@ -239,31 +252,28 @@ public final class TypedInstanceToGraphMapper {
//no entity with the given unique attribute, create new
if (instanceVertex == null) {
- ITypedReferenceableInstance newInstance = classType.convert(instance, Multiplicity.REQUIRED);
+ newInstance = classType.convert(instance, Multiplicity.REQUIRED);
instanceVertex = graphHelper.createVertexWithIdentity(newInstance, classType.getAllSuperTypeNames());
instancesToCreate.add(newInstance);
//Map only unique attributes for cases of circular references
mapInstanceToVertex(newInstance, instanceVertex, classType.fieldMapping().fields, true, Operation.CREATE);
+
} else {
if (!(instance instanceof ReferenceableInstance)) {
throw new IllegalStateException(
String.format("%s is not of type ITypedReferenceableInstance", instance));
}
- instancesToUpdate.add((ITypedReferenceableInstance) instance);
+ newInstance = (ITypedReferenceableInstance) instance;
+ instancesToUpdate.add(newInstance);
}
}
+ //Set the id in the new instance
idToVertexMap.put(id, instanceVertex);
}
}
- return Pair.of(instancesToCreate, instancesToUpdate);
- }
-
- private void addToGuids(ITypedReferenceableInstance typedInstance, List<String> guids) {
- Vertex instanceVertex = idToVertexMap.get(typedInstance.getId());
- String guid = instanceVertex.getProperty(Constants.GUID_PROPERTY_KEY);
- guids.add(guid);
+ return TypeUtils.Pair.of(instancesToCreate, instancesToUpdate);
}
private void addFullTextProperty(List<ITypedReferenceableInstance> instances) throws AtlasException {
@@ -275,7 +285,8 @@ public final class TypedInstanceToGraphMapper {
}
}
- private void addTraits(ITypedReferenceableInstance typedInstance, Vertex instanceVertex, ClassType classType) throws AtlasException {
+ private void addTraits(ITypedReferenceableInstance typedInstance, Vertex instanceVertex, ClassType classType)
+ throws AtlasException {
for (String traitName : typedInstance.getTraits()) {
LOG.debug("mapping trait {}", traitName);
GraphHelper.addProperty(instanceVertex, Constants.TRAIT_NAMES_PROPERTY_KEY, traitName);
@@ -288,7 +299,8 @@ public final class TypedInstanceToGraphMapper {
/******************************************** STRUCT **************************************************/
- private Pair<Vertex, Edge> updateStructVertex(ITypedStruct structInstance, Edge relEdge, Operation operation) throws AtlasException {
+ private TypeUtils.Pair<Vertex, Edge> updateStructVertex(ITypedStruct structInstance, Edge relEdge,
+ Operation operation) throws AtlasException {
//Already existing vertex. Update
Vertex structInstanceVertex = relEdge.getVertex(Direction.IN);
@@ -303,10 +315,11 @@ public final class TypedInstanceToGraphMapper {
mapInstanceToVertex(structInstance, structInstanceVertex, structInstance.fieldMapping().fields, false, operation);
GraphHelper.setProperty(structInstanceVertex, SIGNATURE_HASH_PROPERTY_KEY, String.valueOf(newSignature));
}
- return Pair.of(structInstanceVertex, relEdge);
+ return TypeUtils.Pair.of(structInstanceVertex, relEdge);
}
- private Pair<Vertex, Edge> addStructVertex(ITypedStruct structInstance, Vertex instanceVertex, AttributeInfo attributeInfo, String edgeLabel) throws AtlasException {
+ private TypeUtils.Pair<Vertex, Edge> addStructVertex(ITypedStruct structInstance, Vertex instanceVertex,
+ AttributeInfo attributeInfo, String edgeLabel) throws AtlasException {
// add a new vertex for the struct or trait instance
Vertex structInstanceVertex = graphHelper.createVertexWithoutIdentity(structInstance.getTypeName(), null,
Collections.<String>emptySet()); // no super types for struct type
@@ -317,7 +330,7 @@ public final class TypedInstanceToGraphMapper {
// add an edge to the newly created vertex from the parent
Edge relEdge = graphHelper.addEdge(instanceVertex, structInstanceVertex, edgeLabel);
- return Pair.of(structInstanceVertex, relEdge);
+ return TypeUtils.Pair.of(structInstanceVertex, relEdge);
}
/******************************************** ARRAY **************************************************/
@@ -443,7 +456,7 @@ public final class TypedInstanceToGraphMapper {
private String addOrUpdateStruct(Vertex instanceVertex, AttributeInfo attributeInfo, IDataType elementType,
ITypedStruct structAttr, String curVal,
String edgeLabel, Operation operation) throws AtlasException {
- Pair<Vertex, Edge> vertexEdgePair = null;
+ TypeUtils.Pair<Vertex, Edge> vertexEdgePair = null;
if (curVal != null && structAttr == null) {
//remove edge
removeUnusedReference(curVal, attributeInfo, elementType);
@@ -456,7 +469,7 @@ public final class TypedInstanceToGraphMapper {
vertexEdgePair = addStructVertex(structAttr, instanceVertex, attributeInfo, edgeLabel);
}
- return (vertexEdgePair != null) ? vertexEdgePair.getRight().getId().toString() : null;
+ return (vertexEdgePair != null) ? vertexEdgePair.right.getId().toString() : null;
}
private String addOrUpdateClassVertex(Vertex instanceVertex, AttributeInfo attributeInfo, IDataType elementType,
@@ -468,27 +481,28 @@ public final class TypedInstanceToGraphMapper {
throw new EntityNotFoundException("Could not find vertex for Class Reference " + newVal);
}
- Pair<Vertex, Edge> vertexEdgePair = null;
+ TypeUtils.Pair<Vertex, Edge> vertexEdgePair = null;
if (curVal != null && newVal == null) {
//remove edge
removeUnusedReference(curVal, attributeInfo, elementType);
} else if (curVal != null && newVal != null) {
Edge edge = graphHelper.getOutGoingEdgeById(curVal);
Id classRefId = getId(newVal);
- vertexEdgePair = updateClassEdge(classRefId, newVal, instanceVertex, edge, toVertex, attributeInfo, elementType, edgeLabel, operation);
+ vertexEdgePair = updateClassEdge(classRefId, newVal, instanceVertex, edge, toVertex, attributeInfo,
+ elementType, edgeLabel, operation);
} else if (newVal != null){
vertexEdgePair = addClassEdge(instanceVertex, toVertex, edgeLabel);
}
- return (vertexEdgePair != null) ? vertexEdgePair.getRight().getId().toString() : null;
+ return (vertexEdgePair != null) ? vertexEdgePair.right.getId().toString() : null;
}
/******************************************** CLASS **************************************************/
- private Pair<Vertex, Edge> addClassEdge(Vertex instanceVertex, Vertex toVertex, String edgeLabel) throws AtlasException {
+ private TypeUtils.Pair<Vertex, Edge> addClassEdge(Vertex instanceVertex, Vertex toVertex, String edgeLabel) throws AtlasException {
// add an edge to the class vertex from the instance
Edge edge = graphHelper.addEdge(instanceVertex, toVertex, edgeLabel);
- return Pair.of(toVertex, edge);
+ return TypeUtils.Pair.of(toVertex, edge);
}
private Vertex getClassVertex(ITypedReferenceableInstance typedReference) throws EntityNotFoundException {
@@ -521,11 +535,11 @@ public final class TypedInstanceToGraphMapper {
}
- private Pair<Vertex, Edge> updateClassEdge(Id id, final ITypedReferenceableInstance typedInstance,
+ private TypeUtils.Pair<Vertex, Edge> updateClassEdge(Id id, final ITypedReferenceableInstance typedInstance,
Vertex instanceVertex, Edge edge, Vertex toVertex,
AttributeInfo attributeInfo, IDataType dataType,
String edgeLabel, Operation operation) throws AtlasException {
- Pair<Vertex, Edge> result = Pair.of(toVertex, edge);
+ TypeUtils.Pair<Vertex, Edge> result = TypeUtils.Pair.of(toVertex, edge);
Edge newEdge = edge;
// Update edge if it exists
Vertex invertex = edge.getVertex(Direction.IN);
@@ -535,7 +549,7 @@ public final class TypedInstanceToGraphMapper {
// add an edge to the class vertex from the instance
if(toVertex != null) {
newEdge = graphHelper.addEdge(instanceVertex, toVertex, edgeLabel);
- result = Pair.of(toVertex, newEdge);
+ result = TypeUtils.Pair.of(toVertex, newEdge);
}
removeUnusedReference(edge.getId().toString(), attributeInfo, dataType);
}
http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/8cbb2c13/repository/src/main/java/org/apache/atlas/services/DefaultMetadataService.java
----------------------------------------------------------------------
diff --git a/repository/src/main/java/org/apache/atlas/services/DefaultMetadataService.java b/repository/src/main/java/org/apache/atlas/services/DefaultMetadataService.java
index f605c26..b38face 100755
--- a/repository/src/main/java/org/apache/atlas/services/DefaultMetadataService.java
+++ b/repository/src/main/java/org/apache/atlas/services/DefaultMetadataService.java
@@ -23,16 +23,12 @@ import com.google.common.collect.ImmutableList;
import com.google.inject.Provider;
import org.apache.atlas.AtlasClient;
import org.apache.atlas.AtlasException;
-import org.apache.atlas.repository.RepositoryException;
-import org.apache.atlas.typesystem.exception.EntityNotFoundException;
-import org.apache.atlas.typesystem.exception.TypeNotFoundException;
-import org.apache.atlas.typesystem.persistence.ReferenceableInstance;
-import org.apache.atlas.utils.ParamChecker;
import org.apache.atlas.classification.InterfaceAudience;
import org.apache.atlas.listener.EntityChangeListener;
import org.apache.atlas.listener.TypesChangeListener;
import org.apache.atlas.repository.IndexCreationException;
import org.apache.atlas.repository.MetadataRepository;
+import org.apache.atlas.repository.RepositoryException;
import org.apache.atlas.repository.typestore.ITypeStore;
import org.apache.atlas.typesystem.IStruct;
import org.apache.atlas.typesystem.ITypedReferenceableInstance;
@@ -40,9 +36,12 @@ import org.apache.atlas.typesystem.ITypedStruct;
import org.apache.atlas.typesystem.Referenceable;
import org.apache.atlas.typesystem.Struct;
import org.apache.atlas.typesystem.TypesDef;
+import org.apache.atlas.typesystem.exception.EntityNotFoundException;
+import org.apache.atlas.typesystem.exception.TypeNotFoundException;
import org.apache.atlas.typesystem.json.InstanceSerialization;
import org.apache.atlas.typesystem.json.TypesSerialization;
import org.apache.atlas.typesystem.persistence.Id;
+import org.apache.atlas.typesystem.persistence.ReferenceableInstance;
import org.apache.atlas.typesystem.types.AttributeDefinition;
import org.apache.atlas.typesystem.types.AttributeInfo;
import org.apache.atlas.typesystem.types.ClassType;
@@ -54,25 +53,24 @@ import org.apache.atlas.typesystem.types.Multiplicity;
import org.apache.atlas.typesystem.types.StructTypeDefinition;
import org.apache.atlas.typesystem.types.TraitType;
import org.apache.atlas.typesystem.types.TypeSystem;
+import org.apache.atlas.typesystem.types.TypeUtils;
import org.apache.atlas.typesystem.types.ValueConversionException;
import org.apache.atlas.typesystem.types.utils.TypesUtil;
+import org.apache.atlas.utils.ParamChecker;
import org.codehaus.jettison.json.JSONArray;
import org.codehaus.jettison.json.JSONException;
import org.codehaus.jettison.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import scala.actors.threadpool.Arrays;
import javax.inject.Inject;
import javax.inject.Singleton;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
-import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
-import java.util.Set;
/**
* Simple wrapper over TypeSystem and MetadataRepository services with hooks
@@ -279,17 +277,10 @@ public class DefaultMetadataService implements MetadataService {
ITypedReferenceableInstance[] typedInstances = deserializeClassInstances(entityInstanceDefinition);
- final String[] guids = repository.createEntities(typedInstances);
-
- Set<ITypedReferenceableInstance> entitites = new HashSet<>();
+ final List<String> guids = repository.createEntities(typedInstances);
- for (String guid : guids) {
- entitites.add(repository.getEntityDefinition(guid));
- }
-
- onEntitiesAdded(entitites);
-
- return new JSONArray(Arrays.asList(guids)).toString();
+ onEntitiesAdded(guids);
+ return new JSONArray(guids).toString();
}
private ITypedReferenceableInstance[] deserializeClassInstances(String entityInstanceDefinition)
@@ -390,14 +381,20 @@ public class DefaultMetadataService implements MetadataService {
ParamChecker.notEmpty(entityInstanceDefinition, "Entity instance definition cannot be empty");
ITypedReferenceableInstance[] typedInstances = deserializeClassInstances(entityInstanceDefinition);
- String[] guids = repository.updateEntities(typedInstances);
- onEntitiesAdded(Arrays.asList(typedInstances));
+ TypeUtils.Pair<List<String>, List<String>> guids = repository.updateEntities(typedInstances);
+ return onEntitiesAddedUpdated(guids);
+ }
+
+ private String onEntitiesAddedUpdated(TypeUtils.Pair<List<String>, List<String>> guids) throws AtlasException {
+ onEntitiesAdded(guids.left);
+ onEntitiesUpdated(guids.right);
- return new JSONArray(Arrays.asList(guids)).toString();
+ guids.left.addAll(guids.right);
+ return new JSONArray(guids.left).toString();
}
@Override
- public void updateEntityAttributeByGuid(final String guid, String attributeName, String value) throws AtlasException {
+ public String updateEntityAttributeByGuid(final String guid, String attributeName, String value) throws AtlasException {
ParamChecker.notEmpty(guid, "guid cannot be null");
ParamChecker.notEmpty(attributeName, "property cannot be null");
ParamChecker.notEmpty(value, "property value cannot be null");
@@ -426,10 +423,8 @@ public class DefaultMetadataService implements MetadataService {
}
((ReferenceableInstance)newInstance).replaceWithNewId(new Id(guid, 0, newInstance.getTypeName()));
- repository.updatePartial(newInstance);
- onEntitiesUpdated(new ArrayList<ITypedReferenceableInstance>() {{
- add(repository.getEntityDefinition(guid));
- }});
+ TypeUtils.Pair<List<String>, List<String>> guids = repository.updatePartial(newInstance);
+ return onEntitiesAddedUpdated(guids);
}
private ITypedReferenceableInstance validateEntityExists(String guid)
@@ -442,7 +437,7 @@ public class DefaultMetadataService implements MetadataService {
}
@Override
- public void updateEntityPartialByGuid(final String guid, Referenceable newEntity) throws AtlasException {
+ public String updateEntityPartialByGuid(final String guid, Referenceable newEntity) throws AtlasException {
ParamChecker.notEmpty(guid, "guid cannot be null");
ParamChecker.notNull(newEntity, "updatedEntity cannot be null");
ITypedReferenceableInstance existInstance = validateEntityExists(guid);
@@ -450,10 +445,8 @@ public class DefaultMetadataService implements MetadataService {
ITypedReferenceableInstance newInstance = convertToTypedInstance(newEntity, existInstance.getTypeName());
((ReferenceableInstance)newInstance).replaceWithNewId(new Id(guid, 0, newInstance.getTypeName()));
- repository.updatePartial(newInstance);
- onEntitiesUpdated(new ArrayList<ITypedReferenceableInstance>() {{
- add(repository.getEntityDefinition(guid));
- }});
+ TypeUtils.Pair<List<String>, List<String>> guids = repository.updatePartial(newInstance);
+ return onEntitiesAddedUpdated(guids);
}
private ITypedReferenceableInstance convertToTypedInstance(Referenceable updatedEntity, String typeName) throws AtlasException {
@@ -511,13 +504,8 @@ public class DefaultMetadataService implements MetadataService {
final ITypedReferenceableInstance newInstance = convertToTypedInstance(updatedEntity, typeName);
((ReferenceableInstance)newInstance).replaceWithNewId(oldInstance.getId());
- repository.updatePartial(newInstance);
-
- onEntitiesUpdated(new ArrayList<ITypedReferenceableInstance>() {{
- add(newInstance);
- }});
-
- return newInstance.getId()._getId();
+ TypeUtils.Pair<List<String>, List<String>> guids = repository.updatePartial(newInstance);
+ return onEntitiesAddedUpdated(guids);
}
private void validateTypeExists(String entityType) throws AtlasException {
@@ -633,12 +621,22 @@ public class DefaultMetadataService implements MetadataService {
}
}
- private void onEntitiesAdded(Collection<ITypedReferenceableInstance> entities) throws AtlasException {
+ private void onEntitiesAdded(List<String> guids) throws AtlasException {
+ List<ITypedReferenceableInstance> entities = loadEntities(guids);
for (EntityChangeListener listener : entityChangeListeners) {
listener.onEntitiesAdded(entities);
}
}
+ private List<ITypedReferenceableInstance> loadEntities(List<String> guids) throws EntityNotFoundException,
+ RepositoryException {
+ List<ITypedReferenceableInstance> entities = new ArrayList<>();
+ for (String guid : guids) {
+ entities.add(repository.getEntityDefinition(guid));
+ }
+ return entities;
+ }
+
private void onTypesUpdated(Map<String, IDataType> typesUpdated) throws AtlasException {
Map<TypesChangeListener, Throwable> caughtExceptions = new HashMap<>();
for (Provider<TypesChangeListener> indexerProvider : typeChangeListeners) {
@@ -656,8 +654,8 @@ public class DefaultMetadataService implements MetadataService {
}
}
- private void onEntitiesUpdated(Collection<ITypedReferenceableInstance> entities)
- throws AtlasException {
+ private void onEntitiesUpdated(List<String> guids) throws AtlasException {
+ List<ITypedReferenceableInstance> entities = loadEntities(guids);
for (EntityChangeListener listener : entityChangeListeners) {
listener.onEntitiesUpdated(entities);
}
http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/8cbb2c13/repository/src/test/java/org/apache/atlas/BaseHiveRepositoryTest.java
----------------------------------------------------------------------
diff --git a/repository/src/test/java/org/apache/atlas/BaseHiveRepositoryTest.java b/repository/src/test/java/org/apache/atlas/BaseHiveRepositoryTest.java
index 1075d85..a49967f 100644
--- a/repository/src/test/java/org/apache/atlas/BaseHiveRepositoryTest.java
+++ b/repository/src/test/java/org/apache/atlas/BaseHiveRepositoryTest.java
@@ -346,13 +346,10 @@ public class BaseHiveRepositoryTest {
return createInstance(referenceable, clsType);
}
private Id createInstance(Referenceable referenceable, ClassType clsType) throws Exception {
-// String entityJSON = InstanceSerialization.toJson(referenceable, true);
-
-
ITypedReferenceableInstance typedInstance = clsType.convert(referenceable, Multiplicity.REQUIRED);
- String guid = repository.createEntities(typedInstance)[0];
+ List<String> guids = repository.createEntities(typedInstance);
// return the reference to created instance with guid
- return new Id(guid, 0, referenceable.getTypeName());
+ return new Id(guids.get(guids.size() - 1), 0, referenceable.getTypeName());
}
}
http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/8cbb2c13/repository/src/test/java/org/apache/atlas/repository/graph/GraphBackedMetadataRepositoryTest.java
----------------------------------------------------------------------
diff --git a/repository/src/test/java/org/apache/atlas/repository/graph/GraphBackedMetadataRepositoryTest.java b/repository/src/test/java/org/apache/atlas/repository/graph/GraphBackedMetadataRepositoryTest.java
index 78af443..22ff1d6 100755
--- a/repository/src/test/java/org/apache/atlas/repository/graph/GraphBackedMetadataRepositoryTest.java
+++ b/repository/src/test/java/org/apache/atlas/repository/graph/GraphBackedMetadataRepositoryTest.java
@@ -118,10 +118,10 @@ public class GraphBackedMetadataRepositoryTest {
ClassType deptType = typeSystem.getDataType(ClassType.class, "Department");
ITypedReferenceableInstance hrDept2 = deptType.convert(hrDept, Multiplicity.REQUIRED);
- String[] guids = repositoryService.createEntities(hrDept2);
+ List<String> guids = repositoryService.createEntities(hrDept2);
Assert.assertNotNull(guids);
- Assert.assertEquals(guids.length, 1);
- guid = guids[0];
+ Assert.assertEquals(guids.size(), 5);
+ guid = guids.get(4);
Assert.assertNotNull(guid);
}
@@ -173,14 +173,12 @@ public class GraphBackedMetadataRepositoryTest {
ITypedReferenceableInstance db = dbType.convert(databaseInstance, Multiplicity.REQUIRED);
System.out.println("db = " + db);
- String dbGUID = repositoryService.createEntities(db)[0];
- System.out.println("added db = " + dbGUID);
-
- Referenceable dbInstance = new Referenceable(dbGUID, TestUtils.DATABASE_TYPE, databaseInstance.getValuesMap());
-
- ITypedReferenceableInstance table = createHiveTableInstance(dbInstance);
- String tableGUID = repositoryService.createEntities(table)[0];
- System.out.println("added table = " + tableGUID);
+ //Reuse the same database instance without id, with the same unique attribute
+ ITypedReferenceableInstance table = createHiveTableInstance(databaseInstance);
+ List<String> guids = repositoryService.createEntities(db, table);
+ Assert.assertEquals(guids.size(), 7); //1 db + 5 columns + 1 table. Shouldn't create db again
+ System.out.println("added db = " + guids.get(0));
+ System.out.println("added table = " + guids.get(6));
}
@Test(dependsOnMethods = "testCreateEntity")
@@ -600,9 +598,10 @@ public class GraphBackedMetadataRepositoryTest {
ClassType deptType = typeSystem.getDataType(ClassType.class, "Department");
ITypedReferenceableInstance hrDept2 = deptType.convert(hrDept, Multiplicity.REQUIRED);
- String[] guids = repositoryService.createEntities(hrDept2);
+ List<String> guids = repositoryService.createEntities(hrDept2);
Assert.assertNotNull(guids);
- Assert.assertEquals(guids.length, 1);
- Assert.assertNotNull(guids[0]);
+ Assert.assertEquals(guids.size(), 2);
+ Assert.assertNotNull(guids.get(0));
+ Assert.assertNotNull(guids.get(1));
}
}
http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/8cbb2c13/repository/src/test/java/org/apache/atlas/repository/graph/GraphRepoMapperScaleTest.java
----------------------------------------------------------------------
diff --git a/repository/src/test/java/org/apache/atlas/repository/graph/GraphRepoMapperScaleTest.java b/repository/src/test/java/org/apache/atlas/repository/graph/GraphRepoMapperScaleTest.java
index c25ccf8..2d01bbe 100755
--- a/repository/src/test/java/org/apache/atlas/repository/graph/GraphRepoMapperScaleTest.java
+++ b/repository/src/test/java/org/apache/atlas/repository/graph/GraphRepoMapperScaleTest.java
@@ -101,7 +101,7 @@ public class GraphRepoMapperScaleTest {
ClassType dbType = typeSystem.getDataType(ClassType.class, TestUtils.DATABASE_TYPE);
ITypedReferenceableInstance db = dbType.convert(databaseInstance, Multiplicity.REQUIRED);
- dbGUID = repositoryService.createEntities(db)[0];
+ dbGUID = repositoryService.createEntities(db).get(0);
Referenceable dbInstance = new Referenceable(dbGUID, TestUtils.DATABASE_TYPE, databaseInstance.getValuesMap());
http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/8cbb2c13/repository/src/test/java/org/apache/atlas/service/DefaultMetadataServiceTest.java
----------------------------------------------------------------------
diff --git a/repository/src/test/java/org/apache/atlas/service/DefaultMetadataServiceTest.java b/repository/src/test/java/org/apache/atlas/service/DefaultMetadataServiceTest.java
index 0307fd4..8a80d8c 100644
--- a/repository/src/test/java/org/apache/atlas/service/DefaultMetadataServiceTest.java
+++ b/repository/src/test/java/org/apache/atlas/service/DefaultMetadataServiceTest.java
@@ -54,6 +54,7 @@ import java.util.List;
import java.util.Map;
import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertTrue;
@Guice(modules = RepositoryMetadataModule.class)
@@ -113,7 +114,11 @@ public class DefaultMetadataServiceTest {
JSONArray entitiesJson = new JSONArray();
entitiesJson.put(entityjson);
String response = metadataService.createEntities(entitiesJson.toString());
- return new JSONArray(response).getString(0);
+ JSONArray guids = new JSONArray(response);
+ if (guids != null && guids.length() > 0) {
+ return guids.getString(0);
+ }
+ return null;
}
private String updateInstance(Referenceable entity) throws Exception {
@@ -154,7 +159,7 @@ public class DefaultMetadataServiceTest {
//using the same name should succeed, but not create another entity
String newId = createInstance(entity);
- Assert.assertEquals(newId, id);
+ assertNull(newId);
//Same entity, but different qualified name should succeed
entity.set("name", TestUtils.randomString());
http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/8cbb2c13/server-api/src/main/java/org/apache/atlas/services/MetadataService.java
----------------------------------------------------------------------
diff --git a/server-api/src/main/java/org/apache/atlas/services/MetadataService.java b/server-api/src/main/java/org/apache/atlas/services/MetadataService.java
index 0cfed2e..a91c0a0 100644
--- a/server-api/src/main/java/org/apache/atlas/services/MetadataService.java
+++ b/server-api/src/main/java/org/apache/atlas/services/MetadataService.java
@@ -74,7 +74,7 @@ public interface MetadataService {
* Creates an entity, instance of the type.
*
* @param entityDefinition definition
- * @return guid
+ * @return json array of guids of entities created
*/
String createEntities(String entityDefinition) throws AtlasException;
@@ -107,25 +107,28 @@ public interface MetadataService {
/**
* Adds the property to the given entity id(guid).
* Currently supports updates only on PRIMITIVE, CLASS attribute types
- *
- * @param guid entity id
+ * @param guid entity id
* @param attribute property name
* @param value property value
+ * @return json array of guids of entities created/updated
*/
- void updateEntityAttributeByGuid(String guid, String attribute, String value) throws AtlasException;
+ String updateEntityAttributeByGuid(String guid, String attribute, String value) throws AtlasException;
/**
* Supports Partial updates of an entity. Users can update a subset of attributes for an entity identified by its guid
* Note however that it cannot be used to set attribute values to null or delete attrbute values
- *
+ * @param guid entity id
+ * @param entity
+ * @return json array of guids of entities created/updated
+ * @throws AtlasException
*/
- void updateEntityPartialByGuid(String guid, Referenceable entity) throws AtlasException;
+ String updateEntityPartialByGuid(String guid, Referenceable entity) throws AtlasException;
/**
* Batch API - Adds/Updates the given entity id(guid).
*
* @param entityJson entity json
- * @return List of guids which were updated and ones which were newly created as part of the updated entity
+ * @return json array of guids of entities created/updated
*/
String updateEntities(String entityJson) throws AtlasException;
http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/8cbb2c13/typesystem/src/main/java/org/apache/atlas/typesystem/types/TypeUtils.java
----------------------------------------------------------------------
diff --git a/typesystem/src/main/java/org/apache/atlas/typesystem/types/TypeUtils.java b/typesystem/src/main/java/org/apache/atlas/typesystem/types/TypeUtils.java
index 9d2480b..f5c2ce9 100755
--- a/typesystem/src/main/java/org/apache/atlas/typesystem/types/TypeUtils.java
+++ b/typesystem/src/main/java/org/apache/atlas/typesystem/types/TypeUtils.java
@@ -70,14 +70,18 @@ public class TypeUtils {
return b;
}
- protected static class Pair<L, R> {
- protected L left;
- protected R right;
+ public static class Pair<L, R> {
+ public L left;
+ public R right;
public Pair(L left, R right) {
this.left = left;
this.right = right;
}
+
+ public static <L, R> Pair<L, R> of(L left, R right) {
+ return new Pair<>(left, right);
+ }
}
/**
http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/8cbb2c13/webapp/src/main/java/org/apache/atlas/web/resources/EntityResource.java
----------------------------------------------------------------------
diff --git a/webapp/src/main/java/org/apache/atlas/web/resources/EntityResource.java b/webapp/src/main/java/org/apache/atlas/web/resources/EntityResource.java
index 2ee0027..f5ab4d8 100755
--- a/webapp/src/main/java/org/apache/atlas/web/resources/EntityResource.java
+++ b/webapp/src/main/java/org/apache/atlas/web/resources/EntityResource.java
@@ -118,8 +118,12 @@ public class EntityResource {
JSONObject response = new JSONObject();
response.put(AtlasClient.REQUEST_ID, Servlets.getRequestId());
- response.put(AtlasClient.GUID, new JSONArray(guids));
- response.put(AtlasClient.DEFINITION, new JSONObject(metadataService.getEntityDefinition(new JSONArray(guids).getString(0))));
+ JSONArray guidArray = new JSONArray(guids);
+ response.put(AtlasClient.GUID, guidArray);
+ if (guidArray.length() > 0) {
+ response.put(AtlasClient.DEFINITION,
+ new JSONObject(metadataService.getEntityDefinition(new JSONArray(guids).getString(0))));
+ }
return Response.created(locationURI).entity(response).build();
http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/8cbb2c13/webapp/src/test/java/org/apache/atlas/notification/EntityNotificationIT.java
----------------------------------------------------------------------
diff --git a/webapp/src/test/java/org/apache/atlas/notification/EntityNotificationIT.java b/webapp/src/test/java/org/apache/atlas/notification/EntityNotificationIT.java
index 4d2cce7..b2501ec 100644
--- a/webapp/src/test/java/org/apache/atlas/notification/EntityNotificationIT.java
+++ b/webapp/src/test/java/org/apache/atlas/notification/EntityNotificationIT.java
@@ -58,7 +58,6 @@ public class EntityNotificationIT extends BaseResourceIT {
private static final String ENTITIES = "api/atlas/entities";
private static final String TRAITS = "traits";
- private static final int MAX_WAIT_TIME = 10000;
private final String DATABASE_NAME = "db" + randomString();
private final String TABLE_NAME = "table" + randomString();
@Inject
@@ -98,7 +97,7 @@ public class EntityNotificationIT extends BaseResourceIT {
final String guid = tableId._getId();
- waitForNotification(MAX_WAIT_TIME);
+ waitForNotification(notificationConsumer, MAX_WAIT_TIME);
EntityNotification entityNotification = notificationConsumer.getLastEntityNotification();
@@ -120,7 +119,7 @@ public class EntityNotificationIT extends BaseResourceIT {
serviceClient.updateEntityAttribute(guid, property, newValue);
- waitForNotification(MAX_WAIT_TIME);
+ waitForNotification(notificationConsumer, MAX_WAIT_TIME);
EntityNotification entityNotification = notificationConsumer.getLastEntityNotification();
@@ -155,7 +154,7 @@ public class EntityNotificationIT extends BaseResourceIT {
ClientResponse clientResponse = addTrait(guid, traitInstanceJSON);
assertEquals(clientResponse.getStatus(), Response.Status.CREATED.getStatusCode());
- waitForNotification(MAX_WAIT_TIME);
+ waitForNotification(notificationConsumer, MAX_WAIT_TIME);
EntityNotification entityNotification = notificationConsumer.getLastEntityNotification();
@@ -192,7 +191,7 @@ public class EntityNotificationIT extends BaseResourceIT {
clientResponse = addTrait(guid, traitInstanceJSON);
assertEquals(clientResponse.getStatus(), Response.Status.CREATED.getStatusCode());
- waitForNotification(MAX_WAIT_TIME);
+ waitForNotification(notificationConsumer, MAX_WAIT_TIME);
entityNotification = notificationConsumer.getLastEntityNotification();
@@ -218,7 +217,7 @@ public class EntityNotificationIT extends BaseResourceIT {
ClientResponse clientResponse = deleteTrait(guid, traitName);
Assert.assertEquals(clientResponse.getStatus(), Response.Status.OK.getStatusCode());
- waitForNotification(MAX_WAIT_TIME);
+ waitForNotification(notificationConsumer, MAX_WAIT_TIME);
EntityNotification entityNotification = notificationConsumer.getLastEntityNotification();
@@ -260,51 +259,4 @@ public class EntityNotificationIT extends BaseResourceIT {
return resource.accept(Servlets.JSON_MEDIA_TYPE).type(Servlets.JSON_MEDIA_TYPE)
.method(HttpMethod.DELETE, ClientResponse.class);
}
-
- private void waitForNotification(int maxWait) throws Exception {
- waitFor(maxWait, new Predicate() {
- @Override
- public boolean evaluate() throws Exception {
- return notificationConsumer.getLastEntityNotification() != null;
- }
- });
- }
-
-
- // ----- inner class : EntityNotificationConsumer --------------------------
-
- private static class EntityNotificationConsumer implements Runnable {
- private final NotificationConsumer<EntityNotification> consumerIterator;
- private EntityNotification entityNotification = null;
- private boolean run;
-
- public EntityNotificationConsumer(NotificationConsumer<EntityNotification> consumerIterator) {
- this.consumerIterator = consumerIterator;
- }
-
- @Override
- public void run() {
- while (run && consumerIterator.hasNext()) {
- entityNotification = consumerIterator.next();
- }
- }
-
- public void reset() {
- entityNotification = null;
- }
-
- public void start() {
- Thread thread = new Thread(this);
- run = true;
- thread.start();
- }
-
- public void stop() {
- run = false;
- }
-
- public EntityNotification getLastEntityNotification() {
- return entityNotification;
- }
- }
}
http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/8cbb2c13/webapp/src/test/java/org/apache/atlas/web/resources/BaseResourceIT.java
----------------------------------------------------------------------
diff --git a/webapp/src/test/java/org/apache/atlas/web/resources/BaseResourceIT.java b/webapp/src/test/java/org/apache/atlas/web/resources/BaseResourceIT.java
index 361cece..aba191c 100755
--- a/webapp/src/test/java/org/apache/atlas/web/resources/BaseResourceIT.java
+++ b/webapp/src/test/java/org/apache/atlas/web/resources/BaseResourceIT.java
@@ -25,6 +25,8 @@ import com.sun.jersey.api.client.ClientResponse;
import com.sun.jersey.api.client.WebResource;
import com.sun.jersey.api.client.config.DefaultClientConfig;
import org.apache.atlas.*;
+import org.apache.atlas.notification.NotificationConsumer;
+import org.apache.atlas.notification.entity.EntityNotification;
import org.apache.atlas.typesystem.Referenceable;
import org.apache.atlas.typesystem.Struct;
import org.apache.atlas.typesystem.TypesDef;
@@ -68,6 +70,7 @@ public abstract class BaseResourceIT {
protected WebResource service;
protected AtlasClient serviceClient;
public static final Logger LOG = LoggerFactory.getLogger(BaseResourceIT.class);
+ protected static final int MAX_WAIT_TIME = 1000;
@BeforeClass
public void setUp() throws Exception {
@@ -119,7 +122,10 @@ public abstract class BaseResourceIT {
System.out.println("created instance for type " + typeName + ", guid: " + guids);
// return the reference to created instance with guid
- return new Id(guids.getString(0), 0, referenceable.getTypeName());
+ if (guids.length() > 0) {
+ return new Id(guids.getString(guids.length() - 1), 0, referenceable.getTypeName());
+ }
+ return null;
}
protected static final String DATABASE_TYPE = "hive_db";
@@ -285,4 +291,50 @@ public abstract class BaseResourceIT {
throw new Exception("Waiting timed out after " + timeout + " msec");
}
}
+
+ // ----- inner class : EntityNotificationConsumer --------------------------
+
+ protected static class EntityNotificationConsumer implements Runnable {
+ private final NotificationConsumer<EntityNotification> consumerIterator;
+ private EntityNotification entityNotification = null;
+ private boolean run;
+
+ public EntityNotificationConsumer(NotificationConsumer<EntityNotification> consumerIterator) {
+ this.consumerIterator = consumerIterator;
+ }
+
+ @Override
+ public void run() {
+ while (run && consumerIterator.hasNext()) {
+ entityNotification = consumerIterator.next();
+ }
+ }
+
+ public void reset() {
+ entityNotification = null;
+ }
+
+ public void start() {
+ Thread thread = new Thread(this);
+ run = true;
+ thread.start();
+ }
+
+ public void stop() {
+ run = false;
+ }
+
+ public EntityNotification getLastEntityNotification() {
+ return entityNotification;
+ }
+ }
+
+ protected void waitForNotification(final EntityNotificationConsumer notificationConsumer, int maxWait) throws Exception {
+ waitFor(maxWait, new Predicate() {
+ @Override
+ public boolean evaluate() throws Exception {
+ return notificationConsumer.getLastEntityNotification() != null;
+ }
+ });
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/8cbb2c13/webapp/src/test/java/org/apache/atlas/web/resources/EntityJerseyResourceIT.java
----------------------------------------------------------------------
diff --git a/webapp/src/test/java/org/apache/atlas/web/resources/EntityJerseyResourceIT.java b/webapp/src/test/java/org/apache/atlas/web/resources/EntityJerseyResourceIT.java
index f476af3..a268196 100755
--- a/webapp/src/test/java/org/apache/atlas/web/resources/EntityJerseyResourceIT.java
+++ b/webapp/src/test/java/org/apache/atlas/web/resources/EntityJerseyResourceIT.java
@@ -19,10 +19,15 @@
package org.apache.atlas.web.resources;
import com.google.common.collect.ImmutableList;
+import com.google.inject.Inject;
import com.sun.jersey.api.client.ClientResponse;
import com.sun.jersey.api.client.WebResource;
import org.apache.atlas.AtlasClient;
import org.apache.atlas.AtlasServiceException;
+import org.apache.atlas.notification.NotificationConsumer;
+import org.apache.atlas.notification.NotificationInterface;
+import org.apache.atlas.notification.NotificationModule;
+import org.apache.atlas.notification.entity.EntityNotification;
import org.apache.atlas.typesystem.IStruct;
import org.apache.atlas.typesystem.Referenceable;
import org.apache.atlas.typesystem.Struct;
@@ -43,11 +48,14 @@ import org.apache.atlas.web.util.Servlets;
import org.apache.commons.lang.RandomStringUtils;
import org.codehaus.jettison.json.JSONArray;
import org.codehaus.jettison.json.JSONObject;
+import org.junit.AfterClass;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testng.Assert;
import org.testng.annotations.BeforeClass;
+import org.testng.annotations.BeforeMethod;
import org.testng.annotations.DataProvider;
+import org.testng.annotations.Guice;
import org.testng.annotations.Test;
import javax.ws.rs.HttpMethod;
@@ -59,10 +67,13 @@ import java.util.Map;
import java.util.UUID;
import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.fail;
/**
* Integration tests for Entity Jersey Resource.
*/
+@Guice(modules = NotificationModule.class)
public class EntityJerseyResourceIT extends BaseResourceIT {
private static final Logger LOG = LoggerFactory.getLogger(EntityJerseyResourceIT.class);
@@ -76,11 +87,32 @@ public class EntityJerseyResourceIT extends BaseResourceIT {
private Id tableId;
private String traitName;
+ @Inject
+ private NotificationInterface notificationInterface;
+ private EntityNotificationConsumer notificationConsumer;
+
@BeforeClass
public void setUp() throws Exception {
super.setUp();
createTypeDefinitions();
+
+ List<NotificationConsumer<EntityNotification>> consumers =
+ notificationInterface.createConsumers(NotificationInterface.NotificationType.ENTITIES, 1);
+
+ NotificationConsumer<EntityNotification> consumer = consumers.iterator().next();
+ notificationConsumer = new EntityNotificationConsumer(consumer);
+ notificationConsumer.start();
+ }
+
+ @AfterClass
+ public void tearDown() {
+ notificationConsumer.stop();
+ }
+
+ @BeforeMethod
+ public void setupTest() {
+ notificationConsumer.reset();
}
@Test
@@ -119,18 +151,32 @@ public class EntityJerseyResourceIT extends BaseResourceIT {
@Test
public void testEntityDeduping() throws Exception {
- Referenceable db = new Referenceable(DATABASE_TYPE);
- String dbName = "db" + randomString();
+ final Referenceable db = new Referenceable(DATABASE_TYPE);
+ final String dbName = "db" + randomString();
db.set("name", dbName);
db.set("description", randomString());
- serviceClient.createEntity(db);
+ serviceClient.createEntity(db).getString(0);
+
+ waitForNotification(notificationConsumer, MAX_WAIT_TIME);
+ EntityNotification notification = notificationConsumer.getLastEntityNotification();
+ assertNotNull(notification);
+ assertEquals(notification.getEntity().get("name"), dbName);
+
JSONArray results =
serviceClient.searchByDSL(String.format("%s where name='%s'", DATABASE_TYPE, dbName));
assertEquals(results.length(), 1);
//create entity again shouldn't create another instance with same unique attribute value
+ notificationConsumer.reset();
serviceClient.createEntity(db);
+ try {
+ waitForNotification(notificationConsumer, MAX_WAIT_TIME);
+ fail("Expected time out exception");
+ } catch (Exception e) {
+ //expected timeout
+ }
+
results = serviceClient.searchByDSL(String.format("%s where name='%s'", DATABASE_TYPE, dbName));
assertEquals(results.length(), 1);