You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@atlas.apache.org by ma...@apache.org on 2017/11/06 22:50:05 UTC
[1/3] atlas git commit: ATLAS-2251: notification module updates
Repository: atlas
Updated Branches:
refs/heads/ATLAS-2251 3f44770da -> 64e739da7
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/notification/src/test/java/org/apache/atlas/notification/entity/EntityNotificationImplTest.java
----------------------------------------------------------------------
diff --git a/notification/src/test/java/org/apache/atlas/notification/entity/EntityNotificationImplTest.java b/notification/src/test/java/org/apache/atlas/notification/entity/EntityNotificationImplTest.java
deleted file mode 100644
index 81a0784..0000000
--- a/notification/src/test/java/org/apache/atlas/notification/entity/EntityNotificationImplTest.java
+++ /dev/null
@@ -1,147 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.atlas.notification.entity;
-
-import org.apache.atlas.v1.model.instance.Referenceable;
-import org.apache.atlas.v1.model.instance.Struct;
-import org.apache.atlas.type.AtlasClassificationType;
-import org.apache.atlas.type.AtlasTypeRegistry;
-import org.testng.annotations.Test;
-
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-import static org.testng.Assert.assertEquals;
-import static org.testng.Assert.assertTrue;
-
-/**
- * EntityNotificationImpl tests.
- */
-public class EntityNotificationImplTest {
-
- @Test
- public void testGetEntity() throws Exception {
- Referenceable entity = getEntity("id");
-
- EntityNotificationImpl entityNotification =
- new EntityNotificationImpl(entity, EntityNotification.OperationType.ENTITY_CREATE,
- Collections.<Struct>emptyList());
-
- assertEquals(entity, entityNotification.getEntity());
- }
-
- @Test
- public void testGetOperationType() throws Exception {
- Referenceable entity = getEntity("id");
-
- EntityNotificationImpl entityNotification =
- new EntityNotificationImpl(entity, EntityNotification.OperationType.ENTITY_CREATE,
- Collections.<Struct>emptyList());
-
- assertEquals(EntityNotification.OperationType.ENTITY_CREATE, entityNotification.getOperationType());
- }
-
- @Test
- public void testGetAllTraits() throws Exception {
- Referenceable entity = getEntity("id");
- String traitName = "MyTrait";
- List<Struct> traitInfo = new LinkedList<>();
- Struct trait = new Struct(traitName, Collections.<String, Object>emptyMap());
- traitInfo.add(trait);
-
- EntityNotificationImpl entityNotification =
- new EntityNotificationImpl(entity, EntityNotification.OperationType.TRAIT_ADD, traitInfo);
-
- assertEquals(traitInfo, entityNotification.getAllTraits());
- }
-
- @Test
- public void testGetAllTraitsSuperTraits() throws Exception {
- AtlasTypeRegistry typeRegistry = mock(AtlasTypeRegistry.class);
-
- String traitName = "MyTrait";
- Struct myTrait = new Struct(traitName);
-
- String superTraitName = "MySuperTrait";
-
- AtlasClassificationType traitType = mock(AtlasClassificationType.class);
- Set<String> superTypeNames = Collections.singleton(superTraitName);
-
- AtlasClassificationType superTraitType = mock(AtlasClassificationType.class);
- Set<String> superSuperTypeNames = Collections.emptySet();
-
- Referenceable entity = getEntity("id", myTrait);
-
- when(typeRegistry.getClassificationTypeByName(traitName)).thenReturn(traitType);
- when(typeRegistry.getClassificationTypeByName(superTraitName)).thenReturn(superTraitType);
-
- when(traitType.getAllSuperTypes()).thenReturn(superTypeNames);
- when(superTraitType.getAllSuperTypes()).thenReturn(superSuperTypeNames);
-
- EntityNotificationImpl entityNotification =
- new EntityNotificationImpl(entity, EntityNotification.OperationType.TRAIT_ADD, typeRegistry);
-
- List<Struct> allTraits = entityNotification.getAllTraits();
-
- assertEquals(2, allTraits.size());
-
- for (Struct trait : allTraits) {
- String typeName = trait.getTypeName();
- assertTrue(typeName.equals(traitName) || typeName.equals(superTraitName));
- }
- }
-
- @Test
- public void testEquals() throws Exception {
- Referenceable entity = getEntity("id");
-
- EntityNotificationImpl entityNotification2 =
- new EntityNotificationImpl(entity, EntityNotification.OperationType.ENTITY_CREATE,
- Collections.<Struct>emptyList());
-
- EntityNotificationImpl entityNotification =
- new EntityNotificationImpl(entity, EntityNotification.OperationType.ENTITY_CREATE,
- Collections.<Struct>emptyList());
-
- assertTrue(entityNotification.equals(entityNotification2));
- assertTrue(entityNotification2.equals(entityNotification));
- }
-
- public static Referenceable getEntity(String id, Struct... traits) {
- String typeName = "typeName";
- Map<String, Object> values = new HashMap<>();
-
- List<String> traitNames = new LinkedList<>();
- Map<String, Struct> traitMap = new HashMap<>();
-
- for (Struct trait : traits) {
- String traitName = trait.getTypeName();
-
- traitNames.add(traitName);
- traitMap.put(traitName, trait);
- }
- return new Referenceable(id, typeName, values, traitNames, traitMap);
- }
-}
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/notification/src/test/java/org/apache/atlas/notification/entity/EntityNotificationTest.java
----------------------------------------------------------------------
diff --git a/notification/src/test/java/org/apache/atlas/notification/entity/EntityNotificationTest.java b/notification/src/test/java/org/apache/atlas/notification/entity/EntityNotificationTest.java
new file mode 100644
index 0000000..cedfc01
--- /dev/null
+++ b/notification/src/test/java/org/apache/atlas/notification/entity/EntityNotificationTest.java
@@ -0,0 +1,148 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.atlas.notification.entity;
+
+import org.apache.atlas.v1.model.instance.Referenceable;
+import org.apache.atlas.v1.model.instance.Struct;
+import org.apache.atlas.type.AtlasClassificationType;
+import org.apache.atlas.type.AtlasTypeRegistry;
+import org.apache.atlas.v1.model.notification.EntityNotification;
+import org.testng.annotations.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+
+/**
+ * EntityNotification tests.
+ */
+public class EntityNotificationTest {
+
+ @Test
+ public void testGetEntity() throws Exception {
+ Referenceable entity = getEntity("id");
+
+ EntityNotification entityNotification =
+ new EntityNotification(entity, EntityNotification.OperationType.ENTITY_CREATE,
+ Collections.<Struct>emptyList());
+
+ assertEquals(entity, entityNotification.getEntity());
+ }
+
+ @Test
+ public void testGetOperationType() throws Exception {
+ Referenceable entity = getEntity("id");
+
+ EntityNotification entityNotification =
+ new EntityNotification(entity, EntityNotification.OperationType.ENTITY_CREATE,
+ Collections.<Struct>emptyList());
+
+ assertEquals(EntityNotification.OperationType.ENTITY_CREATE, entityNotification.getOperationType());
+ }
+
+ @Test
+ public void testGetAllTraits() throws Exception {
+ Referenceable entity = getEntity("id");
+ String traitName = "MyTrait";
+ List<Struct> traitInfo = new LinkedList<>();
+ Struct trait = new Struct(traitName, Collections.<String, Object>emptyMap());
+ traitInfo.add(trait);
+
+ EntityNotification entityNotification =
+ new EntityNotification(entity, EntityNotification.OperationType.TRAIT_ADD, traitInfo);
+
+ assertEquals(traitInfo, entityNotification.getAllTraits());
+ }
+
+ @Test
+ public void testGetAllTraitsSuperTraits() throws Exception {
+ AtlasTypeRegistry typeRegistry = mock(AtlasTypeRegistry.class);
+
+ String traitName = "MyTrait";
+ Struct myTrait = new Struct(traitName);
+
+ String superTraitName = "MySuperTrait";
+
+ AtlasClassificationType traitType = mock(AtlasClassificationType.class);
+ Set<String> superTypeNames = Collections.singleton(superTraitName);
+
+ AtlasClassificationType superTraitType = mock(AtlasClassificationType.class);
+ Set<String> superSuperTypeNames = Collections.emptySet();
+
+ Referenceable entity = getEntity("id", myTrait);
+
+ when(typeRegistry.getClassificationTypeByName(traitName)).thenReturn(traitType);
+ when(typeRegistry.getClassificationTypeByName(superTraitName)).thenReturn(superTraitType);
+
+ when(traitType.getAllSuperTypes()).thenReturn(superTypeNames);
+ when(superTraitType.getAllSuperTypes()).thenReturn(superSuperTypeNames);
+
+ EntityNotification entityNotification =
+ new EntityNotification(entity, EntityNotification.OperationType.TRAIT_ADD, typeRegistry);
+
+ List<Struct> allTraits = entityNotification.getAllTraits();
+
+ assertEquals(2, allTraits.size());
+
+ for (Struct trait : allTraits) {
+ String typeName = trait.getTypeName();
+ assertTrue(typeName.equals(traitName) || typeName.equals(superTraitName));
+ }
+ }
+
+ @Test
+ public void testEquals() throws Exception {
+ Referenceable entity = getEntity("id");
+
+ EntityNotification entityNotification2 =
+ new EntityNotification(entity, EntityNotification.OperationType.ENTITY_CREATE,
+ Collections.<Struct>emptyList());
+
+ EntityNotification entityNotification =
+ new EntityNotification(entity, EntityNotification.OperationType.ENTITY_CREATE,
+ Collections.<Struct>emptyList());
+
+ assertTrue(entityNotification.equals(entityNotification2));
+ assertTrue(entityNotification2.equals(entityNotification));
+ }
+
+ public static Referenceable getEntity(String id, Struct... traits) {
+ String typeName = "typeName";
+ Map<String, Object> values = new HashMap<>();
+
+ List<String> traitNames = new LinkedList<>();
+ Map<String, Struct> traitMap = new HashMap<>();
+
+ for (Struct trait : traits) {
+ String traitName = trait.getTypeName();
+
+ traitNames.add(traitName);
+ traitMap.put(traitName, trait);
+ }
+ return new Referenceable(id, typeName, values, traitNames, traitMap);
+ }
+}
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/notification/src/test/java/org/apache/atlas/notification/hook/HookMessageDeserializerTest.java
----------------------------------------------------------------------
diff --git a/notification/src/test/java/org/apache/atlas/notification/hook/HookMessageDeserializerTest.java b/notification/src/test/java/org/apache/atlas/notification/hook/HookMessageDeserializerTest.java
index 758abad..17facf8 100644
--- a/notification/src/test/java/org/apache/atlas/notification/hook/HookMessageDeserializerTest.java
+++ b/notification/src/test/java/org/apache/atlas/notification/hook/HookMessageDeserializerTest.java
@@ -18,12 +18,12 @@
package org.apache.atlas.notification.hook;
+import org.apache.atlas.notification.entity.EntityNotificationTest;
import org.apache.atlas.v1.model.instance.Referenceable;
import org.apache.atlas.v1.model.instance.Struct;
import org.apache.atlas.notification.AbstractNotification;
-import org.apache.atlas.notification.entity.EntityNotificationImplTest;
-import org.apache.atlas.notification.hook.HookNotification.EntityUpdateRequest;
-import org.apache.atlas.notification.hook.HookNotification.HookNotificationMessage;
+import org.apache.atlas.v1.model.notification.HookNotification.EntityUpdateRequest;
+import org.apache.atlas.v1.model.notification.HookNotification.HookNotificationMessage;
import org.apache.atlas.type.AtlasType;
import org.apache.commons.lang3.RandomStringUtils;
import org.testng.annotations.Test;
@@ -40,7 +40,7 @@ import static org.testng.Assert.assertTrue;
* HookMessageDeserializer tests.
*/
public class HookMessageDeserializerTest {
- HookMessageDeserializer deserializer = new HookMessageDeserializer();
+ private HookMessageDeserializer deserializer = new HookMessageDeserializer();
@Test
public void testDeserialize() throws Exception {
@@ -63,7 +63,7 @@ public class HookMessageDeserializerTest {
EntityUpdateRequest message = new EntityUpdateRequest("user1", entity);
String jsonMsg = AtlasType.toV1Json(message);
- HookNotificationMessage deserializedMessage = deserializer.deserialize(jsonMsg);
+ HookNotificationMessage deserializedMessage = deserialize(Collections.singletonList(jsonMsg));
assertEqualMessage(deserializedMessage, message);
}
@@ -106,7 +106,7 @@ public class HookMessageDeserializerTest {
}
private Referenceable generateEntityWithTrait() {
- Referenceable ret = EntityNotificationImplTest.getEntity("id", new Struct("MyTrait", Collections.<String, Object>emptyMap()));
+ Referenceable ret = EntityNotificationTest.getEntity("id", new Struct("MyTrait", Collections.<String, Object>emptyMap()));
return ret;
}
@@ -145,7 +145,7 @@ public class HookMessageDeserializerTest {
}
private Referenceable generateLargeEntityWithTrait() {
- Referenceable ret = EntityNotificationImplTest.getEntity("id", new Struct("MyTrait", Collections.<String, Object>emptyMap()));
+ Referenceable ret = EntityNotificationTest.getEntity("id", new Struct("MyTrait", Collections.<String, Object>emptyMap()));
// add 100 attributes, each with value of size 10k
// Json Size=1,027,984; GZipped Size=16,387 ==> will compress, but not split
@@ -158,7 +158,7 @@ public class HookMessageDeserializerTest {
}
private Referenceable generateVeryLargeEntityWithTrait() {
- Referenceable ret = EntityNotificationImplTest.getEntity("id", new Struct("MyTrait", Collections.<String, Object>emptyMap()));
+ Referenceable ret = EntityNotificationTest.getEntity("id", new Struct("MyTrait", Collections.<String, Object>emptyMap()));
// add 300 attributes, each with value of size 10k
// Json Size=3,082,384; GZipped Size=2,313,357 ==> will compress & split
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/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
index 9ea688c..ec2218c 100644
--- a/notification/src/test/java/org/apache/atlas/notification/hook/HookNotificationTest.java
+++ b/notification/src/test/java/org/apache/atlas/notification/hook/HookNotificationTest.java
@@ -19,6 +19,7 @@ package org.apache.atlas.notification.hook;
import org.apache.atlas.v1.model.instance.Referenceable;
import org.apache.atlas.type.AtlasType;
+import org.apache.atlas.v1.model.notification.HookNotification;
import org.testng.annotations.Test;
import static org.testng.Assert.assertEquals;
@@ -26,8 +27,6 @@ import static org.testng.Assert.assertNull;
public class HookNotificationTest {
- public static final HookMessageDeserializer HOOK_MESSAGE_DESERIALIZER = new HookMessageDeserializer();
-
@Test
public void testNewMessageSerDe() throws Exception {
Referenceable entity1 = new Referenceable("sometype");
@@ -38,8 +37,7 @@ public class HookNotificationTest {
HookNotification.EntityCreateRequest request = new HookNotification.EntityCreateRequest(user, entity1, entity2);
String notificationJson = AtlasType.toV1Json(request);
- HookNotification.HookNotificationMessage actualNotification =
- HOOK_MESSAGE_DESERIALIZER.deserialize(notificationJson);
+ HookNotification.HookNotificationMessage actualNotification = AtlasType.fromV1Json(notificationJson, HookNotification.HookNotificationMessage.class);
assertEquals(actualNotification.getType(), HookNotification.HookNotificationType.ENTITY_CREATE);
assertEquals(actualNotification.getUser(), user);
@@ -49,7 +47,9 @@ public class HookNotificationTest {
Referenceable actualEntity1 = createRequest.getEntities().get(0);
assertEquals(actualEntity1.getTypeName(), "sometype");
+ /* TODO:
assertEquals(((Referenceable)actualEntity1.get("complex")).getTypeName(), "othertype");
+ */
assertEquals(createRequest.getEntities().get(1).getTypeName(), "newtype");
}
@@ -87,11 +87,9 @@ public class HookNotificationTest {
+ "}";
- HookNotification.HookNotificationMessage actualNotification =
- HOOK_MESSAGE_DESERIALIZER.deserialize(notificationJson);
+ HookNotification.HookNotificationMessage actualNotification = AtlasType.fromV1Json(notificationJson, HookNotification.HookNotificationMessage.class);
assertEquals(actualNotification.getType(), HookNotification.HookNotificationType.ENTITY_CREATE);
- assertNull(actualNotification.user);
assertEquals(actualNotification.getUser(), HookNotification.HookNotificationMessage.UNKNOW_USER);
}
}
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/repository/src/test/java/org/apache/atlas/DBSandboxer.java
----------------------------------------------------------------------
diff --git a/repository/src/test/java/org/apache/atlas/DBSandboxer.java b/repository/src/test/java/org/apache/atlas/DBSandboxer.java
new file mode 100644
index 0000000..f4f099a
--- /dev/null
+++ b/repository/src/test/java/org/apache/atlas/DBSandboxer.java
@@ -0,0 +1,53 @@
+/**
+ * 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;
+
+import org.apache.atlas.graph.GraphSandboxUtil;
+import org.apache.atlas.repository.graph.AtlasGraphProvider;
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.testng.ITestContext;
+import org.testng.TestListenerAdapter;
+import org.testng.xml.XmlClass;
+
+import java.util.List;
+
+public class DBSandboxer extends TestListenerAdapter {
+ @Override
+ public void onStart(ITestContext context) {
+ // This will only work if each test is run individually (test suite has only one running test)
+ // If there are multiple tests the the sandbox folder name is not provided and the GraphSandboxUtil provisions
+ // a unique name
+ List<XmlClass> testClassesToRun = context.getCurrentXmlTest().getClasses();
+ if (CollectionUtils.isNotEmpty(testClassesToRun) && 1 == testClassesToRun.size()) {
+ XmlClass currentTestClass = testClassesToRun.get(0);
+ if (null != currentTestClass && StringUtils.isNotEmpty(currentTestClass.getName())) {
+ GraphSandboxUtil.create(currentTestClass.getName());
+ } else {
+ GraphSandboxUtil.create();
+ }
+ } else {
+ GraphSandboxUtil.create();
+ }
+ }
+
+ @Override
+ public void onFinish(ITestContext context) {
+ AtlasGraphProvider.cleanup();
+ }
+}
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/webapp/src/main/java/org/apache/atlas/notification/NotificationEntityChangeListener.java
----------------------------------------------------------------------
diff --git a/webapp/src/main/java/org/apache/atlas/notification/NotificationEntityChangeListener.java b/webapp/src/main/java/org/apache/atlas/notification/NotificationEntityChangeListener.java
index 8f374b9..acbc996 100644
--- a/webapp/src/main/java/org/apache/atlas/notification/NotificationEntityChangeListener.java
+++ b/webapp/src/main/java/org/apache/atlas/notification/NotificationEntityChangeListener.java
@@ -23,8 +23,7 @@ import org.apache.atlas.AtlasException;
import org.apache.atlas.listener.EntityChangeListener;
import org.apache.atlas.v1.model.instance.Referenceable;
import org.apache.atlas.v1.model.instance.Struct;
-import org.apache.atlas.notification.entity.EntityNotification;
-import org.apache.atlas.notification.entity.EntityNotificationImpl;
+import org.apache.atlas.v1.model.notification.EntityNotification;
import org.apache.atlas.repository.graph.GraphHelper;
import org.apache.atlas.type.AtlasClassificationType;
import org.apache.atlas.type.AtlasTypeRegistry;
@@ -41,13 +40,13 @@ import java.util.*;
*/
@Component
public class NotificationEntityChangeListener implements EntityChangeListener {
+ private static final String ATLAS_ENTITY_NOTIFICATION_PROPERTY = "atlas.notification.entity";
- private final NotificationInterface notificationInterface;
- private final AtlasTypeRegistry typeRegistry;
+ private final NotificationInterface notificationInterface;
+ private final AtlasTypeRegistry typeRegistry;
+ private final Map<String, List<String>> notificationAttributesCache = new HashMap<>();
- private Map<String, List<String>> notificationAttributesCache = new HashMap<>();
- private static final String ATLAS_ENTITY_NOTIFICATION_PROPERTY = "atlas.notification.entity";
- static Configuration APPLICATION_PROPERTIES = null;
+ private static Configuration APPLICATION_PROPERTIES = null;
@@ -122,11 +121,17 @@ public class NotificationEntityChangeListener implements EntityChangeListener {
AtlasClassificationType superType = typeRegistry.getClassificationTypeByName(superTypeName);
if (superType != null && MapUtils.isNotEmpty(superType.getAllAttributes())) {
- Map<String, Object> attributes = new HashMap<>();
+ Map<String, Object> superTypeTraitAttributes = new HashMap<>();
+
+ for (Map.Entry<String, Object> attrEntry : trait.getValues().entrySet()) {
+ String attrName = attrEntry.getKey();
- // TODO: add superTypeTrait attributess
+ if (superType.getAllAttributes().containsKey(attrName)) {
+ superTypeTraitAttributes.put(attrName, attrEntry.getValue());
+ }
+ }
- superTypeTrait.setValues(attributes);
+ superTypeTrait.setValues(superTypeTraitAttributes);
}
}
@@ -160,7 +165,7 @@ public class NotificationEntityChangeListener implements EntityChangeListener {
}
}
- EntityNotificationImpl notification = new EntityNotificationImpl(entity, operationType, getAllTraits(entity, typeRegistry));
+ EntityNotification notification = new EntityNotification(entity, operationType, getAllTraits(entity, typeRegistry));
messages.add(notification);
}
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/webapp/src/main/java/org/apache/atlas/notification/NotificationHookConsumer.java
----------------------------------------------------------------------
diff --git a/webapp/src/main/java/org/apache/atlas/notification/NotificationHookConsumer.java b/webapp/src/main/java/org/apache/atlas/notification/NotificationHookConsumer.java
index 463a587..779298a 100644
--- a/webapp/src/main/java/org/apache/atlas/notification/NotificationHookConsumer.java
+++ b/webapp/src/main/java/org/apache/atlas/notification/NotificationHookConsumer.java
@@ -31,11 +31,11 @@ import org.apache.atlas.kafka.AtlasKafkaMessage;
import org.apache.atlas.listener.ActiveStateChangeHandler;
import org.apache.atlas.model.instance.AtlasEntity;
import org.apache.atlas.v1.model.instance.Referenceable;
-import org.apache.atlas.notification.hook.HookNotification.EntityCreateRequest;
-import org.apache.atlas.notification.hook.HookNotification.EntityDeleteRequest;
-import org.apache.atlas.notification.hook.HookNotification.EntityPartialUpdateRequest;
-import org.apache.atlas.notification.hook.HookNotification.EntityUpdateRequest;
-import org.apache.atlas.notification.hook.HookNotification.HookNotificationMessage;
+import org.apache.atlas.v1.model.notification.HookNotification.EntityCreateRequest;
+import org.apache.atlas.v1.model.notification.HookNotification.EntityDeleteRequest;
+import org.apache.atlas.v1.model.notification.HookNotification.EntityPartialUpdateRequest;
+import org.apache.atlas.v1.model.notification.HookNotification.EntityUpdateRequest;
+import org.apache.atlas.v1.model.notification.HookNotification.HookNotificationMessage;
import org.apache.atlas.repository.converters.AtlasInstanceConverter;
import org.apache.atlas.repository.store.graph.AtlasEntityStore;
import org.apache.atlas.repository.store.graph.v1.AtlasEntityStream;
@@ -362,7 +362,7 @@ public class NotificationHookConsumer implements Service, ActiveStateChangeHandl
audit(messageUser, api.getMethod(), api.getNormalizedPath());
}
- // TODO: entities = instanceConverter.toAtlasEntities(createRequest.getEntities());
+ entities = instanceConverter.toAtlasEntities(createRequest.getEntities());
atlasEntityStore.createOrUpdate(new AtlasEntityStream(entities), false);
break;
@@ -377,7 +377,7 @@ public class NotificationHookConsumer implements Service, ActiveStateChangeHandl
}
Referenceable referenceable = partialUpdateRequest.getEntity();
- // TODO: entities = instanceConverter.toAtlasEntity(referenceable);
+ entities = instanceConverter.toAtlasEntity(referenceable);
AtlasEntityType entityType = typeRegistry.getEntityTypeByName(partialUpdateRequest.getTypeName());
String guid = AtlasGraphUtilsV1.getGuidByUniqueAttributes(entityType, new HashMap<String, Object>() {
@@ -420,7 +420,7 @@ public class NotificationHookConsumer implements Service, ActiveStateChangeHandl
audit(messageUser, api.getMethod(), api.getNormalizedPath());
}
- // TODO: entities = instanceConverter.toAtlasEntities(updateRequest.getEntities());
+ entities = instanceConverter.toAtlasEntities(updateRequest.getEntities());
atlasEntityStore.createOrUpdate(new AtlasEntityStream(entities), false);
break;
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/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 9f25b15..517d25f 100644
--- a/webapp/src/test/java/org/apache/atlas/notification/EntityNotificationIT.java
+++ b/webapp/src/test/java/org/apache/atlas/notification/EntityNotificationIT.java
@@ -25,7 +25,7 @@ import org.apache.atlas.v1.model.instance.Id;
import org.apache.atlas.v1.model.instance.Referenceable;
import org.apache.atlas.v1.model.instance.Struct;
import org.apache.atlas.v1.model.typedef.TraitTypeDefinition;
-import org.apache.atlas.notification.entity.EntityNotification;
+import org.apache.atlas.v1.model.notification.EntityNotification;
import org.apache.atlas.type.AtlasType;
import org.apache.atlas.v1.typesystem.types.utils.TypesUtil;
import org.apache.atlas.web.integration.BaseResourceIT;
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/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 d77014a..1f045e4 100644
--- a/webapp/src/test/java/org/apache/atlas/notification/NotificationHookConsumerIT.java
+++ b/webapp/src/test/java/org/apache/atlas/notification/NotificationHookConsumerIT.java
@@ -22,12 +22,12 @@ import org.apache.atlas.EntityAuditEvent;
import org.apache.atlas.kafka.NotificationProvider;
import org.apache.atlas.v1.model.instance.Id;
import org.apache.atlas.v1.model.instance.Referenceable;
-import org.apache.atlas.notification.hook.HookNotification;
-import org.apache.atlas.notification.hook.HookNotification.HookNotificationMessage;
-import org.apache.atlas.notification.hook.HookNotification.EntityDeleteRequest;
-import org.apache.atlas.notification.hook.HookNotification.EntityPartialUpdateRequest;
-import org.apache.atlas.notification.hook.HookNotification.EntityCreateRequest;
-import org.apache.atlas.notification.hook.HookNotification.EntityUpdateRequest;
+import org.apache.atlas.v1.model.notification.HookNotification;
+import org.apache.atlas.v1.model.notification.HookNotification.HookNotificationMessage;
+import org.apache.atlas.v1.model.notification.HookNotification.EntityDeleteRequest;
+import org.apache.atlas.v1.model.notification.HookNotification.EntityPartialUpdateRequest;
+import org.apache.atlas.v1.model.notification.HookNotification.EntityCreateRequest;
+import org.apache.atlas.v1.model.notification.HookNotification.EntityUpdateRequest;
import org.apache.atlas.web.integration.BaseResourceIT;
import org.codehaus.jettison.json.JSONArray;
import org.testng.annotations.AfterClass;
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/webapp/src/test/java/org/apache/atlas/notification/NotificationHookConsumerKafkaTest.java
----------------------------------------------------------------------
diff --git a/webapp/src/test/java/org/apache/atlas/notification/NotificationHookConsumerKafkaTest.java b/webapp/src/test/java/org/apache/atlas/notification/NotificationHookConsumerKafkaTest.java
index 25161f5..68497e0 100644
--- a/webapp/src/test/java/org/apache/atlas/notification/NotificationHookConsumerKafkaTest.java
+++ b/webapp/src/test/java/org/apache/atlas/notification/NotificationHookConsumerKafkaTest.java
@@ -27,7 +27,7 @@ import org.apache.atlas.kafka.KafkaNotification;
import org.apache.atlas.kafka.NotificationProvider;
import org.apache.atlas.model.instance.AtlasEntity;
import org.apache.atlas.v1.model.instance.Referenceable;
-import org.apache.atlas.notification.hook.HookNotification;
+import org.apache.atlas.v1.model.notification.HookNotification;
import org.apache.atlas.repository.converters.AtlasInstanceConverter;
import org.apache.atlas.repository.store.graph.AtlasEntityStore;
import org.apache.atlas.repository.store.graph.v1.EntityStream;
@@ -41,7 +41,7 @@ import org.testng.Assert;
import org.testng.annotations.AfterTest;
import org.testng.annotations.BeforeTest;
import org.testng.annotations.Test;
-import static org.apache.atlas.notification.hook.HookNotification.HookNotificationMessage;
+import static org.apache.atlas.v1.model.notification.HookNotification.HookNotificationMessage;
import java.util.List;
import org.apache.atlas.kafka.AtlasKafkaConsumer;
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/webapp/src/test/java/org/apache/atlas/notification/NotificationHookConsumerTest.java
----------------------------------------------------------------------
diff --git a/webapp/src/test/java/org/apache/atlas/notification/NotificationHookConsumerTest.java b/webapp/src/test/java/org/apache/atlas/notification/NotificationHookConsumerTest.java
index 6b13915..2d3d5ba 100644
--- a/webapp/src/test/java/org/apache/atlas/notification/NotificationHookConsumerTest.java
+++ b/webapp/src/test/java/org/apache/atlas/notification/NotificationHookConsumerTest.java
@@ -25,7 +25,7 @@ import org.apache.atlas.kafka.AtlasKafkaMessage;
import org.apache.atlas.model.instance.AtlasEntity;
import org.apache.atlas.model.instance.EntityMutationResponse;
import org.apache.atlas.v1.model.instance.Referenceable;
-import org.apache.atlas.notification.hook.HookNotification;
+import org.apache.atlas.v1.model.notification.HookNotification;
import org.apache.atlas.repository.converters.AtlasInstanceConverter;
import org.apache.atlas.repository.store.graph.AtlasEntityStore;
import org.apache.atlas.repository.store.graph.v1.EntityStream;
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/webapp/src/test/java/org/apache/atlas/web/integration/BaseResourceIT.java
----------------------------------------------------------------------
diff --git a/webapp/src/test/java/org/apache/atlas/web/integration/BaseResourceIT.java b/webapp/src/test/java/org/apache/atlas/web/integration/BaseResourceIT.java
index 4360a69..e63099d 100755
--- a/webapp/src/test/java/org/apache/atlas/web/integration/BaseResourceIT.java
+++ b/webapp/src/test/java/org/apache/atlas/web/integration/BaseResourceIT.java
@@ -43,7 +43,7 @@ import org.apache.atlas.v1.model.typedef.*;
import org.apache.atlas.v1.model.typedef.EnumTypeDefinition.EnumValue;
import org.apache.atlas.notification.NotificationConsumer;
import org.apache.atlas.kafka.*;
-import org.apache.atlas.notification.entity.EntityNotification;
+import org.apache.atlas.v1.model.notification.EntityNotification;
import org.apache.atlas.type.AtlasType;
import org.apache.atlas.type.AtlasTypeUtil;
import org.apache.atlas.v1.typesystem.types.utils.TypesUtil;
[2/3] atlas git commit: ATLAS-2251: notification module updates
Posted by ma...@apache.org.
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/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 988d98a..8bc7cb4 100644
--- a/notification/src/main/java/org/apache/atlas/notification/AbstractNotification.java
+++ b/notification/src/main/java/org/apache/atlas/notification/AbstractNotification.java
@@ -18,17 +18,19 @@
package org.apache.atlas.notification;
import com.google.common.annotations.VisibleForTesting;
-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.ha.HAConfiguration;
+import org.apache.atlas.model.notification.AtlasNotificationBaseMessage;
+import org.apache.atlas.model.notification.AtlasNotificationMessage;
+import org.apache.atlas.model.notification.AtlasNotificationStringMessage;
import org.apache.atlas.v1.model.instance.Referenceable;
-import org.apache.atlas.notification.AtlasNotificationBaseMessage.CompressionKind;
+import org.apache.atlas.model.notification.AtlasNotificationBaseMessage.CompressionKind;
import org.apache.atlas.type.AtlasType;
+import org.apache.atlas.model.notification.MessageVersion;
import org.apache.commons.configuration.Configuration;
import org.apache.commons.lang.StringUtils;
import org.codehaus.jettison.json.JSONArray;
@@ -44,8 +46,8 @@ import java.util.List;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicInteger;
-import static org.apache.atlas.notification.AtlasNotificationBaseMessage.MESSAGE_COMPRESSION_ENABLED;
-import static org.apache.atlas.notification.AtlasNotificationBaseMessage.MESSAGE_MAX_LENGTH_BYTES;
+import static org.apache.atlas.model.notification.AtlasNotificationBaseMessage.MESSAGE_COMPRESSION_ENABLED;
+import static org.apache.atlas.model.notification.AtlasNotificationBaseMessage.MESSAGE_MAX_LENGTH_BYTES;
/**
* Abstract notification interface implementation.
@@ -78,14 +80,6 @@ public abstract class AbstractNotification implements NotificationInterface {
private final boolean embedded;
private final boolean isHAEnabled;
- /**
- * Used for message serialization.
- */
- public static final Gson GSON = new GsonBuilder().
- registerTypeAdapter(Referenceable.class, new ReferenceableSerializer()).
- registerTypeAdapter(JSONArray.class, new JSONArraySerializer()).
- create();
-
// ----- Constructors ----------------------------------------------------
public AbstractNotification(Configuration applicationProperties) throws AtlasException {
@@ -158,7 +152,7 @@ public abstract class AbstractNotification implements NotificationInterface {
public static String getMessageJson(Object message) {
AtlasNotificationMessage<?> notificationMsg = new AtlasNotificationMessage<>(CURRENT_MESSAGE_VERSION, message);
- return GSON.toJson(notificationMsg);
+ return AtlasType.toV1Json(notificationMsg);
}
private static String getHostAddress() {
@@ -188,7 +182,7 @@ public abstract class AbstractNotification implements NotificationInterface {
*/
public static void createNotificationMessages(Object message, List<String> msgJsonList) {
AtlasNotificationMessage<?> notificationMsg = new AtlasNotificationMessage<>(CURRENT_MESSAGE_VERSION, message, getHostAddress(), getCurrentUser());
- String msgJson = GSON.toJson(notificationMsg);
+ String msgJson = AtlasType.toV1Json(notificationMsg);
boolean msgLengthExceedsLimit = (msgJson.length() * MAX_BYTES_PER_CHAR) > MESSAGE_MAX_LENGTH_BYTES;
@@ -213,7 +207,7 @@ public abstract class AbstractNotification implements NotificationInterface {
if (!msgLengthExceedsLimit) { // no need to split
AtlasNotificationStringMessage compressedMsg = new AtlasNotificationStringMessage(encodedBytes, msgId, compressionKind);
- msgJson = GSON.toJson(compressedMsg); // msgJson will not have multi-byte characters here, due to use of encodeBase64() above
+ msgJson = AtlasType.toV1Json(compressedMsg); // msgJson will not have multi-byte characters here, due to use of encodeBase64() above
msgBytes = null; // not used after this point
} else { // encodedBytes will be split
msgJson = null; // not used after this point
@@ -239,7 +233,7 @@ public abstract class AbstractNotification implements NotificationInterface {
AtlasNotificationStringMessage splitMsg = new AtlasNotificationStringMessage(encodedBytes, offset, length, msgId, compressionKind, i, splitCount);
- String splitMsgJson = GSON.toJson(splitMsg);
+ String splitMsgJson = AtlasType.toV1Json(splitMsg);
msgJsonList.add(splitMsgJson);
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/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 8cf1e8e..c3940ce 100644
--- a/notification/src/main/java/org/apache/atlas/notification/AbstractNotificationConsumer.java
+++ b/notification/src/main/java/org/apache/atlas/notification/AbstractNotificationConsumer.java
@@ -16,30 +16,19 @@
* limitations under the License.
*/
package org.apache.atlas.notification;
+
import org.apache.kafka.common.TopicPartition;
+
/**
* Abstract notification consumer.
*/
public abstract class AbstractNotificationConsumer<T> implements NotificationConsumer<T> {
+ protected final AtlasNotificationMessageDeserializer<T> deserializer;
- /**
- * Deserializer used to deserialize notification messages for this consumer.
- */
- protected final MessageDeserializer<T> deserializer;
-
-
-
- /**
- * Construct an AbstractNotificationConsumer.
- *
- * @param deserializer the message deserializer used by this consumer
- */
- public AbstractNotificationConsumer(MessageDeserializer<T> deserializer) {
+ protected AbstractNotificationConsumer(AtlasNotificationMessageDeserializer<T> deserializer) {
this.deserializer = deserializer;
}
-
-
public abstract void commit(TopicPartition partition, long offset);
}
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/notification/src/main/java/org/apache/atlas/notification/AtlasNotificationBaseMessage.java
----------------------------------------------------------------------
diff --git a/notification/src/main/java/org/apache/atlas/notification/AtlasNotificationBaseMessage.java b/notification/src/main/java/org/apache/atlas/notification/AtlasNotificationBaseMessage.java
deleted file mode 100644
index 3b377de..0000000
--- a/notification/src/main/java/org/apache/atlas/notification/AtlasNotificationBaseMessage.java
+++ /dev/null
@@ -1,194 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.atlas.notification;
-
-
-import org.apache.atlas.AtlasConfiguration;
-import org.apache.commons.codec.binary.Base64;
-import org.apache.commons.codec.binary.StringUtils;
-import org.apache.commons.compress.utils.IOUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.ByteArrayInputStream;
-import java.io.ByteArrayOutputStream;
-import java.io.IOException;
-import java.util.zip.GZIPInputStream;
-import java.util.zip.GZIPOutputStream;
-
-
-public class AtlasNotificationBaseMessage {
- private static final Logger LOG = LoggerFactory.getLogger(AtlasNotificationBaseMessage.class);
-
- public static final int MESSAGE_MAX_LENGTH_BYTES = AtlasConfiguration.NOTIFICATION_MESSAGE_MAX_LENGTH_BYTES.getInt() - 512; // 512 bytes for envelop;
- public static final boolean MESSAGE_COMPRESSION_ENABLED = AtlasConfiguration.NOTIFICATION_MESSAGE_COMPRESSION_ENABLED.getBoolean();
-
- public enum CompressionKind { NONE, GZIP };
-
- private MessageVersion version = null;
- private String msgId = null;
- private CompressionKind msgCompressionKind = CompressionKind.NONE;
- private int msgSplitIdx = 1;
- private int msgSplitCount = 1;
-
-
- public AtlasNotificationBaseMessage() {
- }
-
- public AtlasNotificationBaseMessage(MessageVersion version) {
- this(version, null, CompressionKind.NONE);
- }
-
- public AtlasNotificationBaseMessage(MessageVersion version, String msgId, CompressionKind msgCompressionKind) {
- this.version = version;
- this.msgId = msgId;
- this.msgCompressionKind = msgCompressionKind;
- }
-
- public AtlasNotificationBaseMessage(MessageVersion version, String msgId, CompressionKind msgCompressionKind, int msgSplitIdx, int msgSplitCount) {
- this.version = version;
- this.msgId = msgId;
- this.msgCompressionKind = msgCompressionKind;
- this.msgSplitIdx = msgSplitIdx;
- this.msgSplitCount = msgSplitCount;
- }
-
- public void setVersion(MessageVersion version) {
- this.version = version;
- }
-
- public MessageVersion getVersion() {
- return version;
- }
-
- public String getMsgId() {
- return msgId;
- }
-
- public void setMsgId(String msgId) {
- this.msgId = msgId;
- }
-
- public CompressionKind getMsgCompressionKind() {
- return msgCompressionKind;
- }
-
- public void setMsgCompressed(CompressionKind msgCompressionKind) {
- this.msgCompressionKind = msgCompressionKind;
- }
-
- public int getMsgSplitIdx() {
- return msgSplitIdx;
- }
-
- public void setMsgSplitIdx(int msgSplitIdx) {
- this.msgSplitIdx = msgSplitIdx;
- }
-
- public int getMsgSplitCount() {
- return msgSplitCount;
- }
-
- public void setMsgSplitCount(int msgSplitCount) {
- this.msgSplitCount = msgSplitCount;
- }
-
- /**
- * Compare the version of this message with the given version.
- *
- * @param compareToVersion the version to compare to
- *
- * @return a negative integer, zero, or a positive integer as this message's version is less than, equal to,
- * or greater than the given version.
- */
- public int compareVersion(MessageVersion compareToVersion) {
- return version.compareTo(compareToVersion);
- }
-
-
- public static byte[] getBytesUtf8(String str) {
- return StringUtils.getBytesUtf8(str);
- }
-
- public static String getStringUtf8(byte[] bytes) {
- return StringUtils.newStringUtf8(bytes);
- }
-
- public static byte[] encodeBase64(byte[] bytes) {
- return Base64.encodeBase64(bytes);
- }
-
- public static byte[] decodeBase64(byte[] bytes) {
- return Base64.decodeBase64(bytes);
- }
-
- public static byte[] gzipCompressAndEncodeBase64(byte[] bytes) {
- return encodeBase64(gzipCompress(bytes));
- }
-
- public static byte[] decodeBase64AndGzipUncompress(byte[] bytes) {
- return gzipUncompress(decodeBase64(bytes));
- }
-
- public static String gzipCompress(String str) {
- byte[] bytes = getBytesUtf8(str);
- byte[] compressedBytes = gzipCompress(bytes);
- byte[] encodedBytes = encodeBase64(compressedBytes);
-
- return getStringUtf8(encodedBytes);
- }
-
- public static String gzipUncompress(String str) {
- byte[] encodedBytes = getBytesUtf8(str);
- byte[] compressedBytes = decodeBase64(encodedBytes);
- byte[] bytes = gzipUncompress(compressedBytes);
-
- return getStringUtf8(bytes);
- }
-
- public static byte[] gzipCompress(byte[] content) {
- ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
-
- try {
- GZIPOutputStream gzipOutputStream = new GZIPOutputStream(byteArrayOutputStream);
-
- gzipOutputStream.write(content);
- gzipOutputStream.close();
- } catch (IOException e) {
- LOG.error("gzipCompress(): error compressing {} bytes", content.length, e);
-
- throw new RuntimeException(e);
- }
-
- return byteArrayOutputStream.toByteArray();
- }
-
- public static byte[] gzipUncompress(byte[] content) {
- ByteArrayOutputStream out = new ByteArrayOutputStream();
-
- try {
- IOUtils.copy(new GZIPInputStream(new ByteArrayInputStream(content)), out);
- } catch (IOException e) {
- LOG.error("gzipUncompress(): error uncompressing {} bytes", content.length, e);
- }
-
- return out.toByteArray();
- }
-}
-
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/notification/src/main/java/org/apache/atlas/notification/AtlasNotificationMessage.java
----------------------------------------------------------------------
diff --git a/notification/src/main/java/org/apache/atlas/notification/AtlasNotificationMessage.java b/notification/src/main/java/org/apache/atlas/notification/AtlasNotificationMessage.java
deleted file mode 100644
index 63d93c9..0000000
--- a/notification/src/main/java/org/apache/atlas/notification/AtlasNotificationMessage.java
+++ /dev/null
@@ -1,87 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.atlas.notification;
-
-import org.joda.time.DateTimeZone;
-import org.joda.time.Instant;
-
-/**
- * Represents a notification message that is associated with a version.
- */
-public class AtlasNotificationMessage<T> extends AtlasNotificationBaseMessage {
- private String msgSourceIP;
- private String msgCreatedBy;
- private long msgCreationTime;
-
- /**
- * The actual message.
- */
- private final T message;
-
-
- // ----- Constructors ----------------------------------------------------
-
- /**
- * Create a notification message.
- *
- * @param version the message version
- * @param message the actual message
- */
- public AtlasNotificationMessage(MessageVersion version, T message) {
- this(version, message, null, null);
- }
-
- public AtlasNotificationMessage(MessageVersion version, T message, String msgSourceIP, String createdBy) {
- super(version);
-
- this.msgSourceIP = msgSourceIP;
- this.msgCreatedBy = createdBy;
- this.msgCreationTime = Instant.now().toDateTime(DateTimeZone.UTC).getMillis();
- this.message = message;
- }
-
-
- public String getMsgSourceIP() {
- return msgSourceIP;
- }
-
- public void setMsgSourceIP(String msgSourceIP) {
- this.msgSourceIP = msgSourceIP;
- }
-
- public String getMsgCreatedBy() {
- return msgCreatedBy;
- }
-
- public void setMsgCreatedBy(String msgCreatedBy) {
- this.msgCreatedBy = msgCreatedBy;
- }
-
- public long getMsgCreationTime() {
- return msgCreationTime;
- }
-
- public void setMsgCreationTime(long msgCreationTime) {
- this.msgCreationTime = msgCreationTime;
- }
-
- public T getMessage() {
- return message;
- }
-}
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/notification/src/main/java/org/apache/atlas/notification/AtlasNotificationMessageDeserializer.java
----------------------------------------------------------------------
diff --git a/notification/src/main/java/org/apache/atlas/notification/AtlasNotificationMessageDeserializer.java b/notification/src/main/java/org/apache/atlas/notification/AtlasNotificationMessageDeserializer.java
index 2a175ba..d6e6878 100644
--- a/notification/src/main/java/org/apache/atlas/notification/AtlasNotificationMessageDeserializer.java
+++ b/notification/src/main/java/org/apache/atlas/notification/AtlasNotificationMessageDeserializer.java
@@ -19,14 +19,17 @@
package org.apache.atlas.notification;
import com.google.common.annotations.VisibleForTesting;
-import com.google.gson.Gson;
-import org.apache.atlas.notification.AtlasNotificationBaseMessage.CompressionKind;
+import org.apache.atlas.model.notification.AtlasNotificationBaseMessage;
+import org.apache.atlas.model.notification.AtlasNotificationBaseMessage.CompressionKind;
+import org.apache.atlas.model.notification.AtlasNotificationMessage;
+import org.apache.atlas.model.notification.AtlasNotificationStringMessage;
+import org.apache.atlas.type.AtlasType;
+import org.apache.atlas.model.notification.MessageVersion;
import org.apache.commons.lang3.StringUtils;
+import org.codehaus.jackson.type.TypeReference;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.lang.reflect.ParameterizedType;
-import java.lang.reflect.Type;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
@@ -47,11 +50,10 @@ public abstract class AtlasNotificationMessageDeserializer<T> implements Message
public static final String VERSION_MISMATCH_MSG =
"Notification message version mismatch. Expected %s but recieved %s. Message %s";
- private final Type notificationMessageType;
- private final Type messageType;
- private final MessageVersion expectedVersion;
- private final Logger notificationLogger;
- private final Gson gson;
+ private final TypeReference<T> messageType;
+ private final TypeReference<AtlasNotificationMessage<T>> notificationMessageType;
+ private final MessageVersion expectedVersion;
+ private final Logger notificationLogger;
private final Map<String, SplitMessageAggregator> splitMsgBuffer = new HashMap<>();
@@ -65,33 +67,40 @@ public abstract class AtlasNotificationMessageDeserializer<T> implements Message
/**
* Create a notification message deserializer.
*
- * @param notificationMessageType the type of the notification message
* @param expectedVersion the expected message version
- * @param gson JSON serialization/deserialization
* @param notificationLogger logger for message version mismatch
*/
- public AtlasNotificationMessageDeserializer(Type notificationMessageType, MessageVersion expectedVersion,
- Gson gson, Logger notificationLogger) {
- this(notificationMessageType, expectedVersion, gson, notificationLogger,
+ public AtlasNotificationMessageDeserializer(TypeReference<T> messageType,
+ TypeReference<AtlasNotificationMessage<T>> notificationMessageType,
+ MessageVersion expectedVersion, Logger notificationLogger) {
+ this(messageType, notificationMessageType, expectedVersion, notificationLogger,
NOTIFICATION_SPLIT_MESSAGE_SEGMENTS_WAIT_TIME_SECONDS.getLong() * 1000,
NOTIFICATION_SPLIT_MESSAGE_BUFFER_PURGE_INTERVAL_SECONDS.getLong() * 1000);
}
- public AtlasNotificationMessageDeserializer(Type notificationMessageType, MessageVersion expectedVersion,
- Gson gson, Logger notificationLogger,
+ public AtlasNotificationMessageDeserializer(TypeReference<T> messageType,
+ TypeReference<AtlasNotificationMessage<T>> notificationMessageType,
+ MessageVersion expectedVersion,
+ Logger notificationLogger,
long splitMessageSegmentsWaitTimeMs,
long splitMessageBufferPurgeIntervalMs) {
+ this.messageType = messageType;
this.notificationMessageType = notificationMessageType;
- this.messageType = ((ParameterizedType) notificationMessageType).getActualTypeArguments()[0];
this.expectedVersion = expectedVersion;
- this.gson = gson;
this.notificationLogger = notificationLogger;
this.splitMessageSegmentsWaitTimeMs = splitMessageSegmentsWaitTimeMs;
this.splitMessageBufferPurgeIntervalMs = splitMessageBufferPurgeIntervalMs;
}
- // ----- MessageDeserializer ---------------------------------------------
+ public TypeReference<T> getMessageType() {
+ return messageType;
+ }
+ public TypeReference<AtlasNotificationMessage<T>> getNotificationMessageType() {
+ return notificationMessageType;
+ }
+
+ // ----- MessageDeserializer ---------------------------------------------
@Override
public T deserialize(String messageJson) {
final T ret;
@@ -99,15 +108,15 @@ public abstract class AtlasNotificationMessageDeserializer<T> implements Message
messageCountTotal.incrementAndGet();
messageCountSinceLastInterval.incrementAndGet();
- AtlasNotificationBaseMessage msg = gson.fromJson(messageJson, AtlasNotificationBaseMessage.class);
+ AtlasNotificationBaseMessage msg = AtlasType.fromV1Json(messageJson, AtlasNotificationBaseMessage.class);
if (msg.getVersion() == null) { // older style messages not wrapped with AtlasNotificationMessage
- ret = gson.fromJson(messageJson, messageType);
+ ret = AtlasType.fromV1Json(messageJson, messageType);
} else {
String msgJson = messageJson;
if (msg.getMsgSplitCount() > 1) { // multi-part message
- AtlasNotificationStringMessage splitMsg = gson.fromJson(msgJson, AtlasNotificationStringMessage.class);
+ AtlasNotificationStringMessage splitMsg = AtlasType.fromV1Json(msgJson, AtlasNotificationStringMessage.class);
checkVersion(splitMsg, msgJson);
@@ -184,7 +193,7 @@ public abstract class AtlasNotificationMessageDeserializer<T> implements Message
LOG.info("Received msgID={}: splitCount={}, length={} bytes", msgId, splitCount, bytes.length);
}
- msg = gson.fromJson(msgJson, AtlasNotificationBaseMessage.class);
+ msg = AtlasType.fromV1Json(msgJson, AtlasNotificationBaseMessage.class);
} else {
msg = null;
}
@@ -197,7 +206,7 @@ public abstract class AtlasNotificationMessageDeserializer<T> implements Message
if (msg != null) {
if (CompressionKind.GZIP.equals(msg.getMsgCompressionKind())) {
- AtlasNotificationStringMessage compressedMsg = gson.fromJson(msgJson, AtlasNotificationStringMessage.class);
+ AtlasNotificationStringMessage compressedMsg = AtlasType.fromV1Json(msgJson, AtlasNotificationStringMessage.class);
byte[] encodedBytes = AtlasNotificationBaseMessage.getBytesUtf8(compressedMsg.getMessage());
byte[] bytes = AtlasNotificationBaseMessage.decodeBase64AndGzipUncompress(encodedBytes);
@@ -207,7 +216,7 @@ public abstract class AtlasNotificationMessageDeserializer<T> implements Message
LOG.info("Received msgID={}: compressed={} bytes, uncompressed={} bytes", compressedMsg.getMsgId(), encodedBytes.length, bytes.length);
}
- AtlasNotificationMessage<T> atlasNotificationMessage = gson.fromJson(msgJson, notificationMessageType);
+ AtlasNotificationMessage<T> atlasNotificationMessage = AtlasType.fromV1Json(msgJson, notificationMessageType);
checkVersion(atlasNotificationMessage, msgJson);
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/notification/src/main/java/org/apache/atlas/notification/AtlasNotificationStringMessage.java
----------------------------------------------------------------------
diff --git a/notification/src/main/java/org/apache/atlas/notification/AtlasNotificationStringMessage.java b/notification/src/main/java/org/apache/atlas/notification/AtlasNotificationStringMessage.java
deleted file mode 100644
index 41485a0..0000000
--- a/notification/src/main/java/org/apache/atlas/notification/AtlasNotificationStringMessage.java
+++ /dev/null
@@ -1,66 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.atlas.notification;
-
-
-public class AtlasNotificationStringMessage extends AtlasNotificationBaseMessage {
- private String message = null;
-
- public AtlasNotificationStringMessage() {
- super(AbstractNotification.CURRENT_MESSAGE_VERSION);
- }
-
- public AtlasNotificationStringMessage(String message) {
- super(AbstractNotification.CURRENT_MESSAGE_VERSION);
-
- this.message = message;
- }
-
- public AtlasNotificationStringMessage(String message, String msgId, CompressionKind compressionKind) {
- super(AbstractNotification.CURRENT_MESSAGE_VERSION, msgId, compressionKind);
-
- this.message = message;
- }
-
- public AtlasNotificationStringMessage(String message, String msgId, CompressionKind compressionKind, int msgSplitIdx, int msgSplitCount) {
- super(AbstractNotification.CURRENT_MESSAGE_VERSION, msgId, compressionKind, msgSplitIdx, msgSplitCount);
-
- this.message = message;
- }
-
- public AtlasNotificationStringMessage(byte[] encodedBytes, String msgId, CompressionKind compressionKind) {
- super(AbstractNotification.CURRENT_MESSAGE_VERSION, msgId, compressionKind);
-
- this.message = AtlasNotificationBaseMessage.getStringUtf8(encodedBytes);
- }
-
- public AtlasNotificationStringMessage(byte[] encodedBytes, int offset, int length, String msgId, CompressionKind compressionKind, int msgSplitIdx, int msgSplitCount) {
- super(AbstractNotification.CURRENT_MESSAGE_VERSION, msgId, compressionKind, msgSplitIdx, msgSplitCount);
-
- this.message = new String(encodedBytes, offset, length);
- }
-
- public void setMessage(String message) {
- this.message = message;
- }
-
- public String getMessage() {
- return message;
- }
-}
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/notification/src/main/java/org/apache/atlas/notification/MessageVersion.java
----------------------------------------------------------------------
diff --git a/notification/src/main/java/org/apache/atlas/notification/MessageVersion.java b/notification/src/main/java/org/apache/atlas/notification/MessageVersion.java
deleted file mode 100644
index 7f96638..0000000
--- a/notification/src/main/java/org/apache/atlas/notification/MessageVersion.java
+++ /dev/null
@@ -1,141 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.atlas.notification;
-
-import java.util.ArrayList;
-import java.util.Arrays;
-
-/**
- * Represents the version of a notification message.
- */
-public class MessageVersion implements Comparable<MessageVersion> {
- /**
- * Used for message with no version (old format).
- */
- public static final MessageVersion NO_VERSION = new MessageVersion("0");
- public static final MessageVersion VERSION_1 = new MessageVersion("1.0.0");
-
- public static final MessageVersion CURRENT_VERSION = VERSION_1;
-
- private final String version;
-
-
- // ----- Constructors ----------------------------------------------------
-
- /**
- * Create a message version.
- *
- * @param version the version string
- */
- public MessageVersion(String version) {
- this.version = version;
-
- try {
- getVersionParts();
- } catch (NumberFormatException e) {
- throw new IllegalArgumentException(String.format("Invalid version string : %s.", version), e);
- }
- }
-
-
- // ----- Comparable ------------------------------------------------------
-
- @Override
- public int compareTo(MessageVersion that) {
- if (that == null) {
- return 1;
- }
-
- Integer[] thisParts = getVersionParts();
- Integer[] thatParts = that.getVersionParts();
-
- int length = Math.max(thisParts.length, thatParts.length);
-
- for (int i = 0; i < length; i++) {
-
- int comp = getVersionPart(thisParts, i) - getVersionPart(thatParts, i);
-
- if (comp != 0) {
- return comp;
- }
- }
- return 0;
- }
-
-
- // ----- Object overrides ------------------------------------------------
-
- @Override
- public boolean equals(Object that) {
- if (this == that){
- return true;
- }
-
- if (that == null || getClass() != that.getClass()) {
- return false;
- }
-
- return compareTo((MessageVersion) that) == 0;
- }
-
- @Override
- public int hashCode() {
- return Arrays.hashCode(getVersionParts());
- }
-
-
- @Override
- public String toString() {
- return "MessageVersion[version=" + version + "]";
- }
-
- // ----- helper methods --------------------------------------------------
-
- /**
- * Get the version parts array by splitting the version string.
- * Strip the trailing zeros (i.e. '1.0.0' equals '1').
- *
- * @return the version parts array
- */
- protected Integer[] getVersionParts() {
-
- String[] sParts = version.split("\\.");
- ArrayList<Integer> iParts = new ArrayList<>();
- int trailingZeros = 0;
-
- for (String sPart : sParts) {
- Integer iPart = new Integer(sPart);
-
- if (iPart == 0) {
- ++trailingZeros;
- } else {
- for (int i = 0; i < trailingZeros; ++i) {
- iParts.add(0);
- }
- trailingZeros = 0;
- iParts.add(iPart);
- }
- }
- return iParts.toArray(new Integer[iParts.size()]);
- }
-
- private Integer getVersionPart(Integer[] versionParts, int i) {
- return i < versionParts.length ? versionParts[i] : 0;
- }
-}
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/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 8809225..975967d 100644
--- a/notification/src/main/java/org/apache/atlas/notification/NotificationInterface.java
+++ b/notification/src/main/java/org/apache/atlas/notification/NotificationInterface.java
@@ -18,10 +18,13 @@
package org.apache.atlas.notification;
import com.google.gson.reflect.TypeToken;
+import org.apache.atlas.model.notification.AtlasNotificationMessage;
import org.apache.atlas.notification.entity.EntityMessageDeserializer;
-import org.apache.atlas.notification.entity.EntityNotification;
import org.apache.atlas.notification.hook.HookMessageDeserializer;
-import org.apache.atlas.notification.hook.HookNotification;
+import org.apache.atlas.v1.model.notification.EntityNotification;
+import org.apache.atlas.v1.model.notification.HookNotification;
+import org.codehaus.jackson.type.TypeReference;
+import scala.reflect.internal.Types;
import java.lang.reflect.Type;
import java.util.List;
@@ -53,47 +56,25 @@ public interface NotificationInterface {
/**
* Versioned notification message class types.
*/
- Type HOOK_VERSIONED_MESSAGE_TYPE =
- new TypeToken<AtlasNotificationMessage<HookNotification.HookNotificationMessage>>(){}.getType();
-
Type ENTITY_VERSIONED_MESSAGE_TYPE = new TypeToken<AtlasNotificationMessage<EntityNotification>>(){}.getType();
/**
* Atlas notification types.
*/
enum NotificationType {
-
// Notifications from the Atlas integration hooks.
- HOOK(HOOK_NOTIFICATION_CLASS, new HookMessageDeserializer()),
+ HOOK(new HookMessageDeserializer()),
// Notifications to entity change consumers.
- ENTITIES(ENTITY_NOTIFICATION_CLASS, new EntityMessageDeserializer());
-
-
- /**
- * The notification class associated with this type.
- */
- private final Class classType;
-
- /**
- * The message deserializer for this type.
- */
- private final MessageDeserializer deserializer;
+ ENTITIES(new EntityMessageDeserializer());
+ private final AtlasNotificationMessageDeserializer deserializer;
- NotificationType(Class classType, MessageDeserializer<?> deserializer) {
- this.classType = classType;
+ NotificationType(AtlasNotificationMessageDeserializer deserializer) {
this.deserializer = deserializer;
}
-
- // ----- accessors ---------------------------------------------------
-
- public Class getClassType() {
- return classType;
- }
-
- public MessageDeserializer getDeserializer() {
+ public AtlasNotificationMessageDeserializer getDeserializer() {
return deserializer;
}
}
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/notification/src/main/java/org/apache/atlas/notification/SplitMessageAggregator.java
----------------------------------------------------------------------
diff --git a/notification/src/main/java/org/apache/atlas/notification/SplitMessageAggregator.java b/notification/src/main/java/org/apache/atlas/notification/SplitMessageAggregator.java
index 148b57f..10df121 100644
--- a/notification/src/main/java/org/apache/atlas/notification/SplitMessageAggregator.java
+++ b/notification/src/main/java/org/apache/atlas/notification/SplitMessageAggregator.java
@@ -18,6 +18,8 @@
package org.apache.atlas.notification;
+import org.apache.atlas.model.notification.AtlasNotificationStringMessage;
+
public class SplitMessageAggregator {
private final String msgId;
private final AtlasNotificationStringMessage[] splitMessagesBuffer;
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/notification/src/main/java/org/apache/atlas/notification/entity/EntityMessageDeserializer.java
----------------------------------------------------------------------
diff --git a/notification/src/main/java/org/apache/atlas/notification/entity/EntityMessageDeserializer.java b/notification/src/main/java/org/apache/atlas/notification/entity/EntityMessageDeserializer.java
index a6f7e64..526aa93 100644
--- a/notification/src/main/java/org/apache/atlas/notification/entity/EntityMessageDeserializer.java
+++ b/notification/src/main/java/org/apache/atlas/notification/entity/EntityMessageDeserializer.java
@@ -18,19 +18,14 @@
package org.apache.atlas.notification.entity;
-import com.google.gson.JsonDeserializationContext;
-import com.google.gson.JsonDeserializer;
-import com.google.gson.JsonElement;
+import org.apache.atlas.model.notification.AtlasNotificationMessage;
import org.apache.atlas.notification.AbstractMessageDeserializer;
import org.apache.atlas.notification.AbstractNotification;
-import org.apache.atlas.notification.NotificationInterface;
+import org.apache.atlas.v1.model.notification.EntityNotification;
+import org.codehaus.jackson.type.TypeReference;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.lang.reflect.Type;
-import java.util.Collections;
-import java.util.Map;
-
/**
* Entity notification message deserializer.
*/
@@ -48,29 +43,8 @@ public class EntityMessageDeserializer extends AbstractMessageDeserializer<Entit
* Create an entity notification message deserializer.
*/
public EntityMessageDeserializer() {
- super(NotificationInterface.ENTITY_VERSIONED_MESSAGE_TYPE,
- AbstractNotification.CURRENT_MESSAGE_VERSION, getDeserializerMap(), NOTIFICATION_LOGGER);
- }
-
-
- // ----- helper methods --------------------------------------------------
-
- private static Map<Type, JsonDeserializer> getDeserializerMap() {
- return Collections.<Type, JsonDeserializer>singletonMap(
- NotificationInterface.ENTITY_NOTIFICATION_CLASS, new EntityNotificationDeserializer());
- }
-
-
- // ----- deserializer classes --------------------------------------------
-
- /**
- * Deserializer for EntityNotification.
- */
- protected static final class EntityNotificationDeserializer implements JsonDeserializer<EntityNotification> {
- @Override
- public EntityNotification deserialize(final JsonElement json, final Type type,
- final JsonDeserializationContext context) {
- return context.deserialize(json, EntityNotificationImpl.class);
- }
+ super(new TypeReference<EntityNotification>() {},
+ new TypeReference<AtlasNotificationMessage<EntityNotification>>() {},
+ AbstractNotification.CURRENT_MESSAGE_VERSION, NOTIFICATION_LOGGER);
}
}
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/notification/src/main/java/org/apache/atlas/notification/entity/EntityNotification.java
----------------------------------------------------------------------
diff --git a/notification/src/main/java/org/apache/atlas/notification/entity/EntityNotification.java b/notification/src/main/java/org/apache/atlas/notification/entity/EntityNotification.java
deleted file mode 100644
index 96e2e2f..0000000
--- a/notification/src/main/java/org/apache/atlas/notification/entity/EntityNotification.java
+++ /dev/null
@@ -1,66 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.atlas.notification.entity;
-
-
-import org.apache.atlas.v1.model.instance.Referenceable;
-import org.apache.atlas.v1.model.instance.Struct;
-
-import java.util.List;
-
-/**
- * Notification of entity changes.
- */
-public interface EntityNotification {
-
- /**
- * Operations that result in an entity notification.
- */
- enum OperationType {
- ENTITY_CREATE,
- ENTITY_UPDATE,
- ENTITY_DELETE,
- TRAIT_ADD,
- TRAIT_DELETE,
- TRAIT_UPDATE
- }
-
-
- // ----- EntityNotification ------------------------------------------------
-
- /**
- * Get the entity that is associated with this notification.
- *
- * @return the associated entity
- */
- Referenceable getEntity();
-
- /**
- * Get flattened list of traits that are associated with this entity (includes super traits).
- *
- * @return the list of all traits
- */
- List<Struct> getAllTraits();
-
- /**
- * Get the type of operation that triggered this notification.
- *
- * @return the operation type
- */
- OperationType getOperationType();
-}
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/notification/src/main/java/org/apache/atlas/notification/entity/EntityNotificationImpl.java
----------------------------------------------------------------------
diff --git a/notification/src/main/java/org/apache/atlas/notification/entity/EntityNotificationImpl.java b/notification/src/main/java/org/apache/atlas/notification/entity/EntityNotificationImpl.java
deleted file mode 100644
index ab8e4c8..0000000
--- a/notification/src/main/java/org/apache/atlas/notification/entity/EntityNotificationImpl.java
+++ /dev/null
@@ -1,157 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.atlas.notification.entity;
-
-import org.apache.atlas.AtlasException;
-import org.apache.atlas.v1.model.instance.Referenceable;
-import org.apache.atlas.v1.model.instance.Struct;
-import org.apache.atlas.type.AtlasClassificationType;
-import org.apache.atlas.type.AtlasTypeRegistry;
-import org.apache.commons.collections.CollectionUtils;
-import org.apache.commons.collections.MapUtils;
-
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-import java.util.Objects;
-import java.util.Set;
-
-/**
- * Entity notification implementation.
- */
-public class EntityNotificationImpl implements EntityNotification {
-
- private final Referenceable entity;
- private final OperationType operationType;
- private final List<Struct> traits;
-
-
- // ----- Constructors ------------------------------------------------------
-
- /**
- * No-arg constructor for serialization.
- */
- @SuppressWarnings("unused")
- private EntityNotificationImpl() throws AtlasException {
- this(null, OperationType.ENTITY_CREATE, Collections.<Struct>emptyList());
- }
-
- /**
- * Construct an EntityNotification.
- *
- * @param entity the entity subject of the notification
- * @param operationType the type of operation that caused the notification
- * @param traits the traits for the given entity
- *
- * @throws AtlasException if the entity notification can not be created
- */
- public EntityNotificationImpl(Referenceable entity, OperationType operationType, List<Struct> traits)
- throws AtlasException {
- this.entity = entity;
- this.operationType = operationType;
- this.traits = traits;
- }
-
- /**
- * Construct an EntityNotification.
- *
- * @param entity the entity subject of the notification
- * @param operationType the type of operation that caused the notification
- * @param typeRegistry the Atlas type system
- *
- * @throws AtlasException if the entity notification can not be created
- */
- public EntityNotificationImpl(Referenceable entity, OperationType operationType, AtlasTypeRegistry typeRegistry)
- throws AtlasException {
- this(entity, operationType, getAllTraits(entity, typeRegistry));
- }
-
-
- // ----- EntityNotification ------------------------------------------------
-
- @Override
- public Referenceable getEntity() {
- return entity;
- }
-
- @Override
- public List<Struct> getAllTraits() {
- return traits;
- }
-
- @Override
- public OperationType getOperationType() {
- return operationType;
- }
-
-
- // ----- Object overrides --------------------------------------------------
-
- @Override
- public boolean equals(Object o) {
- if (this == o) return true;
- if (o == null || getClass() != o.getClass()) return false;
- EntityNotificationImpl that = (EntityNotificationImpl) o;
- return Objects.equals(entity, that.entity) &&
- operationType == that.operationType &&
- Objects.equals(traits, that.traits);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(entity, operationType, traits);
- }
-
-
- // ----- helper methods ----------------------------------------------------
-
- private static List<Struct> getAllTraits(Referenceable entityDefinition, AtlasTypeRegistry typeRegistry) throws AtlasException {
- List<Struct> ret = new LinkedList<>();
-
- for (String traitName : entityDefinition.getTraitNames()) {
- Struct trait = entityDefinition.getTrait(traitName);
- AtlasClassificationType traitType = typeRegistry.getClassificationTypeByName(traitName);
- Set<String> superTypeNames = traitType != null ? traitType.getAllSuperTypes() : null;
-
- ret.add(trait);
-
- if (CollectionUtils.isNotEmpty(superTypeNames)) {
- for (String superTypeName : superTypeNames) {
- Struct superTypeTrait = new Struct(superTypeName);
-
- if (MapUtils.isNotEmpty(trait.getValues())) {
- AtlasClassificationType superType = typeRegistry.getClassificationTypeByName(superTypeName);
-
- if (superType != null && MapUtils.isNotEmpty(superType.getAllAttributes())) {
- Map<String, Object> attributes = new HashMap<>();
-
- // TODO: add superTypeTrait attributess
-
- superTypeTrait.setValues(attributes);
- }
- }
-
- ret.add(superTypeTrait);
- }
- }
- }
-
- return ret;
- }}
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/notification/src/main/java/org/apache/atlas/notification/hook/HookMessageDeserializer.java
----------------------------------------------------------------------
diff --git a/notification/src/main/java/org/apache/atlas/notification/hook/HookMessageDeserializer.java b/notification/src/main/java/org/apache/atlas/notification/hook/HookMessageDeserializer.java
index 8337de0..1b337d4 100644
--- a/notification/src/main/java/org/apache/atlas/notification/hook/HookMessageDeserializer.java
+++ b/notification/src/main/java/org/apache/atlas/notification/hook/HookMessageDeserializer.java
@@ -18,16 +18,15 @@
package org.apache.atlas.notification.hook;
-import com.google.gson.JsonDeserializer;
+import org.apache.atlas.model.notification.AtlasNotificationMessage;
import org.apache.atlas.notification.AbstractMessageDeserializer;
import org.apache.atlas.notification.AbstractNotification;
-import org.apache.atlas.notification.NotificationInterface;
+import org.apache.atlas.v1.model.notification.HookNotification;
+import org.codehaus.jackson.type.TypeReference;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.lang.reflect.Type;
-import java.util.Collections;
-import java.util.Map;
+
/**
* Hook notification message deserializer.
@@ -46,15 +45,11 @@ public class HookMessageDeserializer extends AbstractMessageDeserializer<HookNot
* Create a hook notification message deserializer.
*/
public HookMessageDeserializer() {
- super(NotificationInterface.HOOK_VERSIONED_MESSAGE_TYPE,
- AbstractNotification.CURRENT_MESSAGE_VERSION, getDeserializerMap(), NOTIFICATION_LOGGER);
+ super(new TypeReference<HookNotification.HookNotificationMessage>() {},
+ new TypeReference<AtlasNotificationMessage<HookNotification.HookNotificationMessage>>() {},
+ AbstractNotification.CURRENT_MESSAGE_VERSION, NOTIFICATION_LOGGER);
}
// ----- helper methods --------------------------------------------------
-
- private static Map<Type, JsonDeserializer> getDeserializerMap() {
- return Collections.<Type, JsonDeserializer>singletonMap(
- NotificationInterface.HOOK_NOTIFICATION_CLASS, new HookNotification());
- }
}
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/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
deleted file mode 100644
index ca596ea..0000000
--- a/notification/src/main/java/org/apache/atlas/notification/hook/HookNotification.java
+++ /dev/null
@@ -1,275 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.atlas.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.v1.model.instance.Referenceable;
-import org.apache.atlas.v1.model.typedef.TypesDef;
-import org.apache.atlas.type.AtlasType;
-import org.apache.commons.lang.StringUtils;
-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;
-
-/**
- * Contains the structure of messages transferred from hooks to atlas.
- */
-public class HookNotification implements JsonDeserializer<HookNotification.HookNotificationMessage> {
-
- @Override
- public HookNotificationMessage deserialize(JsonElement json, Type typeOfT,
- JsonDeserializationContext context) {
- 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 ENTITY_DELETE:
- return context.deserialize(json, EntityDeleteRequest.class);
-
- case TYPE_CREATE:
- case TYPE_UPDATE:
- return context.deserialize(json, TypeRequest.class);
-
- default:
- throw new IllegalStateException("Unhandled type " + type);
- }
- }
-
- /**
- * Type of the hook message.
- */
- public enum HookNotificationType {
- TYPE_CREATE, TYPE_UPDATE, ENTITY_CREATE, ENTITY_PARTIAL_UPDATE, ENTITY_FULL_UPDATE, ENTITY_DELETE
- }
-
- /**
- * Base type of hook message.
- */
- public static class HookNotificationMessage {
- public static final String UNKNOW_USER = "UNKNOWN";
- protected HookNotificationType type;
- protected String user;
-
- private HookNotificationMessage() {
- }
-
- public HookNotificationMessage(HookNotificationType type, String user) {
- this.type = type;
- this.user = user;
- }
-
- public HookNotificationType getType() {
- return type;
- }
-
- public String getUser() {
- if (StringUtils.isEmpty(user)) {
- return UNKNOW_USER;
- }
- return user;
- }
-
-
- }
-
- /**
- * Hook message for create type definitions.
- */
- public static class TypeRequest extends HookNotificationMessage {
- private TypesDef typesDef;
-
- private TypeRequest() {
- }
-
- public TypeRequest(HookNotificationType type, TypesDef typesDef, String user) {
- super(type, user);
- this.typesDef = typesDef;
- }
-
- public TypesDef getTypesDef() {
- return typesDef;
- }
- }
-
- /**
- * Hook message for creating new entities.
- */
- public static class EntityCreateRequest extends HookNotificationMessage {
- private List<Referenceable> entities;
-
- private EntityCreateRequest() {
- }
-
- public EntityCreateRequest(String user, Referenceable... entities) {
- this(HookNotificationType.ENTITY_CREATE, Arrays.asList(entities), user);
- }
-
- public EntityCreateRequest(String user, List<Referenceable> entities) {
- this(HookNotificationType.ENTITY_CREATE, entities, user);
- }
-
- protected EntityCreateRequest(HookNotificationType type, List<Referenceable> entities, String user) {
- super(type, user);
- this.entities = entities;
- }
-
- public EntityCreateRequest(String user, JSONArray jsonArray) {
- super(HookNotificationType.ENTITY_CREATE, user);
- entities = new ArrayList<>();
- for (int index = 0; index < jsonArray.length(); index++) {
- try {
- entities.add(AtlasType.fromV1Json(jsonArray.getString(index), Referenceable.class));
- } catch (JSONException e) {
- throw new JsonParseException(e);
- }
- }
- }
-
- public List<Referenceable> getEntities() {
- return entities;
- }
-
- @Override
- public String toString() {
- return entities.toString();
- }
- }
-
- /**
- * Hook message for updating entities(full update).
- */
- public static class EntityUpdateRequest extends EntityCreateRequest {
- public EntityUpdateRequest(String user, Referenceable... entities) {
- this(user, Arrays.asList(entities));
- }
-
- public EntityUpdateRequest(String user, List<Referenceable> entities) {
- super(HookNotificationType.ENTITY_FULL_UPDATE, entities, user);
- }
- }
-
- /**
- * Hook message for updating entities(partial update).
- */
- public static class EntityPartialUpdateRequest extends HookNotificationMessage {
- private String typeName;
- private String attribute;
- private Referenceable entity;
- private String attributeValue;
-
- private EntityPartialUpdateRequest() {
- }
-
- public EntityPartialUpdateRequest(String user, String typeName, String attribute, String attributeValue,
- Referenceable entity) {
- super(HookNotificationType.ENTITY_PARTIAL_UPDATE, user);
- 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;
- }
-
- @Override
- public String toString() {
- return "{"
- + "entityType='" + typeName + '\''
- + ", attribute=" + attribute
- + ", value=" + attributeValue
- + ", entity=" + entity
- + '}';
- }
- }
-
- /**
- * Hook message for creating new entities.
- */
- public static class EntityDeleteRequest extends HookNotificationMessage {
-
- private String typeName;
- private String attribute;
- private String attributeValue;
-
- private EntityDeleteRequest() {
- }
-
- public EntityDeleteRequest(String user, String typeName, String attribute, String attributeValue) {
- this(HookNotificationType.ENTITY_DELETE, user, typeName, attribute, attributeValue);
- }
-
- protected EntityDeleteRequest(HookNotificationType type,
- String user, String typeName, String attribute, String attributeValue) {
- super(type, user);
- this.typeName = typeName;
- this.attribute = attribute;
- this.attributeValue = attributeValue;
- }
-
- public String getTypeName() {
- return typeName;
- }
-
- public String getAttribute() {
- return attribute;
- }
-
- public String getAttributeValue() {
- return attributeValue;
- }
-
- @Override
- public String toString() {
- return "{"
- + "entityType='" + typeName + '\''
- + ", attribute=" + attribute
- + ", value=" + attributeValue
- + '}';
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/notification/src/test/java/org/apache/atlas/hook/AtlasHookTest.java
----------------------------------------------------------------------
diff --git a/notification/src/test/java/org/apache/atlas/hook/AtlasHookTest.java b/notification/src/test/java/org/apache/atlas/hook/AtlasHookTest.java
index d59cb1c..9ce2a50 100644
--- a/notification/src/test/java/org/apache/atlas/hook/AtlasHookTest.java
+++ b/notification/src/test/java/org/apache/atlas/hook/AtlasHookTest.java
@@ -20,7 +20,7 @@ package org.apache.atlas.hook;
import org.apache.atlas.notification.NotificationException;
import org.apache.atlas.notification.NotificationInterface;
-import org.apache.atlas.notification.hook.HookNotification;
+import org.apache.atlas.v1.model.notification.HookNotification;
import org.mockito.Mock;
import org.mockito.MockitoAnnotations;
import org.testng.annotations.BeforeMethod;
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/notification/src/test/java/org/apache/atlas/kafka/KafkaConsumerTest.java
----------------------------------------------------------------------
diff --git a/notification/src/test/java/org/apache/atlas/kafka/KafkaConsumerTest.java b/notification/src/test/java/org/apache/atlas/kafka/KafkaConsumerTest.java
index 071a725..f1fc741 100644
--- a/notification/src/test/java/org/apache/atlas/kafka/KafkaConsumerTest.java
+++ b/notification/src/test/java/org/apache/atlas/kafka/KafkaConsumerTest.java
@@ -22,10 +22,11 @@ import kafka.message.MessageAndMetadata;
import org.apache.atlas.v1.model.instance.Referenceable;
import org.apache.atlas.v1.model.instance.Struct;
import org.apache.atlas.notification.*;
-import org.apache.atlas.notification.AtlasNotificationMessage;
-import org.apache.atlas.notification.entity.EntityNotificationImplTest;
-import org.apache.atlas.notification.hook.HookNotification;
+import org.apache.atlas.model.notification.AtlasNotificationMessage;
+import org.apache.atlas.notification.entity.EntityNotificationTest;
+import org.apache.atlas.v1.model.notification.HookNotification;
import org.apache.atlas.type.AtlasType;
+import org.apache.atlas.model.notification.MessageVersion;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
@@ -96,7 +97,7 @@ public class KafkaConsumerTest {
when(messageAndMetadata.message()).thenReturn(json);
- AtlasKafkaConsumer consumer = new AtlasKafkaConsumer(NotificationInterface.NotificationType.HOOK.getDeserializer(), kafkaConsumer, false, 100L);
+ AtlasKafkaConsumer consumer = new AtlasKafkaConsumer(NotificationInterface.NotificationType.HOOK, kafkaConsumer, false, 100L);
List<AtlasKafkaMessage<HookNotification.HookNotificationMessage>> messageList = consumer.receive();
assertTrue(messageList.size() > 0);
@@ -131,7 +132,7 @@ public class KafkaConsumerTest {
when(kafkaConsumer.poll(100L)).thenReturn(records);
when(messageAndMetadata.message()).thenReturn(json);
- AtlasKafkaConsumer consumer =new AtlasKafkaConsumer(NotificationInterface.NotificationType.HOOK.getDeserializer(), kafkaConsumer ,false, 100L);
+ AtlasKafkaConsumer consumer =new AtlasKafkaConsumer(NotificationInterface.NotificationType.HOOK, kafkaConsumer ,false, 100L);
try {
List<AtlasKafkaMessage<HookNotification.HookNotificationMessage>> messageList = consumer.receive();
assertTrue(messageList.size() > 0);
@@ -151,7 +152,7 @@ public class KafkaConsumerTest {
TopicPartition tp = new TopicPartition("ATLAS_HOOK",0);
- AtlasKafkaConsumer consumer =new AtlasKafkaConsumer(NotificationInterface.NotificationType.HOOK.getDeserializer(), kafkaConsumer, false, 100L);
+ AtlasKafkaConsumer consumer =new AtlasKafkaConsumer(NotificationInterface.NotificationType.HOOK, kafkaConsumer, false, 100L);
consumer.commit(tp, 1);
@@ -163,7 +164,7 @@ public class KafkaConsumerTest {
TopicPartition tp = new TopicPartition("ATLAS_HOOK",0);
- AtlasKafkaConsumer consumer =new AtlasKafkaConsumer(NotificationInterface.NotificationType.HOOK.getDeserializer(), kafkaConsumer, true , 100L);
+ AtlasKafkaConsumer consumer =new AtlasKafkaConsumer(NotificationInterface.NotificationType.HOOK, kafkaConsumer, true , 100L);
consumer.commit(tp, 1);
@@ -171,7 +172,7 @@ public class KafkaConsumerTest {
}
private Referenceable getEntity(String traitName) {
- Referenceable entity = EntityNotificationImplTest.getEntity("id");
+ Referenceable entity = EntityNotificationTest.getEntity("id");
List<Struct> traitInfo = new LinkedList<>();
Struct trait = new Struct(traitName, Collections.<String, Object>emptyMap());
traitInfo.add(trait);
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/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 5e3cf41..fe019e1 100644
--- a/notification/src/test/java/org/apache/atlas/kafka/KafkaNotificationTest.java
+++ b/notification/src/test/java/org/apache/atlas/kafka/KafkaNotificationTest.java
@@ -22,13 +22,13 @@ import org.apache.atlas.ApplicationProperties;
import org.apache.atlas.v1.model.instance.Referenceable;
import org.apache.atlas.notification.NotificationConsumer;
import org.apache.atlas.notification.NotificationInterface;
-import org.apache.atlas.notification.hook.HookNotification;
+import org.apache.atlas.v1.model.notification.HookNotification;
import org.apache.commons.configuration.Configuration;
import org.apache.commons.lang.RandomStringUtils;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
-import static org.apache.atlas.notification.hook.HookNotification.HookNotificationMessage;
+import static org.apache.atlas.v1.model.notification.HookNotification.HookNotificationMessage;
import java.util.List;
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/notification/src/test/java/org/apache/atlas/notification/AbstractNotificationConsumerTest.java
----------------------------------------------------------------------
diff --git a/notification/src/test/java/org/apache/atlas/notification/AbstractNotificationConsumerTest.java b/notification/src/test/java/org/apache/atlas/notification/AbstractNotificationConsumerTest.java
index f313ddc..caa72ce 100644
--- a/notification/src/test/java/org/apache/atlas/notification/AbstractNotificationConsumerTest.java
+++ b/notification/src/test/java/org/apache/atlas/notification/AbstractNotificationConsumerTest.java
@@ -18,13 +18,15 @@
package org.apache.atlas.notification;
-import com.google.gson.Gson;
-import com.google.gson.reflect.TypeToken;
import org.apache.atlas.kafka.AtlasKafkaMessage;
+import org.apache.atlas.model.notification.AtlasNotificationMessage;
+import org.apache.atlas.type.AtlasType;
+import org.apache.atlas.model.notification.MessageVersion;
+import org.codehaus.jackson.type.TypeReference;
import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import org.testng.annotations.Test;
-import java.lang.reflect.Type;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
@@ -41,8 +43,6 @@ import org.apache.kafka.common.TopicPartition;
*/
public class AbstractNotificationConsumerTest {
- private static final Gson GSON = new Gson();
-
@Test
public void testReceive() throws Exception {
Logger logger = mock(Logger.class);
@@ -54,27 +54,24 @@ public class AbstractNotificationConsumerTest {
List jsonList = new LinkedList<>();
- jsonList.add(GSON.toJson(new AtlasNotificationMessage<>(new MessageVersion("1.0.0"), testMessage1)));
- jsonList.add(GSON.toJson(new AtlasNotificationMessage<>(new MessageVersion("1.0.0"), testMessage2)));
- jsonList.add(GSON.toJson(new AtlasNotificationMessage<>(new MessageVersion("1.0.0"), testMessage3)));
- jsonList.add(GSON.toJson(new AtlasNotificationMessage<>(new MessageVersion("1.0.0"), testMessage4)));
-
- Type notificationMessageType = new TypeToken<AtlasNotificationMessage<TestMessage>>(){}.getType();
+ jsonList.add(AtlasType.toV1Json(new AtlasNotificationMessage<>(new MessageVersion("1.0.0"), testMessage1)));
+ jsonList.add(AtlasType.toV1Json(new AtlasNotificationMessage<>(new MessageVersion("1.0.0"), testMessage2)));
+ jsonList.add(AtlasType.toV1Json(new AtlasNotificationMessage<>(new MessageVersion("1.0.0"), testMessage3)));
+ jsonList.add(AtlasType.toV1Json(new AtlasNotificationMessage<>(new MessageVersion("1.0.0"), testMessage4)));
- NotificationConsumer<TestMessage> consumer =
- new TestNotificationConsumer<>(notificationMessageType, jsonList, logger);
+ NotificationConsumer<TestMessage> consumer = new TestNotificationConsumer(jsonList, logger);
List<AtlasKafkaMessage<TestMessage>> messageList = consumer.receive();
assertFalse(messageList.isEmpty());
- assertEquals(testMessage1, messageList.get(0).getMessage());
+ assertEquals(messageList.get(0).getMessage(), testMessage1);
- assertEquals(testMessage2, messageList.get(1).getMessage());
+ assertEquals(messageList.get(1).getMessage(), testMessage2);
- assertEquals(testMessage3, messageList.get(2).getMessage());
+ assertEquals(messageList.get(2).getMessage(), testMessage3);
- assertEquals(testMessage4, messageList.get(3).getMessage());
+ assertEquals(messageList.get(3).getMessage(), testMessage4);
}
@Test
@@ -88,20 +85,17 @@ public class AbstractNotificationConsumerTest {
List jsonList = new LinkedList<>();
- String json1 = GSON.toJson(new AtlasNotificationMessage<>(new MessageVersion("1.0.0"), testMessage1));
- String json2 = GSON.toJson(new AtlasNotificationMessage<>(new MessageVersion("0.0.5"), testMessage2));
- String json3 = GSON.toJson(new AtlasNotificationMessage<>(new MessageVersion("0.5.0"), testMessage3));
- String json4 = GSON.toJson(testMessage4);
+ String json1 = AtlasType.toV1Json(new AtlasNotificationMessage<>(new MessageVersion("1.0.0"), testMessage1));
+ String json2 = AtlasType.toV1Json(new AtlasNotificationMessage<>(new MessageVersion("0.0.5"), testMessage2));
+ String json3 = AtlasType.toV1Json(new AtlasNotificationMessage<>(new MessageVersion("0.5.0"), testMessage3));
+ String json4 = AtlasType.toV1Json(testMessage4);
jsonList.add(json1);
jsonList.add(json2);
jsonList.add(json3);
jsonList.add(json4);
- Type notificationMessageType = new TypeToken<AtlasNotificationMessage<TestMessage>>(){}.getType();
-
- NotificationConsumer<TestMessage> consumer =
- new TestNotificationConsumer<>(notificationMessageType, jsonList, logger);
+ NotificationConsumer<TestMessage> consumer = new TestNotificationConsumer(jsonList, logger);
List<AtlasKafkaMessage<TestMessage>> messageList = consumer.receive();
@@ -124,16 +118,13 @@ public class AbstractNotificationConsumerTest {
List jsonList = new LinkedList<>();
- String json1 = GSON.toJson(new AtlasNotificationMessage<>(new MessageVersion("1.0.0"), testMessage1));
- String json2 = GSON.toJson(new AtlasNotificationMessage<>(new MessageVersion("2.0.0"), testMessage2));
+ String json1 = AtlasType.toV1Json(new AtlasNotificationMessage<>(new MessageVersion("1.0.0"), testMessage1));
+ String json2 = AtlasType.toV1Json(new AtlasNotificationMessage<>(new MessageVersion("2.0.0"), testMessage2));
jsonList.add(json1);
jsonList.add(json2);
- Type notificationMessageType = new TypeToken<AtlasNotificationMessage<TestMessage>>(){}.getType();
-
- NotificationConsumer<TestMessage> consumer =
- new TestNotificationConsumer<>(notificationMessageType, jsonList, logger);
+ NotificationConsumer<TestMessage> consumer = new TestNotificationConsumer(jsonList, logger);
try {
List<AtlasKafkaMessage<TestMessage>> messageList = consumer.receive();
@@ -150,7 +141,10 @@ public class AbstractNotificationConsumerTest {
private static class TestMessage {
private String s;
- private int i;
+ private int i;
+
+ public TestMessage() {
+ }
public TestMessage(String s, int i) {
this.s = s;
@@ -165,6 +159,14 @@ public class AbstractNotificationConsumerTest {
this.s = s;
}
+ public int getI() {
+ return i;
+ }
+
+ public void setI(int i) {
+ this.i = i;
+ }
+
@Override
public boolean equals(Object o) {
if (this == o) return true;
@@ -180,12 +182,14 @@ public class AbstractNotificationConsumerTest {
}
}
- private static class TestNotificationConsumer<T> extends AbstractNotificationConsumer<T> {
- private final List<T> messageList;
- private int index = 0;
+ private static class TestNotificationConsumer extends AbstractNotificationConsumer<TestMessage> {
+ private final List<TestMessage> messageList;
+ private int index = 0;
+
+
+ public TestNotificationConsumer(List<TestMessage> messages, Logger logger) {
+ super(new TestMessageDeserializer());
- public TestNotificationConsumer(Type notificationMessageType, List<T> messages, Logger logger) {
- super(new TestDeserializer<T>(notificationMessageType, logger));
this.messageList = messages;
}
@@ -205,24 +209,35 @@ public class AbstractNotificationConsumerTest {
}
@Override
- public List<AtlasKafkaMessage<T>> receive() {
+ public List<AtlasKafkaMessage<TestMessage>> receive() {
return receive(1000L);
}
@Override
- public List<AtlasKafkaMessage<T>> receive(long timeoutMilliSeconds) {
- List<AtlasKafkaMessage<T>> tempMessageList = new ArrayList();
+ public List<AtlasKafkaMessage<TestMessage>> receive(long timeoutMilliSeconds) {
+ List<AtlasKafkaMessage<TestMessage>> tempMessageList = new ArrayList();
for(Object json : messageList) {
- tempMessageList.add(new AtlasKafkaMessage(deserializer.deserialize((String)json), -1, -1));
+ tempMessageList.add(new AtlasKafkaMessage(deserializer.deserialize((String) json), -1, -1));
}
return tempMessageList;
}
}
- private static final class TestDeserializer<T> extends AtlasNotificationMessageDeserializer<T> {
+ public static class TestMessageDeserializer extends AbstractMessageDeserializer<TestMessage> {
+ /**
+ * Logger for hook notification messages.
+ */
+ private static final Logger NOTIFICATION_LOGGER = LoggerFactory.getLogger(TestMessageDeserializer.class);
+
+
+ // ----- Constructors ----------------------------------------------------
- private TestDeserializer(Type notificationMessageType, Logger logger) {
- super(notificationMessageType, AbstractNotification.CURRENT_MESSAGE_VERSION, GSON, logger);
+ /**
+ * Create a hook notification message deserializer.
+ */
+ public TestMessageDeserializer() {
+ super(new TypeReference<TestMessage>() {}, new TypeReference<AtlasNotificationMessage<TestMessage>>() {},
+ AbstractNotification.CURRENT_MESSAGE_VERSION, NOTIFICATION_LOGGER);
}
}
}
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/notification/src/test/java/org/apache/atlas/notification/AbstractNotificationTest.java
----------------------------------------------------------------------
diff --git a/notification/src/test/java/org/apache/atlas/notification/AbstractNotificationTest.java b/notification/src/test/java/org/apache/atlas/notification/AbstractNotificationTest.java
index 655252c..98d7d2c 100644
--- a/notification/src/test/java/org/apache/atlas/notification/AbstractNotificationTest.java
+++ b/notification/src/test/java/org/apache/atlas/notification/AbstractNotificationTest.java
@@ -19,7 +19,8 @@
package org.apache.atlas.notification;
import org.apache.atlas.AtlasException;
-import org.apache.atlas.notification.hook.HookNotification;
+import org.apache.atlas.type.AtlasType;
+import org.apache.atlas.v1.model.notification.HookNotification;
import org.apache.commons.configuration.Configuration;
import org.testng.annotations.Test;
@@ -98,8 +99,8 @@ public class AbstractNotificationTest {
// ignore msgCreationTime in Json
private void assertEqualsMessageJson(String msgJsonActual, String msgJsonExpected) {
- Map<Object, Object> msgActual = AbstractNotification.GSON.fromJson(msgJsonActual, Map.class);
- Map<Object, Object> msgExpected = AbstractNotification.GSON.fromJson(msgJsonExpected, Map.class);
+ Map<Object, Object> msgActual = AtlasType.fromV1Json(msgJsonActual, Map.class);
+ Map<Object, Object> msgExpected = AtlasType.fromV1Json(msgJsonExpected, Map.class);
msgActual.remove("msgCreationTime");
msgExpected.remove("msgCreationTime");
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/notification/src/test/java/org/apache/atlas/notification/AtlasNotificationMessageTest.java
----------------------------------------------------------------------
diff --git a/notification/src/test/java/org/apache/atlas/notification/AtlasNotificationMessageTest.java b/notification/src/test/java/org/apache/atlas/notification/AtlasNotificationMessageTest.java
index 27b5034..91a195d 100644
--- a/notification/src/test/java/org/apache/atlas/notification/AtlasNotificationMessageTest.java
+++ b/notification/src/test/java/org/apache/atlas/notification/AtlasNotificationMessageTest.java
@@ -18,6 +18,8 @@
package org.apache.atlas.notification;
+import org.apache.atlas.model.notification.AtlasNotificationMessage;
+import org.apache.atlas.model.notification.MessageVersion;
import org.testng.annotations.Test;
import static org.testng.Assert.*;
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/notification/src/test/java/org/apache/atlas/notification/MessageVersionTest.java
----------------------------------------------------------------------
diff --git a/notification/src/test/java/org/apache/atlas/notification/MessageVersionTest.java b/notification/src/test/java/org/apache/atlas/notification/MessageVersionTest.java
index d1af4b0..d8b3b34 100644
--- a/notification/src/test/java/org/apache/atlas/notification/MessageVersionTest.java
+++ b/notification/src/test/java/org/apache/atlas/notification/MessageVersionTest.java
@@ -18,6 +18,7 @@
package org.apache.atlas.notification;
+import org.apache.atlas.model.notification.MessageVersion;
import org.testng.annotations.Test;
import java.util.Arrays;
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/notification/src/test/java/org/apache/atlas/notification/SplitMessageAggregatorTest.java
----------------------------------------------------------------------
diff --git a/notification/src/test/java/org/apache/atlas/notification/SplitMessageAggregatorTest.java b/notification/src/test/java/org/apache/atlas/notification/SplitMessageAggregatorTest.java
index 0807221..b79735a 100644
--- a/notification/src/test/java/org/apache/atlas/notification/SplitMessageAggregatorTest.java
+++ b/notification/src/test/java/org/apache/atlas/notification/SplitMessageAggregatorTest.java
@@ -17,7 +17,8 @@
*/
package org.apache.atlas.notification;
-import org.apache.atlas.notification.AtlasNotificationBaseMessage.CompressionKind;
+import org.apache.atlas.model.notification.AtlasNotificationBaseMessage.CompressionKind;
+import org.apache.atlas.model.notification.AtlasNotificationStringMessage;
import org.testng.Assert;
import org.testng.annotations.Test;
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/notification/src/test/java/org/apache/atlas/notification/entity/EntityMessageDeserializerTest.java
----------------------------------------------------------------------
diff --git a/notification/src/test/java/org/apache/atlas/notification/entity/EntityMessageDeserializerTest.java b/notification/src/test/java/org/apache/atlas/notification/entity/EntityMessageDeserializerTest.java
index faafb87..ddb63b5 100644
--- a/notification/src/test/java/org/apache/atlas/notification/entity/EntityMessageDeserializerTest.java
+++ b/notification/src/test/java/org/apache/atlas/notification/entity/EntityMessageDeserializerTest.java
@@ -21,6 +21,7 @@ package org.apache.atlas.notification.entity;
import org.apache.atlas.v1.model.instance.Referenceable;
import org.apache.atlas.v1.model.instance.Struct;
import org.apache.atlas.notification.AbstractNotification;
+import org.apache.atlas.v1.model.notification.EntityNotification;
import org.testng.annotations.Test;
import java.util.ArrayList;
@@ -34,19 +35,18 @@ import static org.testng.Assert.assertEquals;
* EntityMessageDeserializer tests.
*/
public class EntityMessageDeserializerTest {
+ private EntityMessageDeserializer deserializer = new EntityMessageDeserializer();
@Test
public void testDeserialize() throws Exception {
- EntityMessageDeserializer deserializer = new EntityMessageDeserializer();
-
- Referenceable entity = EntityNotificationImplTest.getEntity("id");
+ Referenceable entity = EntityNotificationTest.getEntity("id");
String traitName = "MyTrait";
List<Struct> traitInfo = new LinkedList<>();
Struct trait = new Struct(traitName, Collections.<String, Object>emptyMap());
traitInfo.add(trait);
- EntityNotificationImpl notification =
- new EntityNotificationImpl(entity, EntityNotification.OperationType.TRAIT_ADD, traitInfo);
+ EntityNotification notification =
+ new EntityNotification(entity, EntityNotification.OperationType.TRAIT_ADD, traitInfo);
List<String> jsonMsgList = new ArrayList<>();
@@ -55,7 +55,7 @@ public class EntityMessageDeserializerTest {
EntityNotification deserializedNotification = null;
for (String jsonMsg : jsonMsgList) {
- deserializedNotification = deserializer.deserialize(jsonMsg);
+ deserializedNotification = deserializer.deserialize(jsonMsg);
if (deserializedNotification != null) {
break;
[3/3] atlas git commit: ATLAS-2251: notification module updates
Posted by ma...@apache.org.
ATLAS-2251: notification module updates
Project: http://git-wip-us.apache.org/repos/asf/atlas/repo
Commit: http://git-wip-us.apache.org/repos/asf/atlas/commit/64e739da
Tree: http://git-wip-us.apache.org/repos/asf/atlas/tree/64e739da
Diff: http://git-wip-us.apache.org/repos/asf/atlas/diff/64e739da
Branch: refs/heads/ATLAS-2251
Commit: 64e739da7e3bb36398c5ceb966016b435ae76a00
Parents: 3f44770
Author: Madhan Neethiraj <ma...@apache.org>
Authored: Sun Nov 5 15:05:12 2017 -0800
Committer: Madhan Neethiraj <ma...@apache.org>
Committed: Mon Nov 6 14:49:49 2017 -0800
----------------------------------------------------------------------
.../apache/atlas/falcon/hook/FalconHook.java | 2 +-
.../atlas/hbase/bridge/HBaseAtlasHook.java | 2 +-
.../hbase/model/HBaseOperationContext.java | 2 +-
.../org/apache/atlas/hive/hook/HiveHook.java | 2 +-
.../org/apache/atlas/sqoop/hook/SqoopHook.java | 2 +-
.../org/apache/atlas/AtlasConfiguration.java | 83 ----
.../org/apache/atlas/AtlasConfiguration.java | 83 ++++
.../AtlasNotificationBaseMessage.java | 208 ++++++++++
.../notification/AtlasNotificationMessage.java | 101 +++++
.../AtlasNotificationStringMessage.java | 82 ++++
.../model/notification/MessageVersion.java | 170 ++++++++
.../java/org/apache/atlas/type/AtlasType.java | 62 ++-
.../atlas/typesystem/types/DataTypes.java | 4 +-
.../model/instance/AtlasSystemAttributes.java | 21 +
.../org/apache/atlas/v1/model/instance/Id.java | 19 +
.../atlas/v1/model/instance/Referenceable.java | 27 ++
.../apache/atlas/v1/model/instance/Struct.java | 21 +
.../model/notification/EntityNotification.java | 218 ++++++++++
.../v1/model/notification/HookNotification.java | 415 +++++++++++++++++++
.../apache/atlas/v1/model/typedef/TypesDef.java | 27 ++
.../java/org/apache/atlas/hook/AtlasHook.java | 2 +-
.../apache/atlas/kafka/AtlasKafkaConsumer.java | 18 +-
.../apache/atlas/kafka/KafkaNotification.java | 5 +-
.../AbstractMessageDeserializer.java | 129 +-----
.../notification/AbstractNotification.java | 28 +-
.../AbstractNotificationConsumer.java | 19 +-
.../AtlasNotificationBaseMessage.java | 194 ---------
.../notification/AtlasNotificationMessage.java | 87 ----
.../AtlasNotificationMessageDeserializer.java | 59 +--
.../AtlasNotificationStringMessage.java | 66 ---
.../atlas/notification/MessageVersion.java | 141 -------
.../notification/NotificationInterface.java | 39 +-
.../notification/SplitMessageAggregator.java | 2 +
.../entity/EntityMessageDeserializer.java | 38 +-
.../notification/entity/EntityNotification.java | 66 ---
.../entity/EntityNotificationImpl.java | 157 -------
.../hook/HookMessageDeserializer.java | 19 +-
.../notification/hook/HookNotification.java | 275 ------------
.../org/apache/atlas/hook/AtlasHookTest.java | 2 +-
.../apache/atlas/kafka/KafkaConsumerTest.java | 17 +-
.../atlas/kafka/KafkaNotificationTest.java | 4 +-
.../AbstractNotificationConsumerTest.java | 103 +++--
.../notification/AbstractNotificationTest.java | 7 +-
.../AtlasNotificationMessageTest.java | 2 +
.../atlas/notification/MessageVersionTest.java | 1 +
.../SplitMessageAggregatorTest.java | 3 +-
.../entity/EntityMessageDeserializerTest.java | 12 +-
.../entity/EntityNotificationImplTest.java | 147 -------
.../entity/EntityNotificationTest.java | 148 +++++++
.../hook/HookMessageDeserializerTest.java | 16 +-
.../notification/hook/HookNotificationTest.java | 12 +-
.../test/java/org/apache/atlas/DBSandboxer.java | 53 +++
.../NotificationEntityChangeListener.java | 27 +-
.../notification/NotificationHookConsumer.java | 16 +-
.../notification/EntityNotificationIT.java | 2 +-
.../NotificationHookConsumerIT.java | 12 +-
.../NotificationHookConsumerKafkaTest.java | 4 +-
.../NotificationHookConsumerTest.java | 2 +-
.../atlas/web/integration/BaseResourceIT.java | 2 +-
59 files changed, 1893 insertions(+), 1594 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/addons/falcon-bridge/src/main/java/org/apache/atlas/falcon/hook/FalconHook.java
----------------------------------------------------------------------
diff --git a/addons/falcon-bridge/src/main/java/org/apache/atlas/falcon/hook/FalconHook.java b/addons/falcon-bridge/src/main/java/org/apache/atlas/falcon/hook/FalconHook.java
index e9b9765..5912cb0 100644
--- a/addons/falcon-bridge/src/main/java/org/apache/atlas/falcon/hook/FalconHook.java
+++ b/addons/falcon-bridge/src/main/java/org/apache/atlas/falcon/hook/FalconHook.java
@@ -26,7 +26,7 @@ import org.apache.atlas.falcon.publisher.FalconEventPublisher;
import org.apache.atlas.hook.AtlasHook;
import org.apache.atlas.kafka.NotificationProvider;
import org.apache.atlas.v1.model.instance.Referenceable;
-import org.apache.atlas.notification.hook.HookNotification;
+import org.apache.atlas.v1.model.notification.HookNotification;
import org.apache.falcon.FalconException;
import org.apache.falcon.entity.store.ConfigurationStore;
import org.apache.falcon.entity.v0.feed.Feed;
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/addons/hbase-bridge/src/main/java/org/apache/atlas/hbase/bridge/HBaseAtlasHook.java
----------------------------------------------------------------------
diff --git a/addons/hbase-bridge/src/main/java/org/apache/atlas/hbase/bridge/HBaseAtlasHook.java b/addons/hbase-bridge/src/main/java/org/apache/atlas/hbase/bridge/HBaseAtlasHook.java
index af5eda8..6fcaf1b 100644
--- a/addons/hbase-bridge/src/main/java/org/apache/atlas/hbase/bridge/HBaseAtlasHook.java
+++ b/addons/hbase-bridge/src/main/java/org/apache/atlas/hbase/bridge/HBaseAtlasHook.java
@@ -24,7 +24,7 @@ import org.apache.atlas.hbase.model.HBaseOperationContext;
import org.apache.atlas.hbase.model.HBaseDataTypes;
import org.apache.atlas.hook.AtlasHook;
import org.apache.atlas.v1.model.instance.Referenceable;
-import org.apache.atlas.notification.hook.HookNotification;
+import org.apache.atlas.v1.model.notification.HookNotification;
import org.apache.commons.configuration.Configuration;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.HColumnDescriptor;
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/addons/hbase-bridge/src/main/java/org/apache/atlas/hbase/model/HBaseOperationContext.java
----------------------------------------------------------------------
diff --git a/addons/hbase-bridge/src/main/java/org/apache/atlas/hbase/model/HBaseOperationContext.java b/addons/hbase-bridge/src/main/java/org/apache/atlas/hbase/model/HBaseOperationContext.java
index ce0f212..33858d4 100644
--- a/addons/hbase-bridge/src/main/java/org/apache/atlas/hbase/model/HBaseOperationContext.java
+++ b/addons/hbase-bridge/src/main/java/org/apache/atlas/hbase/model/HBaseOperationContext.java
@@ -19,7 +19,7 @@
package org.apache.atlas.hbase.model;
import org.apache.atlas.hbase.bridge.HBaseAtlasHook;
-import org.apache.atlas.notification.hook.HookNotification;
+import org.apache.atlas.v1.model.notification.HookNotification;
import org.apache.hadoop.hbase.NamespaceDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.HColumnDescriptor;
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/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 9a5d1b9..5f8dcdb 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
@@ -29,7 +29,7 @@ import org.apache.atlas.hive.model.HiveDataTypes;
import org.apache.atlas.hook.AtlasHook;
import org.apache.atlas.hook.AtlasHookException;
import org.apache.atlas.v1.model.instance.Referenceable;
-import org.apache.atlas.notification.hook.HookNotification;
+import org.apache.atlas.v1.model.notification.HookNotification;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.hadoop.fs.Path;
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/addons/sqoop-bridge/src/main/java/org/apache/atlas/sqoop/hook/SqoopHook.java
----------------------------------------------------------------------
diff --git a/addons/sqoop-bridge/src/main/java/org/apache/atlas/sqoop/hook/SqoopHook.java b/addons/sqoop-bridge/src/main/java/org/apache/atlas/sqoop/hook/SqoopHook.java
index 6c1aa99..aee24ab 100644
--- a/addons/sqoop-bridge/src/main/java/org/apache/atlas/sqoop/hook/SqoopHook.java
+++ b/addons/sqoop-bridge/src/main/java/org/apache/atlas/sqoop/hook/SqoopHook.java
@@ -27,7 +27,7 @@ import org.apache.atlas.hive.model.HiveDataTypes;
import org.apache.atlas.hook.AtlasHook;
import org.apache.atlas.hook.AtlasHookException;
import org.apache.atlas.v1.model.instance.Referenceable;
-import org.apache.atlas.notification.hook.HookNotification;
+import org.apache.atlas.v1.model.notification.HookNotification;
import org.apache.atlas.sqoop.model.SqoopDataTypes;
import org.apache.commons.configuration.Configuration;
import org.apache.commons.lang3.StringUtils;
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/common/src/main/java/org/apache/atlas/AtlasConfiguration.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/atlas/AtlasConfiguration.java b/common/src/main/java/org/apache/atlas/AtlasConfiguration.java
deleted file mode 100644
index bd2bf7f..0000000
--- a/common/src/main/java/org/apache/atlas/AtlasConfiguration.java
+++ /dev/null
@@ -1,83 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.atlas;
-
-import org.apache.commons.configuration.Configuration;
-
-/**
- * Enum that encapsulated each property name and its default value.
- */
-public enum AtlasConfiguration {
- //web server configuration
- WEBSERVER_MIN_THREADS("atlas.webserver.minthreads", 10),
- WEBSERVER_MAX_THREADS("atlas.webserver.maxthreads", 100),
- WEBSERVER_KEEPALIVE_SECONDS("atlas.webserver.keepalivetimesecs", 60),
- WEBSERVER_QUEUE_SIZE("atlas.webserver.queuesize", 100),
- WEBSERVER_REQUEST_BUFFER_SIZE("atlas.jetty.request.buffer.size", 16192),
-
- QUERY_PARAM_MAX_LENGTH("atlas.query.param.max.length", 4*1024),
-
- NOTIFICATION_MESSAGE_MAX_LENGTH_BYTES("atlas.notification.message.max.length.bytes", (1000 * 1000)),
- NOTIFICATION_MESSAGE_COMPRESSION_ENABLED("atlas.notification.message.compression.enabled", true),
- NOTIFICATION_SPLIT_MESSAGE_SEGMENTS_WAIT_TIME_SECONDS("atlas.notification.split.message.segments.wait.time.seconds", 15 * 60),
- NOTIFICATION_SPLIT_MESSAGE_BUFFER_PURGE_INTERVAL_SECONDS("atlas.notification.split.message.buffer.purge.interval.seconds", 5 * 60),
-
- //search configuration
- SEARCH_MAX_LIMIT("atlas.search.maxlimit", 10000),
- SEARCH_DEFAULT_LIMIT("atlas.search.defaultlimit", 100);
-
- private static final Configuration APPLICATION_PROPERTIES;
-
- static {
- try {
- APPLICATION_PROPERTIES = ApplicationProperties.get();
- } catch (AtlasException e) {
- throw new RuntimeException(e);
- }
- }
-
- private final String propertyName;
- private final Object defaultValue;
-
- AtlasConfiguration(String propertyName, Object defaultValue) {
- this.propertyName = propertyName;
- this.defaultValue = defaultValue;
- }
-
- public int getInt() {
- return APPLICATION_PROPERTIES.getInt(propertyName, Integer.valueOf(defaultValue.toString()).intValue());
- }
-
- public long getLong() {
- return APPLICATION_PROPERTIES.getLong(propertyName, Long.valueOf(defaultValue.toString()).longValue());
- }
-
- public boolean getBoolean() {
- return APPLICATION_PROPERTIES.getBoolean(propertyName, Boolean.valueOf(defaultValue.toString()).booleanValue());
- }
-
- public String getString() {
- return APPLICATION_PROPERTIES.getString(propertyName, defaultValue.toString());
- }
-
- public Object get() {
- Object value = APPLICATION_PROPERTIES.getProperty(propertyName);
- return value == null ? defaultValue : value;
- }
-}
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/intg/src/main/java/org/apache/atlas/AtlasConfiguration.java
----------------------------------------------------------------------
diff --git a/intg/src/main/java/org/apache/atlas/AtlasConfiguration.java b/intg/src/main/java/org/apache/atlas/AtlasConfiguration.java
new file mode 100644
index 0000000..bd2bf7f
--- /dev/null
+++ b/intg/src/main/java/org/apache/atlas/AtlasConfiguration.java
@@ -0,0 +1,83 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.atlas;
+
+import org.apache.commons.configuration.Configuration;
+
+/**
+ * Enum that encapsulated each property name and its default value.
+ */
+public enum AtlasConfiguration {
+ //web server configuration
+ WEBSERVER_MIN_THREADS("atlas.webserver.minthreads", 10),
+ WEBSERVER_MAX_THREADS("atlas.webserver.maxthreads", 100),
+ WEBSERVER_KEEPALIVE_SECONDS("atlas.webserver.keepalivetimesecs", 60),
+ WEBSERVER_QUEUE_SIZE("atlas.webserver.queuesize", 100),
+ WEBSERVER_REQUEST_BUFFER_SIZE("atlas.jetty.request.buffer.size", 16192),
+
+ QUERY_PARAM_MAX_LENGTH("atlas.query.param.max.length", 4*1024),
+
+ NOTIFICATION_MESSAGE_MAX_LENGTH_BYTES("atlas.notification.message.max.length.bytes", (1000 * 1000)),
+ NOTIFICATION_MESSAGE_COMPRESSION_ENABLED("atlas.notification.message.compression.enabled", true),
+ NOTIFICATION_SPLIT_MESSAGE_SEGMENTS_WAIT_TIME_SECONDS("atlas.notification.split.message.segments.wait.time.seconds", 15 * 60),
+ NOTIFICATION_SPLIT_MESSAGE_BUFFER_PURGE_INTERVAL_SECONDS("atlas.notification.split.message.buffer.purge.interval.seconds", 5 * 60),
+
+ //search configuration
+ SEARCH_MAX_LIMIT("atlas.search.maxlimit", 10000),
+ SEARCH_DEFAULT_LIMIT("atlas.search.defaultlimit", 100);
+
+ private static final Configuration APPLICATION_PROPERTIES;
+
+ static {
+ try {
+ APPLICATION_PROPERTIES = ApplicationProperties.get();
+ } catch (AtlasException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private final String propertyName;
+ private final Object defaultValue;
+
+ AtlasConfiguration(String propertyName, Object defaultValue) {
+ this.propertyName = propertyName;
+ this.defaultValue = defaultValue;
+ }
+
+ public int getInt() {
+ return APPLICATION_PROPERTIES.getInt(propertyName, Integer.valueOf(defaultValue.toString()).intValue());
+ }
+
+ public long getLong() {
+ return APPLICATION_PROPERTIES.getLong(propertyName, Long.valueOf(defaultValue.toString()).longValue());
+ }
+
+ public boolean getBoolean() {
+ return APPLICATION_PROPERTIES.getBoolean(propertyName, Boolean.valueOf(defaultValue.toString()).booleanValue());
+ }
+
+ public String getString() {
+ return APPLICATION_PROPERTIES.getString(propertyName, defaultValue.toString());
+ }
+
+ public Object get() {
+ Object value = APPLICATION_PROPERTIES.getProperty(propertyName);
+ return value == null ? defaultValue : value;
+ }
+}
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/intg/src/main/java/org/apache/atlas/model/notification/AtlasNotificationBaseMessage.java
----------------------------------------------------------------------
diff --git a/intg/src/main/java/org/apache/atlas/model/notification/AtlasNotificationBaseMessage.java b/intg/src/main/java/org/apache/atlas/model/notification/AtlasNotificationBaseMessage.java
new file mode 100644
index 0000000..2411808
--- /dev/null
+++ b/intg/src/main/java/org/apache/atlas/model/notification/AtlasNotificationBaseMessage.java
@@ -0,0 +1,208 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.atlas.model.notification;
+
+
+import org.apache.atlas.AtlasConfiguration;
+import org.apache.commons.codec.binary.Base64;
+import org.apache.commons.codec.binary.StringUtils;
+import org.apache.commons.compress.utils.IOUtils;
+import org.codehaus.jackson.annotate.JsonAutoDetect;
+import org.codehaus.jackson.annotate.JsonIgnoreProperties;
+import org.codehaus.jackson.map.annotate.JsonSerialize;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.xml.bind.annotation.XmlAccessType;
+import javax.xml.bind.annotation.XmlAccessorType;
+import javax.xml.bind.annotation.XmlRootElement;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.util.zip.GZIPInputStream;
+import java.util.zip.GZIPOutputStream;
+
+import static org.codehaus.jackson.annotate.JsonAutoDetect.Visibility.NONE;
+import static org.codehaus.jackson.annotate.JsonAutoDetect.Visibility.PUBLIC_ONLY;
+
+
+@JsonAutoDetect(getterVisibility=PUBLIC_ONLY, setterVisibility=PUBLIC_ONLY, fieldVisibility=NONE)
+@JsonSerialize(include=JsonSerialize.Inclusion.NON_NULL)
+@JsonIgnoreProperties(ignoreUnknown=true)
+@XmlRootElement
+@XmlAccessorType(XmlAccessType.PROPERTY)
+public class AtlasNotificationBaseMessage {
+ private static final Logger LOG = LoggerFactory.getLogger(AtlasNotificationBaseMessage.class);
+
+ public static final int MESSAGE_MAX_LENGTH_BYTES = AtlasConfiguration.NOTIFICATION_MESSAGE_MAX_LENGTH_BYTES.getInt() - 512; // 512 bytes for envelop;
+ public static final boolean MESSAGE_COMPRESSION_ENABLED = AtlasConfiguration.NOTIFICATION_MESSAGE_COMPRESSION_ENABLED.getBoolean();
+
+ public enum CompressionKind { NONE, GZIP };
+
+ private MessageVersion version = null;
+ private String msgId = null;
+ private CompressionKind msgCompressionKind = CompressionKind.NONE;
+ private int msgSplitIdx = 1;
+ private int msgSplitCount = 1;
+
+
+ public AtlasNotificationBaseMessage() {
+ }
+
+ public AtlasNotificationBaseMessage(MessageVersion version) {
+ this(version, null, CompressionKind.NONE);
+ }
+
+ public AtlasNotificationBaseMessage(MessageVersion version, String msgId, CompressionKind msgCompressionKind) {
+ this.version = version;
+ this.msgId = msgId;
+ this.msgCompressionKind = msgCompressionKind;
+ }
+
+ public AtlasNotificationBaseMessage(MessageVersion version, String msgId, CompressionKind msgCompressionKind, int msgSplitIdx, int msgSplitCount) {
+ this.version = version;
+ this.msgId = msgId;
+ this.msgCompressionKind = msgCompressionKind;
+ this.msgSplitIdx = msgSplitIdx;
+ this.msgSplitCount = msgSplitCount;
+ }
+
+ public void setVersion(MessageVersion version) {
+ this.version = version;
+ }
+
+ public MessageVersion getVersion() {
+ return version;
+ }
+
+ public String getMsgId() {
+ return msgId;
+ }
+
+ public void setMsgId(String msgId) {
+ this.msgId = msgId;
+ }
+
+ public CompressionKind getMsgCompressionKind() {
+ return msgCompressionKind;
+ }
+
+ public void setMsgCompressed(CompressionKind msgCompressionKind) {
+ this.msgCompressionKind = msgCompressionKind;
+ }
+
+ public int getMsgSplitIdx() {
+ return msgSplitIdx;
+ }
+
+ public void setMsgSplitIdx(int msgSplitIdx) {
+ this.msgSplitIdx = msgSplitIdx;
+ }
+
+ public int getMsgSplitCount() {
+ return msgSplitCount;
+ }
+
+ public void setMsgSplitCount(int msgSplitCount) {
+ this.msgSplitCount = msgSplitCount;
+ }
+
+ /**
+ * Compare the version of this message with the given version.
+ *
+ * @param compareToVersion the version to compare to
+ *
+ * @return a negative integer, zero, or a positive integer as this message's version is less than, equal to,
+ * or greater than the given version.
+ */
+ public int compareVersion(MessageVersion compareToVersion) {
+ return version.compareTo(compareToVersion);
+ }
+
+
+ public static byte[] getBytesUtf8(String str) {
+ return StringUtils.getBytesUtf8(str);
+ }
+
+ public static String getStringUtf8(byte[] bytes) {
+ return StringUtils.newStringUtf8(bytes);
+ }
+
+ public static byte[] encodeBase64(byte[] bytes) {
+ return Base64.encodeBase64(bytes);
+ }
+
+ public static byte[] decodeBase64(byte[] bytes) {
+ return Base64.decodeBase64(bytes);
+ }
+
+ public static byte[] gzipCompressAndEncodeBase64(byte[] bytes) {
+ return encodeBase64(gzipCompress(bytes));
+ }
+
+ public static byte[] decodeBase64AndGzipUncompress(byte[] bytes) {
+ return gzipUncompress(decodeBase64(bytes));
+ }
+
+ public static String gzipCompress(String str) {
+ byte[] bytes = getBytesUtf8(str);
+ byte[] compressedBytes = gzipCompress(bytes);
+ byte[] encodedBytes = encodeBase64(compressedBytes);
+
+ return getStringUtf8(encodedBytes);
+ }
+
+ public static String gzipUncompress(String str) {
+ byte[] encodedBytes = getBytesUtf8(str);
+ byte[] compressedBytes = decodeBase64(encodedBytes);
+ byte[] bytes = gzipUncompress(compressedBytes);
+
+ return getStringUtf8(bytes);
+ }
+
+ public static byte[] gzipCompress(byte[] content) {
+ ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
+
+ try {
+ GZIPOutputStream gzipOutputStream = new GZIPOutputStream(byteArrayOutputStream);
+
+ gzipOutputStream.write(content);
+ gzipOutputStream.close();
+ } catch (IOException e) {
+ LOG.error("gzipCompress(): error compressing {} bytes", content.length, e);
+
+ throw new RuntimeException(e);
+ }
+
+ return byteArrayOutputStream.toByteArray();
+ }
+
+ public static byte[] gzipUncompress(byte[] content) {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+
+ try {
+ IOUtils.copy(new GZIPInputStream(new ByteArrayInputStream(content)), out);
+ } catch (IOException e) {
+ LOG.error("gzipUncompress(): error uncompressing {} bytes", content.length, e);
+ }
+
+ return out.toByteArray();
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/intg/src/main/java/org/apache/atlas/model/notification/AtlasNotificationMessage.java
----------------------------------------------------------------------
diff --git a/intg/src/main/java/org/apache/atlas/model/notification/AtlasNotificationMessage.java b/intg/src/main/java/org/apache/atlas/model/notification/AtlasNotificationMessage.java
new file mode 100644
index 0000000..5a5b63f
--- /dev/null
+++ b/intg/src/main/java/org/apache/atlas/model/notification/AtlasNotificationMessage.java
@@ -0,0 +1,101 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.atlas.model.notification;
+
+import org.codehaus.jackson.annotate.JsonAutoDetect;
+import org.codehaus.jackson.annotate.JsonIgnoreProperties;
+import org.codehaus.jackson.map.annotate.JsonSerialize;
+
+import javax.xml.bind.annotation.XmlAccessType;
+import javax.xml.bind.annotation.XmlAccessorType;
+import javax.xml.bind.annotation.XmlRootElement;
+import java.util.Date;
+
+import static org.codehaus.jackson.annotate.JsonAutoDetect.Visibility.NONE;
+import static org.codehaus.jackson.annotate.JsonAutoDetect.Visibility.PUBLIC_ONLY;
+
+/**
+ * Represents a notification message that is associated with a version.
+ */
+@JsonAutoDetect(getterVisibility=PUBLIC_ONLY, setterVisibility=PUBLIC_ONLY, fieldVisibility=NONE)
+@JsonSerialize(include=JsonSerialize.Inclusion.NON_NULL)
+@JsonIgnoreProperties(ignoreUnknown=true)
+@XmlRootElement
+@XmlAccessorType(XmlAccessType.PROPERTY)
+public class AtlasNotificationMessage<T> extends AtlasNotificationBaseMessage {
+ private String msgSourceIP;
+ private String msgCreatedBy;
+ private long msgCreationTime;
+
+ /**
+ * The actual message.
+ */
+ private T message;
+
+
+ // ----- Constructors ----------------------------------------------------
+ public AtlasNotificationMessage() {
+ }
+
+ public AtlasNotificationMessage(MessageVersion version, T message) {
+ this(version, message, null, null);
+ }
+
+ public AtlasNotificationMessage(MessageVersion version, T message, String msgSourceIP, String createdBy) {
+ super(version);
+
+ this.msgSourceIP = msgSourceIP;
+ this.msgCreatedBy = createdBy;
+ this.msgCreationTime = (new Date()).getTime();
+ this.message = message;
+ }
+
+
+ public String getMsgSourceIP() {
+ return msgSourceIP;
+ }
+
+ public void setMsgSourceIP(String msgSourceIP) {
+ this.msgSourceIP = msgSourceIP;
+ }
+
+ public String getMsgCreatedBy() {
+ return msgCreatedBy;
+ }
+
+ public void setMsgCreatedBy(String msgCreatedBy) {
+ this.msgCreatedBy = msgCreatedBy;
+ }
+
+ public long getMsgCreationTime() {
+ return msgCreationTime;
+ }
+
+ public void setMsgCreationTime(long msgCreationTime) {
+ this.msgCreationTime = msgCreationTime;
+ }
+
+ public T getMessage() {
+ return message;
+ }
+
+ public void setMessage(T message) {
+ this.message = message;
+ }
+}
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/intg/src/main/java/org/apache/atlas/model/notification/AtlasNotificationStringMessage.java
----------------------------------------------------------------------
diff --git a/intg/src/main/java/org/apache/atlas/model/notification/AtlasNotificationStringMessage.java b/intg/src/main/java/org/apache/atlas/model/notification/AtlasNotificationStringMessage.java
new file mode 100644
index 0000000..9064b6c
--- /dev/null
+++ b/intg/src/main/java/org/apache/atlas/model/notification/AtlasNotificationStringMessage.java
@@ -0,0 +1,82 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.atlas.model.notification;
+
+
+import org.codehaus.jackson.annotate.JsonAutoDetect;
+import org.codehaus.jackson.annotate.JsonIgnoreProperties;
+import org.codehaus.jackson.map.annotate.JsonSerialize;
+
+import javax.xml.bind.annotation.XmlAccessType;
+import javax.xml.bind.annotation.XmlAccessorType;
+import javax.xml.bind.annotation.XmlRootElement;
+
+import static org.codehaus.jackson.annotate.JsonAutoDetect.Visibility.NONE;
+import static org.codehaus.jackson.annotate.JsonAutoDetect.Visibility.PUBLIC_ONLY;
+
+@JsonAutoDetect(getterVisibility=PUBLIC_ONLY, setterVisibility=PUBLIC_ONLY, fieldVisibility=NONE)
+@JsonSerialize(include=JsonSerialize.Inclusion.NON_NULL)
+@JsonIgnoreProperties(ignoreUnknown=true)
+@XmlRootElement
+@XmlAccessorType(XmlAccessType.PROPERTY)
+public class AtlasNotificationStringMessage extends AtlasNotificationBaseMessage {
+ private String message = null;
+
+ public AtlasNotificationStringMessage() {
+ super(MessageVersion.CURRENT_VERSION);
+ }
+
+ public AtlasNotificationStringMessage(String message) {
+ super(MessageVersion.CURRENT_VERSION);
+
+ this.message = message;
+ }
+
+ public AtlasNotificationStringMessage(String message, String msgId, CompressionKind compressionKind) {
+ super(MessageVersion.CURRENT_VERSION, msgId, compressionKind);
+
+ this.message = message;
+ }
+
+ public AtlasNotificationStringMessage(String message, String msgId, CompressionKind compressionKind, int msgSplitIdx, int msgSplitCount) {
+ super(MessageVersion.CURRENT_VERSION, msgId, compressionKind, msgSplitIdx, msgSplitCount);
+
+ this.message = message;
+ }
+
+ public AtlasNotificationStringMessage(byte[] encodedBytes, String msgId, CompressionKind compressionKind) {
+ super(MessageVersion.CURRENT_VERSION, msgId, compressionKind);
+
+ this.message = AtlasNotificationBaseMessage.getStringUtf8(encodedBytes);
+ }
+
+ public AtlasNotificationStringMessage(byte[] encodedBytes, int offset, int length, String msgId, CompressionKind compressionKind, int msgSplitIdx, int msgSplitCount) {
+ super(MessageVersion.CURRENT_VERSION, msgId, compressionKind, msgSplitIdx, msgSplitCount);
+
+ this.message = new String(encodedBytes, offset, length);
+ }
+
+ public void setMessage(String message) {
+ this.message = message;
+ }
+
+ public String getMessage() {
+ return message;
+ }
+}
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/intg/src/main/java/org/apache/atlas/model/notification/MessageVersion.java
----------------------------------------------------------------------
diff --git a/intg/src/main/java/org/apache/atlas/model/notification/MessageVersion.java b/intg/src/main/java/org/apache/atlas/model/notification/MessageVersion.java
new file mode 100644
index 0000000..1dafa94
--- /dev/null
+++ b/intg/src/main/java/org/apache/atlas/model/notification/MessageVersion.java
@@ -0,0 +1,170 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.atlas.model.notification;
+
+import org.codehaus.jackson.annotate.JsonAutoDetect;
+import org.codehaus.jackson.annotate.JsonIgnoreProperties;
+import org.codehaus.jackson.map.annotate.JsonSerialize;
+
+import javax.xml.bind.annotation.XmlAccessType;
+import javax.xml.bind.annotation.XmlAccessorType;
+import javax.xml.bind.annotation.XmlRootElement;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Arrays;
+
+import static org.codehaus.jackson.annotate.JsonAutoDetect.Visibility.NONE;
+import static org.codehaus.jackson.annotate.JsonAutoDetect.Visibility.PUBLIC_ONLY;
+
+/**
+ * Represents the version of a notification message.
+ */
+@JsonAutoDetect(getterVisibility=PUBLIC_ONLY, setterVisibility=PUBLIC_ONLY, fieldVisibility=NONE)
+@JsonSerialize(include=JsonSerialize.Inclusion.NON_NULL)
+@JsonIgnoreProperties(ignoreUnknown=true)
+@XmlRootElement
+@XmlAccessorType(XmlAccessType.PROPERTY)
+public class MessageVersion implements Comparable<MessageVersion>, Serializable {
+ private static final long serialVersionUID = 1L;
+
+ /**
+ * Used for message with no version (old format).
+ */
+ public static final MessageVersion NO_VERSION = new MessageVersion("0");
+ public static final MessageVersion VERSION_1 = new MessageVersion("1.0.0");
+
+ public static final MessageVersion CURRENT_VERSION = VERSION_1;
+
+ private String version;
+
+
+ // ----- Constructors ----------------------------------------------------
+ public MessageVersion() {
+ this.version = CURRENT_VERSION.version;
+ }
+
+ /**
+ * Create a message version.
+ *
+ * @param version the version string
+ */
+ public MessageVersion(String version) {
+ this.version = version;
+
+ try {
+ getVersionParts();
+ } catch (NumberFormatException e) {
+ throw new IllegalArgumentException(String.format("Invalid version string : %s.", version), e);
+ }
+ }
+
+ public String getVersion() {
+ return version;
+ }
+
+ public void setVersion(String version) {
+ this.version = version;
+ }
+
+
+ // ----- Comparable ------------------------------------------------------
+
+ @Override
+ public int compareTo(MessageVersion that) {
+ if (that == null) {
+ return 1;
+ }
+
+ Integer[] thisParts = getVersionParts();
+ Integer[] thatParts = that.getVersionParts();
+
+ int length = Math.max(thisParts.length, thatParts.length);
+
+ for (int i = 0; i < length; i++) {
+
+ int comp = getVersionPart(thisParts, i) - getVersionPart(thatParts, i);
+
+ if (comp != 0) {
+ return comp;
+ }
+ }
+ return 0;
+ }
+
+
+ // ----- Object overrides ------------------------------------------------
+
+ @Override
+ public boolean equals(Object that) {
+ if (this == that){
+ return true;
+ }
+
+ if (that == null || getClass() != that.getClass()) {
+ return false;
+ }
+
+ return compareTo((MessageVersion) that) == 0;
+ }
+
+ @Override
+ public int hashCode() {
+ return Arrays.hashCode(getVersionParts());
+ }
+
+
+ @Override
+ public String toString() {
+ return "MessageVersion[version=" + version + "]";
+ }
+
+ // ----- helper methods --------------------------------------------------
+
+ /**
+ * Get the version parts array by splitting the version string.
+ * Strip the trailing zeros (i.e. '1.0.0' equals '1').
+ *
+ * @return the version parts array
+ */
+ public Integer[] getVersionParts() {
+
+ String[] sParts = version.split("\\.");
+ ArrayList<Integer> iParts = new ArrayList<>();
+ int trailingZeros = 0;
+
+ for (String sPart : sParts) {
+ Integer iPart = new Integer(sPart);
+
+ if (iPart == 0) {
+ ++trailingZeros;
+ } else {
+ for (int i = 0; i < trailingZeros; ++i) {
+ iParts.add(0);
+ }
+ trailingZeros = 0;
+ iParts.add(iPart);
+ }
+ }
+ return iParts.toArray(new Integer[iParts.size()]);
+ }
+
+ public Integer getVersionPart(Integer[] versionParts, int i) {
+ return i < versionParts.length ? versionParts[i] : 0;
+ }
+}
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/intg/src/main/java/org/apache/atlas/type/AtlasType.java
----------------------------------------------------------------------
diff --git a/intg/src/main/java/org/apache/atlas/type/AtlasType.java b/intg/src/main/java/org/apache/atlas/type/AtlasType.java
index 551ee21..1b09b93 100644
--- a/intg/src/main/java/org/apache/atlas/type/AtlasType.java
+++ b/intg/src/main/java/org/apache/atlas/type/AtlasType.java
@@ -21,10 +21,19 @@ package org.apache.atlas.type;
import org.apache.atlas.exception.AtlasBaseException;
import org.apache.atlas.model.TypeCategory;
import org.apache.atlas.model.typedef.AtlasBaseTypeDef;
+import org.apache.atlas.v1.model.notification.HookNotification.EntityCreateRequest;
+import org.apache.atlas.v1.model.notification.HookNotification.EntityDeleteRequest;
+import org.apache.atlas.v1.model.notification.HookNotification.EntityPartialUpdateRequest;
+import org.apache.atlas.v1.model.notification.HookNotification.EntityUpdateRequest;
+import org.apache.atlas.v1.model.notification.HookNotification.HookNotificationMessage;
+import org.apache.atlas.v1.model.notification.HookNotification.HookNotificationType;
+import org.apache.atlas.v1.model.notification.HookNotification.TypeRequest;
import org.apache.atlas.v1.model.typedef.Multiplicity;
import org.codehaus.jackson.*;
import org.codehaus.jackson.map.*;
import org.codehaus.jackson.map.module.SimpleModule;
+import org.codehaus.jackson.node.ObjectNode;
+import org.codehaus.jackson.type.TypeReference;
import java.io.IOException;
import java.text.ParseException;
@@ -32,8 +41,6 @@ import java.util.Date;
import java.util.List;
-
-
/**
* base class that declares interface for all Atlas types.
*/
@@ -53,6 +60,7 @@ public abstract class AtlasType {
atlasSerDeModule.addDeserializer(Date.class, new DateDeserializer());
atlasSerDeModule.addSerializer(Multiplicity.class, new MultiplicitySerializer());
atlasSerDeModule.addDeserializer(Multiplicity.class, new MultiplicityDeserializer());
+ atlasSerDeModule.addDeserializer(HookNotificationMessage.class, new HookMessageDeserializer());
mapperV1.registerModule(atlasSerDeModule);
}
@@ -148,6 +156,7 @@ public abstract class AtlasType {
try {
ret = mapperV1.writeValueAsString(obj);
}catch (IOException e){
+ e.printStackTrace(System.out);
ret = null;
}
return ret;
@@ -163,6 +172,16 @@ public abstract class AtlasType {
return ret;
}
+ public static <T> T fromV1Json(String jsonStr, TypeReference<T> type) {
+ T ret;
+ try {
+ ret = mapperV1.readValue(jsonStr, type);
+ }catch (IOException e){
+ ret = null;
+ }
+ return ret;
+ }
+
static class DateSerializer extends JsonSerializer<Date> {
@Override
public void serialize(Date value, JsonGenerator jgen, SerializerProvider provider) throws IOException {
@@ -233,4 +252,43 @@ public abstract class AtlasType {
return ret;
}
}
+
+ static class HookMessageDeserializer extends JsonDeserializer<HookNotificationMessage> {
+ @Override
+ public HookNotificationMessage deserialize(JsonParser parser, DeserializationContext context) throws IOException {
+ HookNotificationMessage ret = null;
+
+ ObjectMapper mapper = (ObjectMapper) parser.getCodec();
+ ObjectNode root = (ObjectNode) mapper.readTree(parser);
+
+ JsonNode typeNode = root.get("type");
+ String strType = typeNode.asText();
+ HookNotificationType notificationType = HookNotificationType.valueOf(strType);
+
+ switch (notificationType) {
+ case TYPE_CREATE:
+ case TYPE_UPDATE:
+ ret = mapper.readValue(root, TypeRequest.class);
+ break;
+
+ case ENTITY_CREATE:
+ ret = mapper.readValue(root, EntityCreateRequest.class);
+ break;
+
+ case ENTITY_PARTIAL_UPDATE:
+ ret = mapper.readValue(root, EntityPartialUpdateRequest.class);
+ break;
+
+ case ENTITY_FULL_UPDATE:
+ ret = mapper.readValue(root, EntityUpdateRequest.class);
+ break;
+
+ case ENTITY_DELETE:
+ ret = mapper.readValue(root, EntityDeleteRequest.class);
+ break;
+ }
+
+ return ret;
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/intg/src/main/java/org/apache/atlas/typesystem/types/DataTypes.java
----------------------------------------------------------------------
diff --git a/intg/src/main/java/org/apache/atlas/typesystem/types/DataTypes.java b/intg/src/main/java/org/apache/atlas/typesystem/types/DataTypes.java
index 804ad33..dba2d88 100644
--- a/intg/src/main/java/org/apache/atlas/typesystem/types/DataTypes.java
+++ b/intg/src/main/java/org/apache/atlas/typesystem/types/DataTypes.java
@@ -18,7 +18,9 @@
package org.apache.atlas.typesystem.types;
-
+/*
+ * this enum must be in package org.apache.atlas.typesystem.types, since vertex property in GraphDB has reference to this type
+ */
public class DataTypes {
public enum TypeCategory {
PRIMITIVE,
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/intg/src/main/java/org/apache/atlas/v1/model/instance/AtlasSystemAttributes.java
----------------------------------------------------------------------
diff --git a/intg/src/main/java/org/apache/atlas/v1/model/instance/AtlasSystemAttributes.java b/intg/src/main/java/org/apache/atlas/v1/model/instance/AtlasSystemAttributes.java
index 43eca0b..fba22cc 100644
--- a/intg/src/main/java/org/apache/atlas/v1/model/instance/AtlasSystemAttributes.java
+++ b/intg/src/main/java/org/apache/atlas/v1/model/instance/AtlasSystemAttributes.java
@@ -120,4 +120,25 @@ public class AtlasSystemAttributes implements Serializable {
public int hashCode() {
return Objects.hash(createdBy, modifiedBy, createdTime, modifiedTime);
}
+
+
+ @Override
+ public String toString() {
+ return toString(new StringBuilder()).toString();
+ }
+
+ public StringBuilder toString(StringBuilder sb) {
+ if (sb == null) {
+ sb = new StringBuilder();
+ }
+
+ sb.append("AtlasSystemAttributes{")
+ .append("createdBy=").append(createdBy)
+ .append(", modifiedBy=").append(modifiedBy)
+ .append(", createdTime=").append(createdTime)
+ .append(", modifiedTime=").append(modifiedTime)
+ .append("}");
+
+ return sb;
+ }
}
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/intg/src/main/java/org/apache/atlas/v1/model/instance/Id.java
----------------------------------------------------------------------
diff --git a/intg/src/main/java/org/apache/atlas/v1/model/instance/Id.java b/intg/src/main/java/org/apache/atlas/v1/model/instance/Id.java
index 1b250f0..cd2951e 100644
--- a/intg/src/main/java/org/apache/atlas/v1/model/instance/Id.java
+++ b/intg/src/main/java/org/apache/atlas/v1/model/instance/Id.java
@@ -156,6 +156,25 @@ public class Id implements Serializable {
}
+ @Override
+ public String toString() {
+ return toString(new StringBuilder()).toString();
+ }
+
+ public StringBuilder toString(StringBuilder sb) {
+ if (sb == null) {
+ sb = new StringBuilder();
+ }
+
+ sb.append("Id{")
+ .append("id=").append(id)
+ .append(", typeName=").append(typeName)
+ .append(", version=").append(version)
+ .append(", state=").append(state)
+ .append("}");
+
+ return sb;
+ }
private static long nextNegativeLong() {
long ret = s_nextId.getAndDecrement();
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/intg/src/main/java/org/apache/atlas/v1/model/instance/Referenceable.java
----------------------------------------------------------------------
diff --git a/intg/src/main/java/org/apache/atlas/v1/model/instance/Referenceable.java b/intg/src/main/java/org/apache/atlas/v1/model/instance/Referenceable.java
index 44b7a5c..158da45 100644
--- a/intg/src/main/java/org/apache/atlas/v1/model/instance/Referenceable.java
+++ b/intg/src/main/java/org/apache/atlas/v1/model/instance/Referenceable.java
@@ -20,6 +20,7 @@ package org.apache.atlas.v1.model.instance;
+import org.apache.atlas.model.typedef.AtlasBaseTypeDef;
import org.codehaus.jackson.annotate.JsonAutoDetect;
import org.codehaus.jackson.annotate.JsonIgnore;
import org.codehaus.jackson.annotate.JsonIgnoreProperties;
@@ -198,4 +199,30 @@ public class Referenceable extends Struct implements Serializable {
public int hashCode() {
return Objects.hash(id, traits, traitNames, systemAttributes);
}
+
+
+ @Override
+ public StringBuilder toString(StringBuilder sb) {
+ if (sb == null) {
+ sb = new StringBuilder();
+ }
+
+ sb.append("Referenceable{");
+ super.toString(sb);
+ sb.append(", id=");
+ if (id != null) {
+ id.toString(sb);
+ }
+ sb.append(", triats={");
+ AtlasBaseTypeDef.dumpObjects(this.traits, sb);
+ sb.append("}, traitNames=[");
+ AtlasBaseTypeDef.dumpObjects(traitNames, sb);
+ sb.append("], systemAttributes=");
+ if (systemAttributes != null) {
+ systemAttributes.toString(sb);
+ }
+ sb.append("}");
+
+ return sb;
+ }
}
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/intg/src/main/java/org/apache/atlas/v1/model/instance/Struct.java
----------------------------------------------------------------------
diff --git a/intg/src/main/java/org/apache/atlas/v1/model/instance/Struct.java b/intg/src/main/java/org/apache/atlas/v1/model/instance/Struct.java
index 53e00ca..5fa4080 100644
--- a/intg/src/main/java/org/apache/atlas/v1/model/instance/Struct.java
+++ b/intg/src/main/java/org/apache/atlas/v1/model/instance/Struct.java
@@ -19,6 +19,7 @@
package org.apache.atlas.v1.model.instance;
+import org.apache.atlas.model.typedef.AtlasBaseTypeDef;
import org.codehaus.jackson.annotate.JsonAutoDetect;
import org.codehaus.jackson.annotate.JsonIgnore;
import org.codehaus.jackson.annotate.JsonIgnoreProperties;
@@ -138,4 +139,24 @@ public class Struct implements Serializable {
public int hashCode() {
return Objects.hash(typeName, values);
}
+
+ @Override
+ public String toString() {
+ return toString(new StringBuilder()).toString();
+ }
+
+ public StringBuilder toString(StringBuilder sb) {
+ if (sb == null) {
+ sb = new StringBuilder();
+ }
+
+ sb.append("Struct{");
+ sb.append("typeName=").append(typeName);
+ sb.append(", values={");
+ AtlasBaseTypeDef.dumpObjects(values, sb);
+ sb.append("}");
+ sb.append("}");
+
+ return sb;
+ }
}
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/intg/src/main/java/org/apache/atlas/v1/model/notification/EntityNotification.java
----------------------------------------------------------------------
diff --git a/intg/src/main/java/org/apache/atlas/v1/model/notification/EntityNotification.java b/intg/src/main/java/org/apache/atlas/v1/model/notification/EntityNotification.java
new file mode 100644
index 0000000..e4305dd
--- /dev/null
+++ b/intg/src/main/java/org/apache/atlas/v1/model/notification/EntityNotification.java
@@ -0,0 +1,218 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.atlas.v1.model.notification;
+
+import org.apache.atlas.model.typedef.AtlasBaseTypeDef;
+import org.apache.atlas.v1.model.instance.Referenceable;
+import org.apache.atlas.v1.model.instance.Struct;
+import org.apache.atlas.type.AtlasClassificationType;
+import org.apache.atlas.type.AtlasTypeRegistry;
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.collections.MapUtils;
+import org.codehaus.jackson.annotate.JsonAutoDetect;
+import org.codehaus.jackson.annotate.JsonIgnore;
+import org.codehaus.jackson.annotate.JsonIgnoreProperties;
+import org.codehaus.jackson.map.annotate.JsonSerialize;
+
+import javax.xml.bind.annotation.XmlAccessType;
+import javax.xml.bind.annotation.XmlAccessorType;
+import javax.xml.bind.annotation.XmlRootElement;
+import java.io.Serializable;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+import static org.codehaus.jackson.annotate.JsonAutoDetect.Visibility.NONE;
+import static org.codehaus.jackson.annotate.JsonAutoDetect.Visibility.PUBLIC_ONLY;
+
+/**
+ * Entity notification
+ */
+@JsonAutoDetect(getterVisibility=PUBLIC_ONLY, setterVisibility=PUBLIC_ONLY, fieldVisibility=NONE)
+@JsonSerialize(include=JsonSerialize.Inclusion.NON_NULL)
+@JsonIgnoreProperties(ignoreUnknown=true)
+@XmlRootElement
+@XmlAccessorType(XmlAccessType.PROPERTY)
+public class EntityNotification implements Serializable {
+ private static final long serialVersionUID = 1L;
+
+ public enum OperationType {
+ ENTITY_CREATE,
+ ENTITY_UPDATE,
+ ENTITY_DELETE,
+ TRAIT_ADD,
+ TRAIT_DELETE,
+ TRAIT_UPDATE
+ }
+
+ private Referenceable entity;
+ private OperationType operationType;
+ private List<Struct> traits;
+
+
+ // ----- Constructors ------------------------------------------------------
+
+ /**
+ * No-arg constructor for serialization.
+ */
+ public EntityNotification() {
+ }
+
+ /**
+ * Construct an EntityNotification.
+ *
+ * @param entity the entity subject of the notification
+ * @param operationType the type of operation that caused the notification
+ * @param traits the traits for the given entity
+ */
+ public EntityNotification(Referenceable entity, OperationType operationType, List<Struct> traits) {
+ this.entity = entity;
+ this.operationType = operationType;
+ this.traits = traits;
+ }
+
+ /**
+ * Construct an EntityNotification.
+ *
+ * @param entity the entity subject of the notification
+ * @param operationType the type of operation that caused the notification
+ * @param typeRegistry the Atlas type system
+ */
+ public EntityNotification(Referenceable entity, OperationType operationType, AtlasTypeRegistry typeRegistry) {
+ this(entity, operationType, getAllTraits(entity, typeRegistry));
+ }
+
+ public Referenceable getEntity() {
+ return entity;
+ }
+
+ public void setEntity(Referenceable entity) {
+ this.entity = entity;
+ }
+
+ public OperationType getOperationType() {
+ return operationType;
+ }
+
+ public void setOperationType(OperationType operationType) {
+ this.operationType = operationType;
+ }
+
+ public List<Struct> getTraits() {
+ return traits;
+ }
+
+ public void setTraits(List<Struct> traits) {
+ this.traits = traits;
+ }
+
+ @JsonIgnore
+ public List<Struct> getAllTraits() {
+ return traits;
+ }
+
+
+ // ----- Object overrides --------------------------------------------------
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ EntityNotification that = (EntityNotification) o;
+ return Objects.equals(entity, that.entity) &&
+ operationType == that.operationType &&
+ Objects.equals(traits, that.traits);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(entity, operationType, traits);
+ }
+
+ @Override
+ public String toString() {
+ return toString(new StringBuilder()).toString();
+ }
+
+ public StringBuilder toString(StringBuilder sb) {
+ if (sb == null) {
+ sb = new StringBuilder();
+ }
+
+ sb.append("EntityNotification{");
+ sb.append("entity=");
+ if (entity != null) {
+ entity.toString(sb);
+ } else {
+ sb.append(entity);
+ }
+ sb.append(", operationType=").append(operationType);
+ sb.append(", traits=[");
+ AtlasBaseTypeDef.dumpObjects(traits, sb);
+ sb.append("]");
+ sb.append("}");
+
+ return sb;
+ }
+
+
+ // ----- helper methods ----------------------------------------------------
+
+ private static List<Struct> getAllTraits(Referenceable entityDefinition, AtlasTypeRegistry typeRegistry) {
+ List<Struct> ret = new LinkedList<>();
+
+ for (String traitName : entityDefinition.getTraitNames()) {
+ Struct trait = entityDefinition.getTrait(traitName);
+ AtlasClassificationType traitType = typeRegistry.getClassificationTypeByName(traitName);
+ Set<String> superTypeNames = traitType != null ? traitType.getAllSuperTypes() : null;
+
+ ret.add(trait);
+
+ if (CollectionUtils.isNotEmpty(superTypeNames)) {
+ for (String superTypeName : superTypeNames) {
+ Struct superTypeTrait = new Struct(superTypeName);
+
+ if (MapUtils.isNotEmpty(trait.getValues())) {
+ AtlasClassificationType superType = typeRegistry.getClassificationTypeByName(superTypeName);
+
+ if (superType != null && MapUtils.isNotEmpty(superType.getAllAttributes())) {
+ Map<String, Object> superTypeTraitAttributes = new HashMap<>();
+
+ for (Map.Entry<String, Object> attrEntry : trait.getValues().entrySet()) {
+ String attrName = attrEntry.getKey();
+
+ if (superType.getAllAttributes().containsKey(attrName)) {
+ superTypeTraitAttributes.put(attrName, attrEntry.getValue());
+ }
+ }
+
+ superTypeTrait.setValues(superTypeTraitAttributes);
+ }
+ }
+
+ ret.add(superTypeTrait);
+ }
+ }
+ }
+
+ return ret;
+ }
+}
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/intg/src/main/java/org/apache/atlas/v1/model/notification/HookNotification.java
----------------------------------------------------------------------
diff --git a/intg/src/main/java/org/apache/atlas/v1/model/notification/HookNotification.java b/intg/src/main/java/org/apache/atlas/v1/model/notification/HookNotification.java
new file mode 100644
index 0000000..7be5e0b
--- /dev/null
+++ b/intg/src/main/java/org/apache/atlas/v1/model/notification/HookNotification.java
@@ -0,0 +1,415 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.atlas.v1.model.notification;
+
+import com.google.gson.JsonParseException;
+import org.apache.atlas.model.typedef.AtlasBaseTypeDef;
+import org.apache.atlas.v1.model.instance.Referenceable;
+import org.apache.atlas.v1.model.typedef.TypesDef;
+import org.apache.atlas.type.AtlasType;
+import org.apache.commons.lang.StringUtils;
+import org.codehaus.jackson.annotate.JsonAutoDetect;
+import org.codehaus.jackson.annotate.JsonIgnoreProperties;
+import org.codehaus.jackson.map.annotate.JsonSerialize;
+import org.codehaus.jettison.json.JSONArray;
+import org.codehaus.jettison.json.JSONException;
+
+import javax.xml.bind.annotation.XmlAccessType;
+import javax.xml.bind.annotation.XmlAccessorType;
+import javax.xml.bind.annotation.XmlRootElement;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.codehaus.jackson.annotate.JsonAutoDetect.Visibility.NONE;
+import static org.codehaus.jackson.annotate.JsonAutoDetect.Visibility.PUBLIC_ONLY;
+
+/**
+ * Contains the structure of messages transferred from hooks to atlas.
+ */
+public class HookNotification {
+ /**
+ * Type of the hook message.
+ */
+ public enum HookNotificationType {
+ TYPE_CREATE, TYPE_UPDATE, ENTITY_CREATE, ENTITY_PARTIAL_UPDATE, ENTITY_FULL_UPDATE, ENTITY_DELETE
+ }
+
+ /**
+ * Base type of hook message.
+ */
+ @JsonAutoDetect(getterVisibility=PUBLIC_ONLY, setterVisibility=PUBLIC_ONLY, fieldVisibility=NONE)
+ @JsonSerialize(include=JsonSerialize.Inclusion.NON_NULL)
+ @JsonIgnoreProperties(ignoreUnknown=true)
+ @XmlRootElement
+ @XmlAccessorType(XmlAccessType.PROPERTY)
+ public static class HookNotificationMessage implements Serializable {
+ private static final long serialVersionUID = 1L;
+
+ public static final String UNKNOW_USER = "UNKNOWN";
+
+ protected HookNotificationType type;
+ protected String user;
+
+ public HookNotificationMessage() {
+ }
+
+ public HookNotificationMessage(HookNotificationType type, String user) {
+ this.type = type;
+ this.user = user;
+ }
+
+ public HookNotificationType getType() {
+ return type;
+ }
+
+ public void setType(HookNotificationType type) {
+ this.type = type;
+ }
+
+ public String getUser() {
+ if (StringUtils.isEmpty(user)) {
+ return UNKNOW_USER;
+ }
+
+ return user;
+ }
+
+ public void setUser(String user) {
+ this.user = user;
+ }
+
+ @Override
+ public String toString() {
+ return toString(new StringBuilder()).toString();
+ }
+
+ public StringBuilder toString(StringBuilder sb) {
+ if (sb == null) {
+ sb = new StringBuilder();
+ }
+
+ sb.append("HookNotificationMessage{");
+ sb.append("type=").append(type);
+ sb.append(", user=").append(user);
+ sb.append("}");
+
+ return sb;
+ }
+ }
+
+ /**
+ * Hook message for create type definitions.
+ */
+ @JsonAutoDetect(getterVisibility=PUBLIC_ONLY, setterVisibility=PUBLIC_ONLY, fieldVisibility=NONE)
+ @JsonSerialize(include=JsonSerialize.Inclusion.NON_NULL)
+ @JsonIgnoreProperties(ignoreUnknown=true)
+ @XmlRootElement
+ @XmlAccessorType(XmlAccessType.PROPERTY)
+ public static class TypeRequest extends HookNotificationMessage implements Serializable {
+ private static final long serialVersionUID = 1L;
+
+ private TypesDef typesDef;
+
+ public TypeRequest() {
+ }
+
+ public TypeRequest(HookNotificationType type, TypesDef typesDef, String user) {
+ super(type, user);
+ this.typesDef = typesDef;
+ }
+
+ public TypesDef getTypesDef() {
+ return typesDef;
+ }
+
+ public void setTypesDef(TypesDef typesDef) {
+ this.typesDef = typesDef;
+ }
+
+ @Override
+ public StringBuilder toString(StringBuilder sb) {
+ if (sb == null) {
+ sb = new StringBuilder();
+ }
+
+ sb.append("TypeRequest{");
+ super.toString(sb);
+ sb.append("typesDef=");
+ if (typesDef != null) {
+ typesDef.toString(sb);
+ }
+ sb.append("}");
+
+ return sb;
+ }
+ }
+
+ /**
+ * Hook message for creating new entities.
+ */
+ @JsonAutoDetect(getterVisibility=PUBLIC_ONLY, setterVisibility=PUBLIC_ONLY, fieldVisibility=NONE)
+ @JsonSerialize(include=JsonSerialize.Inclusion.NON_NULL)
+ @JsonIgnoreProperties(ignoreUnknown=true)
+ @XmlRootElement
+ @XmlAccessorType(XmlAccessType.PROPERTY)
+ public static class EntityCreateRequest extends HookNotificationMessage implements Serializable {
+ private static final long serialVersionUID = 1L;
+
+ private List<Referenceable> entities;
+
+ public EntityCreateRequest() {
+ }
+
+ public EntityCreateRequest(String user, Referenceable... entities) {
+ this(HookNotificationType.ENTITY_CREATE, Arrays.asList(entities), user);
+ }
+
+ public EntityCreateRequest(String user, List<Referenceable> entities) {
+ this(HookNotificationType.ENTITY_CREATE, entities, user);
+ }
+
+ protected EntityCreateRequest(HookNotificationType type, List<Referenceable> entities, String user) {
+ super(type, user);
+ this.entities = entities;
+ }
+
+ public EntityCreateRequest(String user, JSONArray jsonArray) {
+ super(HookNotificationType.ENTITY_CREATE, user);
+ entities = new ArrayList<>();
+ for (int index = 0; index < jsonArray.length(); index++) {
+ try {
+ entities.add(AtlasType.fromV1Json(jsonArray.getString(index), Referenceable.class));
+ } catch (JSONException e) {
+ throw new JsonParseException(e);
+ }
+ }
+ }
+
+ public List<Referenceable> getEntities() {
+ return entities;
+ }
+
+ public void setEntities(List<Referenceable> entities) {
+ this.entities = entities;
+ }
+
+ @Override
+ public StringBuilder toString(StringBuilder sb) {
+ if (sb == null) {
+ sb = new StringBuilder();
+ }
+
+ sb.append("EntityCreateRequest{");
+ super.toString(sb);
+ sb.append("entities=[");
+ AtlasBaseTypeDef.dumpObjects(getEntities(), sb);
+ sb.append("]");
+ sb.append("}");
+
+ return sb;
+ }
+ }
+
+ /**
+ * Hook message for updating entities(full update).
+ */
+ @JsonAutoDetect(getterVisibility=PUBLIC_ONLY, setterVisibility=PUBLIC_ONLY, fieldVisibility=NONE)
+ @JsonSerialize(include=JsonSerialize.Inclusion.NON_NULL)
+ @JsonIgnoreProperties(ignoreUnknown=true)
+ @XmlRootElement
+ @XmlAccessorType(XmlAccessType.PROPERTY)
+ public static class EntityUpdateRequest extends EntityCreateRequest implements Serializable {
+ private static final long serialVersionUID = 1L;
+
+ public EntityUpdateRequest() {
+ }
+
+ public EntityUpdateRequest(String user, Referenceable... entities) {
+ this(user, Arrays.asList(entities));
+ }
+
+ public EntityUpdateRequest(String user, List<Referenceable> entities) {
+ super(HookNotificationType.ENTITY_FULL_UPDATE, entities, user);
+ }
+
+ @Override
+ public StringBuilder toString(StringBuilder sb) {
+ if (sb == null) {
+ sb = new StringBuilder();
+ }
+
+ sb.append("EntityUpdateRequest{");
+ super.toString(sb);
+ sb.append("entities=[");
+ AtlasBaseTypeDef.dumpObjects(getEntities(), sb);
+ sb.append("]");
+ sb.append("}");
+
+ return sb;
+ }
+ }
+
+ /**
+ * Hook message for updating entities(partial update).
+ */
+ public static class EntityPartialUpdateRequest extends HookNotificationMessage {
+ private static final long serialVersionUID = 1L;
+
+ private String typeName;
+ private String attribute;
+ private String attributeValue;
+ private Referenceable entity;
+
+ public EntityPartialUpdateRequest() {
+ }
+
+ public EntityPartialUpdateRequest(String user, String typeName, String attribute, String attributeValue, Referenceable entity) {
+ super(HookNotificationType.ENTITY_PARTIAL_UPDATE, user);
+
+ this.typeName = typeName;
+ this.attribute = attribute;
+ this.attributeValue = attributeValue;
+ this.entity = entity;
+ }
+
+ public String getTypeName() {
+ return typeName;
+ }
+
+ public void setTypeName(String typeName) {
+ this.typeName = typeName;
+ }
+
+ public String getAttribute() {
+ return attribute;
+ }
+
+ public void setAttribute(String attribute) {
+ this.attribute = attribute;
+ }
+
+ public String getAttributeValue() {
+ return attributeValue;
+ }
+
+ public void setAttributeValue(String attributeValue) {
+ this.attributeValue = attributeValue;
+ }
+
+ public Referenceable getEntity() {
+ return entity;
+ }
+
+ public void setEntity(Referenceable entity) {
+ this.entity = entity;
+ }
+
+ @Override
+ public StringBuilder toString(StringBuilder sb) {
+ if (sb == null) {
+ sb = new StringBuilder();
+ }
+
+ sb.append("EntityPartialUpdateRequest{");
+ super.toString(sb);
+ sb.append("typeName=").append(typeName);
+ sb.append("attribute=").append(attribute);
+ sb.append("attributeValue=").append(attributeValue);
+ sb.append("entity=");
+ if (entity != null) {
+ entity.toString(sb);
+ }
+ sb.append("}");
+
+ return sb;
+ }
+ }
+
+ /**
+ * Hook message for creating new entities.
+ */
+ @JsonAutoDetect(getterVisibility=PUBLIC_ONLY, setterVisibility=PUBLIC_ONLY, fieldVisibility=NONE)
+ @JsonSerialize(include=JsonSerialize.Inclusion.NON_NULL)
+ @JsonIgnoreProperties(ignoreUnknown=true)
+ @XmlRootElement
+ @XmlAccessorType(XmlAccessType.PROPERTY)
+ public static class EntityDeleteRequest extends HookNotificationMessage implements Serializable {
+ private static final long serialVersionUID = 1L;
+
+ private String typeName;
+ private String attribute;
+ private String attributeValue;
+
+ public EntityDeleteRequest() {
+ }
+
+ public EntityDeleteRequest(String user, String typeName, String attribute, String attributeValue) {
+ this(HookNotificationType.ENTITY_DELETE, user, typeName, attribute, attributeValue);
+ }
+
+ protected EntityDeleteRequest(HookNotificationType type, String user, String typeName, String attribute, String attributeValue) {
+ super(type, user);
+
+ this.typeName = typeName;
+ this.attribute = attribute;
+ this.attributeValue = attributeValue;
+ }
+
+ public String getTypeName() {
+ return typeName;
+ }
+
+ public void setTypeName(String typeName) {
+ this.typeName = typeName;
+ }
+
+ public String getAttribute() {
+ return attribute;
+ }
+
+ public void setAttribute(String attribute) {
+ this.attribute = attribute;
+ }
+
+ public String getAttributeValue() {
+ return attributeValue;
+ }
+
+ public void setAttributeValue(String attributeValue) {
+ this.attributeValue = attributeValue;
+ }
+
+ @Override
+ public StringBuilder toString(StringBuilder sb) {
+ if (sb == null) {
+ sb = new StringBuilder();
+ }
+
+ sb.append("EntityDeleteRequest{");
+ super.toString(sb);
+ sb.append("typeName=").append(typeName);
+ sb.append("attribute=").append(attribute);
+ sb.append("attributeValue=").append(attributeValue);
+ sb.append("}");
+
+ return sb;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/intg/src/main/java/org/apache/atlas/v1/model/typedef/TypesDef.java
----------------------------------------------------------------------
diff --git a/intg/src/main/java/org/apache/atlas/v1/model/typedef/TypesDef.java b/intg/src/main/java/org/apache/atlas/v1/model/typedef/TypesDef.java
index 1e67839..f8bcfa3 100644
--- a/intg/src/main/java/org/apache/atlas/v1/model/typedef/TypesDef.java
+++ b/intg/src/main/java/org/apache/atlas/v1/model/typedef/TypesDef.java
@@ -18,6 +18,7 @@
package org.apache.atlas.v1.model.typedef;
+import org.apache.atlas.model.typedef.AtlasBaseTypeDef;
import org.codehaus.jackson.annotate.JsonAutoDetect;
import org.codehaus.jackson.annotate.JsonIgnoreProperties;
import org.codehaus.jackson.map.annotate.JsonSerialize;
@@ -88,4 +89,30 @@ public class TypesDef implements Serializable {
public void setClassTypes(List<ClassTypeDefinition> classTypes) {
this.classTypes = classTypes;
}
+
+
+ @Override
+ public String toString() {
+ return toString(new StringBuilder()).toString();
+ }
+
+ public StringBuilder toString(StringBuilder sb) {
+ if (sb == null) {
+ sb = new StringBuilder();
+ }
+
+ sb.append("TypesDef{");
+ sb.append("enumTypes=[");
+ AtlasBaseTypeDef.dumpObjects(enumTypes, sb);
+ sb.append("], structTypes=[");
+ AtlasBaseTypeDef.dumpObjects(structTypes, sb);
+ sb.append("], traitTypes=[");
+ AtlasBaseTypeDef.dumpObjects(traitTypes, sb);
+ sb.append("], classTypes=[");
+ AtlasBaseTypeDef.dumpObjects(classTypes, sb);
+ sb.append("]");
+ sb.append("}");
+
+ return sb;
+ }
}
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/notification/src/main/java/org/apache/atlas/hook/AtlasHook.java
----------------------------------------------------------------------
diff --git a/notification/src/main/java/org/apache/atlas/hook/AtlasHook.java b/notification/src/main/java/org/apache/atlas/hook/AtlasHook.java
index f412217..a225f3c 100644
--- a/notification/src/main/java/org/apache/atlas/hook/AtlasHook.java
+++ b/notification/src/main/java/org/apache/atlas/hook/AtlasHook.java
@@ -24,7 +24,7 @@ import org.apache.atlas.kafka.NotificationProvider;
import org.apache.atlas.v1.model.instance.Referenceable;
import org.apache.atlas.notification.NotificationException;
import org.apache.atlas.notification.NotificationInterface;
-import org.apache.atlas.notification.hook.HookNotification;
+import org.apache.atlas.v1.model.notification.HookNotification;
import org.apache.atlas.security.InMemoryJAASConfiguration;
import org.apache.atlas.type.AtlasType;
import org.apache.commons.configuration.Configuration;
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/notification/src/main/java/org/apache/atlas/kafka/AtlasKafkaConsumer.java
----------------------------------------------------------------------
diff --git a/notification/src/main/java/org/apache/atlas/kafka/AtlasKafkaConsumer.java b/notification/src/main/java/org/apache/atlas/kafka/AtlasKafkaConsumer.java
index e3bb71c..fd0c4e4 100644
--- a/notification/src/main/java/org/apache/atlas/kafka/AtlasKafkaConsumer.java
+++ b/notification/src/main/java/org/apache/atlas/kafka/AtlasKafkaConsumer.java
@@ -18,7 +18,8 @@
package org.apache.atlas.kafka;
import org.apache.atlas.notification.AbstractNotificationConsumer;
-import org.apache.atlas.notification.MessageDeserializer;
+import org.apache.atlas.notification.AtlasNotificationMessageDeserializer;
+import org.apache.atlas.notification.NotificationInterface;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -41,13 +42,18 @@ public class AtlasKafkaConsumer<T> extends AbstractNotificationConsumer<T> {
private static final Logger LOG = LoggerFactory.getLogger(AtlasKafkaConsumer.class);
private final KafkaConsumer kafkaConsumer;
- private final boolean autoCommitEnabled;
- private long pollTimeoutMilliSeconds = 1000L;
+ private final boolean autoCommitEnabled;
+ private long pollTimeoutMilliSeconds = 1000L;
- public AtlasKafkaConsumer(MessageDeserializer<T> deserializer, KafkaConsumer kafkaConsumer, boolean autoCommitEnabled, long pollTimeoutMilliSeconds) {
+ public AtlasKafkaConsumer(NotificationInterface.NotificationType notificationType, KafkaConsumer kafkaConsumer, boolean autoCommitEnabled, long pollTimeoutMilliSeconds) {
+ this(notificationType.getDeserializer(), kafkaConsumer, autoCommitEnabled, pollTimeoutMilliSeconds);
+ }
+
+ public AtlasKafkaConsumer(AtlasNotificationMessageDeserializer<T> deserializer, KafkaConsumer kafkaConsumer, boolean autoCommitEnabled, long pollTimeoutMilliSeconds) {
super(deserializer);
- this.kafkaConsumer = kafkaConsumer;
- this.autoCommitEnabled = autoCommitEnabled;
+
+ this.autoCommitEnabled = autoCommitEnabled;
+ this.kafkaConsumer = kafkaConsumer;
this.pollTimeoutMilliSeconds = pollTimeoutMilliSeconds;
}
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/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 6bb8d73..4d6b444 100644
--- a/notification/src/main/java/org/apache/atlas/kafka/KafkaNotification.java
+++ b/notification/src/main/java/org/apache/atlas/kafka/KafkaNotification.java
@@ -23,6 +23,7 @@ import kafka.server.KafkaServer;
import kafka.utils.Time;
import org.apache.atlas.ApplicationProperties;
import org.apache.atlas.AtlasException;
+import org.apache.atlas.model.notification.AtlasNotificationMessage;
import org.apache.atlas.notification.AbstractNotification;
import org.apache.atlas.notification.NotificationConsumer;
import org.apache.atlas.notification.NotificationException;
@@ -40,6 +41,7 @@ import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.zookeeper.server.NIOServerCnxnFactory;
import org.apache.zookeeper.server.ServerCnxnFactory;
import org.apache.zookeeper.server.ZooKeeperServer;
+import org.codehaus.jackson.type.TypeReference;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.annotation.Order;
@@ -185,7 +187,8 @@ public class KafkaNotification extends AbstractNotification implements Service {
Properties consumerProperties = getConsumerProperties(notificationType);
List<NotificationConsumer<T>> consumers = new ArrayList<>();
- AtlasKafkaConsumer kafkaConsumer = new AtlasKafkaConsumer(notificationType.getDeserializer(), getKafkaConsumer(consumerProperties,notificationType, autoCommitEnabled), autoCommitEnabled, pollTimeOutMs );
+ AtlasKafkaConsumer kafkaConsumer =new AtlasKafkaConsumer(notificationType, getKafkaConsumer(consumerProperties, notificationType, autoCommitEnabled), autoCommitEnabled, pollTimeOutMs);
+
consumers.add(kafkaConsumer);
return consumers;
}
http://git-wip-us.apache.org/repos/asf/atlas/blob/64e739da/notification/src/main/java/org/apache/atlas/notification/AbstractMessageDeserializer.java
----------------------------------------------------------------------
diff --git a/notification/src/main/java/org/apache/atlas/notification/AbstractMessageDeserializer.java b/notification/src/main/java/org/apache/atlas/notification/AbstractMessageDeserializer.java
index 5bfe90e..a787141 100644
--- a/notification/src/main/java/org/apache/atlas/notification/AbstractMessageDeserializer.java
+++ b/notification/src/main/java/org/apache/atlas/notification/AbstractMessageDeserializer.java
@@ -18,145 +18,30 @@
package org.apache.atlas.notification;
-import com.google.common.collect.ImmutableList;
-import com.google.common.collect.ImmutableMap;
-import com.google.gson.Gson;
-import com.google.gson.GsonBuilder;
-import com.google.gson.JsonDeserializationContext;
-import com.google.gson.JsonDeserializer;
-import com.google.gson.JsonElement;
-import com.google.gson.JsonParseException;
-import com.google.gson.reflect.TypeToken;
-import org.apache.atlas.v1.model.instance.Referenceable;
-import org.apache.atlas.v1.model.instance.Struct;
-import org.apache.atlas.type.AtlasType;
-import org.codehaus.jettison.json.JSONArray;
-import org.codehaus.jettison.json.JSONException;
+import org.apache.atlas.model.notification.AtlasNotificationMessage;
+import org.apache.atlas.model.notification.MessageVersion;
+import org.codehaus.jackson.type.TypeReference;
import org.slf4j.Logger;
-import java.lang.reflect.Type;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
/**
* Base notification message deserializer.
*/
public abstract class AbstractMessageDeserializer<T> extends AtlasNotificationMessageDeserializer<T> {
- private static final Map<Type, JsonDeserializer> DESERIALIZER_MAP = new HashMap<>();
-
- static {
- DESERIALIZER_MAP.put(ImmutableList.class, new ImmutableListDeserializer());
- DESERIALIZER_MAP.put(ImmutableMap.class, new ImmutableMapDeserializer());
- DESERIALIZER_MAP.put(JSONArray.class, new JSONArrayDeserializer());
- DESERIALIZER_MAP.put(Struct.class, new StructDeserializer());
- DESERIALIZER_MAP.put(Referenceable.class, new ReferenceableDeserializer());
- }
-
-
// ----- Constructors ----------------------------------------------------
/**
* Create a deserializer.
*
- * @param notificationMessageType the type of the notification message
* @param expectedVersion the expected message version
- * @param deserializerMap map of individual deserializers used to define this message deserializer
* @param notificationLogger logger for message version mismatch
*/
- public AbstractMessageDeserializer(Type notificationMessageType,
- MessageVersion expectedVersion,
- Map<Type, JsonDeserializer> deserializerMap,
- Logger notificationLogger) {
- super(notificationMessageType, expectedVersion, getDeserializer(deserializerMap), notificationLogger);
+ public AbstractMessageDeserializer(TypeReference<T> messageType,
+ TypeReference<AtlasNotificationMessage<T>> notificationMessageType,
+ MessageVersion expectedVersion, Logger notificationLogger) {
+ super(messageType, notificationMessageType, expectedVersion, notificationLogger);
}
// ----- helper methods --------------------------------------------------
-
- private static Gson getDeserializer(Map<Type, JsonDeserializer> deserializerMap) {
- GsonBuilder builder = new GsonBuilder();
-
- for (Map.Entry<Type, JsonDeserializer> entry : DESERIALIZER_MAP.entrySet()) {
- builder.registerTypeAdapter(entry.getKey(), entry.getValue());
- }
-
- for (Map.Entry<Type, JsonDeserializer> entry : deserializerMap.entrySet()) {
- builder.registerTypeAdapter(entry.getKey(), entry.getValue());
- }
- return builder.create();
- }
-
-
- // ----- deserializer classes --------------------------------------------
-
- /**
- * Deserializer for ImmutableList.
- */
- protected static class ImmutableListDeserializer implements JsonDeserializer<ImmutableList<?>> {
- public static final Type LIST_TYPE = new TypeToken<List<?>>() {
- }.getType();
-
- @Override
- public ImmutableList<?> deserialize(JsonElement json, Type type,
- JsonDeserializationContext context) {
- final List<?> list = context.deserialize(json, LIST_TYPE);
- return ImmutableList.copyOf(list);
- }
- }
-
- /**
- * Deserializer for ImmutableMap.
- */
- protected static class ImmutableMapDeserializer implements JsonDeserializer<ImmutableMap<?, ?>> {
-
- public static final Type MAP_TYPE = new TypeToken<Map<?, ?>>() {
- }.getType();
-
- @Override
- public ImmutableMap<?, ?> deserialize(JsonElement json, Type type,
- JsonDeserializationContext context) {
- final Map<?, ?> map = context.deserialize(json, MAP_TYPE);
- return ImmutableMap.copyOf(map);
- }
- }
-
- /**
- * Deserializer for JSONArray.
- */
- public static final class JSONArrayDeserializer implements JsonDeserializer<JSONArray> {
- @Override
- public JSONArray deserialize(final JsonElement json, final Type type,
- final JsonDeserializationContext context) {
- try {
- return new JSONArray(json.toString());
- } catch (JSONException e) {
- throw new JsonParseException(e.getMessage(), e);
- }
- }
- }
-
- /**
- * Deserializer for Struct.
- */
- protected static final class StructDeserializer implements JsonDeserializer<Struct> {
- @Override
- public Struct deserialize(final JsonElement json, final Type type,
- final JsonDeserializationContext context) {
- return context.deserialize(json, Struct.class);
- }
- }
-
- /**
- * Deserializer for Referenceable.
- */
- protected static final class ReferenceableDeserializer implements JsonDeserializer<Referenceable> {
- @Override
- public Referenceable deserialize(final JsonElement json, final Type type,
- final JsonDeserializationContext context) {
-
- return AtlasType.fromV1Json(json.toString(), Referenceable.class);
- }
- }
}