This is an automated email from the ASF dual-hosted git repository.
madhan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/atlas.git
The following commit(s) were added to refs/heads/master by this push:
new 3841a1b ATLAS-3054: updated notification processing to support
batch-commits
3841a1b is described below
commit 3841a1bfd41b485f3cdeaad05abba2b14289c9d6
Author: Madhan Neethiraj <mad...@apache.org>
AuthorDate: Thu Feb 21 07:31:56 2019 -0800
ATLAS-3054: updated notification processing to support batch-commits
---
.../store/graph/v2/AtlasEntityStream.java | 18 +++----
.../notification/NotificationHookConsumer.java | 55 +++++++++++++++-------
.../preprocessor/PreprocessorContext.java | 29 ++++++++++++
3 files changed, 77 insertions(+), 25 deletions(-)
diff --git
a/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/AtlasEntityStream.java
b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/AtlasEntityStream.java
index 75a7e61..d12b036 100644
---
a/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/AtlasEntityStream.java
+++
b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/AtlasEntityStream.java
@@ -30,27 +30,27 @@ public class AtlasEntityStream implements EntityStream {
public AtlasEntityStream(AtlasEntity entity) {
- this(new AtlasEntitiesWithExtInfo(entity));
+ this(new AtlasEntitiesWithExtInfo(entity), null);
}
public AtlasEntityStream(AtlasEntityWithExtInfo entityWithExtInfo) {
- this(new AtlasEntitiesWithExtInfo(entityWithExtInfo));
+ this(new AtlasEntitiesWithExtInfo(entityWithExtInfo), null);
}
public AtlasEntityStream(AtlasEntitiesWithExtInfo entitiesWithExtInfo) {
- this.entitiesWithExtInfo = entitiesWithExtInfo;
- this.iterator =
this.entitiesWithExtInfo.getEntities().iterator();
- this.entityStream = null;
+ this(entitiesWithExtInfo, null);
}
public AtlasEntityStream(AtlasEntity entity, EntityStream entityStream) {
- this.entitiesWithExtInfo = new AtlasEntitiesWithExtInfo(entity);
- this.iterator =
this.entitiesWithExtInfo.getEntities().iterator();
- this.entityStream = entityStream;
+ this(new AtlasEntitiesWithExtInfo(entity), entityStream);
}
public AtlasEntityStream(AtlasEntityWithExtInfo entityWithExtInfo,
EntityStream entityStream) {
- this.entitiesWithExtInfo = new
AtlasEntitiesWithExtInfo(entityWithExtInfo);
+ this(new AtlasEntitiesWithExtInfo(entityWithExtInfo), entityStream);
+ }
+
+ public AtlasEntityStream(AtlasEntitiesWithExtInfo entitiesWithExtInfo,
EntityStream entityStream) {
+ this.entitiesWithExtInfo = entitiesWithExtInfo;
this.iterator =
this.entitiesWithExtInfo.getEntities().iterator();
this.entityStream = entityStream;
}
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 f829260..6d03ba4 100644
---
a/webapp/src/main/java/org/apache/atlas/notification/NotificationHookConsumer.java
+++
b/webapp/src/main/java/org/apache/atlas/notification/NotificationHookConsumer.java
@@ -27,6 +27,7 @@ import org.apache.atlas.AtlasConfiguration;
import org.apache.atlas.AtlasException;
import org.apache.atlas.AtlasServiceException;
import org.apache.atlas.RequestContext;
+import org.apache.atlas.exception.AtlasBaseException;
import org.apache.atlas.ha.HAConfiguration;
import org.apache.atlas.kafka.AtlasKafkaMessage;
import org.apache.atlas.listener.ActiveStateChangeHandler;
@@ -113,8 +114,10 @@ public class NotificationHookConsumer implements Service,
ActiveStateChangeHandl
public static final String CONSUMER_RETRY_INTERVAL =
"atlas.notification.consumer.retry.interval";
public static final String CONSUMER_MIN_RETRY_INTERVAL =
"atlas.notification.consumer.min.retry.interval";
public static final String CONSUMER_MAX_RETRY_INTERVAL =
"atlas.notification.consumer.max.retry.interval";
+ public static final String CONSUMER_COMMIT_BATCH_SIZE =
"atlas.notification.consumer.commit.batch.size";
public static final String CONSUMER_DISABLED =
"atlas.notification.consumer.disabled";
+
public static final String CONSUMER_SKIP_HIVE_COLUMN_LINEAGE_HIVE_20633
=
"atlas.notification.consumer.skip.hive_column_lineage.hive-20633";
public static final String
CONSUMER_SKIP_HIVE_COLUMN_LINEAGE_HIVE_20633_INPUTS_THRESHOLD =
"atlas.notification.consumer.skip.hive_column_lineage.hive-20633.inputs.threshold";
public static final String CONSUMER_PREPROCESS_HIVE_TABLE_IGNORE_PATTERN
=
"atlas.notification.consumer.preprocess.hive_table.ignore.pattern";
@@ -131,6 +134,7 @@ public class NotificationHookConsumer implements Service,
ActiveStateChangeHandl
private final int failedMsgCacheSize;
private final int minWaitDuration;
private final int maxWaitDuration;
+ private final int commitBatchSize;
private final boolean skipHiveColumnLineageHive20633;
private final int
skipHiveColumnLineageHive20633InputsThreshold;
private final int
largeMessageProcessingTimeThresholdMs;
@@ -166,10 +170,11 @@ public class NotificationHookConsumer implements Service,
ActiveStateChangeHandl
consumerRetryInterval =
applicationProperties.getInt(CONSUMER_RETRY_INTERVAL, 500);
minWaitDuration =
applicationProperties.getInt(CONSUMER_MIN_RETRY_INTERVAL,
consumerRetryInterval); // 500 ms by default
maxWaitDuration =
applicationProperties.getInt(CONSUMER_MAX_RETRY_INTERVAL, minWaitDuration *
60); // 30 sec by default
+ commitBatchSize =
applicationProperties.getInt(CONSUMER_COMMIT_BATCH_SIZE, 50);
skipHiveColumnLineageHive20633 =
applicationProperties.getBoolean(CONSUMER_SKIP_HIVE_COLUMN_LINEAGE_HIVE_20633,
false);
skipHiveColumnLineageHive20633InputsThreshold =
applicationProperties.getInt(CONSUMER_SKIP_HIVE_COLUMN_LINEAGE_HIVE_20633_INPUTS_THRESHOLD,
15); // skip if avg # of inputs is > 15
- consumerDisabled
= applicationProperties.getBoolean(CONSUMER_DISABLED, false);
+ consumerDisabled =
applicationProperties.getBoolean(CONSUMER_DISABLED, false);
largeMessageProcessingTimeThresholdMs =
applicationProperties.getInt("atlas.notification.consumer.large.message.processing.time.threshold.ms",
60 * 1000); // 60 sec by default
String[] patternHiveTablesToIgnore =
applicationProperties.getStringArray(CONSUMER_PREPROCESS_HIVE_TABLE_IGNORE_PATTERN);
@@ -500,7 +505,7 @@ public class NotificationHookConsumer implements Service,
ActiveStateChangeHandl
AtlasClient.API_V1.CREATE_ENTITY.getNormalizedPath());
}
- atlasEntityStore.createOrUpdate(new
AtlasEntityStream(entities), false);
+ createOrUpdate(entities, false);
}
break;
@@ -521,7 +526,7 @@ public class NotificationHookConsumer implements Service,
ActiveStateChangeHandl
// There should only be one root entity
entities.getEntities().get(0).setGuid(guid);
- atlasEntityStore.createOrUpdate(new
AtlasEntityStream(entities), true);
+ createOrUpdate(entities, true);
}
break;
@@ -554,7 +559,7 @@ public class NotificationHookConsumer implements Service,
ActiveStateChangeHandl
AtlasClientV2.API_V2.UPDATE_ENTITY.getNormalizedPath());
}
- atlasEntityStore.createOrUpdate(new
AtlasEntityStream(entities), false);
+ createOrUpdate(entities, false);
}
break;
@@ -568,7 +573,7 @@ public class NotificationHookConsumer implements Service,
ActiveStateChangeHandl
AtlasClientV2.API_V2.CREATE_ENTITY.getNormalizedPath());
}
- atlasEntityStore.createOrUpdate(new
AtlasEntityStream(entities), false);
+ createOrUpdate(entities, false);
}
break;
@@ -597,7 +602,7 @@ public class NotificationHookConsumer implements Service,
ActiveStateChangeHandl
AtlasClientV2.API_V2.UPDATE_ENTITY.getNormalizedPath());
}
- atlasEntityStore.createOrUpdate(new
AtlasEntityStream(entities), false);
+ createOrUpdate(entities, false);
}
break;
@@ -682,6 +687,32 @@ public class NotificationHookConsumer implements Service,
ActiveStateChangeHandl
}
}
+ private void createOrUpdate(AtlasEntitiesWithExtInfo entities, boolean
isPartialUpdate) throws AtlasBaseException {
+ List<AtlasEntity> entitiesList = entities.getEntities();
+ AtlasEntityStream entityStream = new AtlasEntityStream(entities);
+
+ if (entitiesList.size() <= commitBatchSize) {
+ atlasEntityStore.createOrUpdate(entityStream, isPartialUpdate);
+ } else {
+ for (int fromIdx = 0; fromIdx < entitiesList.size(); fromIdx
+= commitBatchSize) {
+ int toIndex = fromIdx + commitBatchSize;
+
+ if (toIndex > entitiesList.size()) {
+ toIndex = entitiesList.size();
+ }
+
+ AtlasEntitiesWithExtInfo batch = new
AtlasEntitiesWithExtInfo(new ArrayList<>(entitiesList.subList(fromIdx,
toIndex)));
+ AtlasEntityStream batchStream = new
AtlasEntityStream(batch, entityStream);
+
+ atlasEntityStore.createOrUpdate(batchStream,
isPartialUpdate);
+
+ RequestContext.get().resetEntityGuidUpdates();
+
+ RequestContext.get().clearCache();
+ }
+ }
+ }
+
private void recordFailedMessages() {
//logging failed messages
for (String message : failedMessages) {
@@ -767,6 +798,8 @@ public class NotificationHookConsumer implements Service,
ActiveStateChangeHandl
if (skipHiveColumnLineageHive20633) {
skipHiveColumnLineage(context);
}
+
+ context.moveRegisteredReferredEntities();
}
private void ignoreOrPruneHiveTables(PreprocessorContext context) {
@@ -801,16 +834,6 @@ public class NotificationHookConsumer implements Service,
ActiveStateChangeHandl
}
}
}
-
- for (String guid : context.getReferredEntitiesToMove()) {
- AtlasEntity entity = referredEntities.remove(guid);
-
- if (entity != null) {
- entities.add(entity);
-
- LOG.info("moved referred entity: typeName={},
qualifiedName={}. topic-offset={}, partition={}", entity.getTypeName(),
EntityPreprocessor.getQualifiedName(entity), context.getKafkaMessageOffset(),
context.getKafkaPartition());
- }
- }
}
int ignoredEntities = context.getIgnoredEntities().size();
diff --git
a/webapp/src/main/java/org/apache/atlas/notification/preprocessor/PreprocessorContext.java
b/webapp/src/main/java/org/apache/atlas/notification/preprocessor/PreprocessorContext.java
index 7b4229b..2d2c09a 100644
---
a/webapp/src/main/java/org/apache/atlas/notification/preprocessor/PreprocessorContext.java
+++
b/webapp/src/main/java/org/apache/atlas/notification/preprocessor/PreprocessorContext.java
@@ -173,6 +173,35 @@ public class PreprocessorContext {
collectGuids(obj, prunedEntities);
}
+ public void moveRegisteredReferredEntities() {
+ List<AtlasEntity> entities = getEntities();
+ Map<String, AtlasEntity> referredEntities = getReferredEntities();
+
+ if (entities != null && referredEntities != null &&
!referredEntitiesToMove.isEmpty()) {
+ AtlasEntity firstEntity = entities.isEmpty() ? null :
entities.get(0);
+
+ for (String guid : referredEntitiesToMove) {
+ AtlasEntity entity = referredEntities.remove(guid);
+
+ if (entity != null) {
+ entities.add(entity);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("moved referred entity: typeName={},
qualifiedName={}. topic-offset={}, partition={}", entity.getTypeName(),
EntityPreprocessor.getQualifiedName(entity), kafkaMessage.getOffset(),
kafkaMessage.getPartition());
+ }
+ }
+ }
+
+ if (firstEntity != null) {
+ LOG.info("moved {} referred-entities to end of entities-list
(firstEntity:typeName={}, qualifiedName={}). topic-offset={}, partition={}",
referredEntitiesToMove.size(), firstEntity.getTypeName(),
EntityPreprocessor.getQualifiedName(firstEntity), kafkaMessage.getOffset(),
kafkaMessage.getPartition());
+ } else {
+ LOG.info("moved {} referred-entities to entities-list.
topic-offset={}, partition={}", referredEntitiesToMove.size(),
kafkaMessage.getOffset(), kafkaMessage.getPartition());
+ }
+
+ referredEntitiesToMove.clear();
+ }
+ }
+
public String getGuid(Object obj) {
Object ret = null;
