Commit 3de30f55 by Ashutosh Mestry

ATLAS-3799: EntityConumer only adds entity GUIDs that are added when they were produced.

parent 214f2a76
...@@ -59,7 +59,7 @@ public class EntityConsumer extends WorkItemConsumer<AtlasEntity.AtlasEntityWith ...@@ -59,7 +59,7 @@ public class EntityConsumer extends WorkItemConsumer<AtlasEntity.AtlasEntityWith
private final EntityGraphRetriever entityRetrieverBulk; private final EntityGraphRetriever entityRetrieverBulk;
private List<AtlasEntity.AtlasEntityWithExtInfo> entityBuffer = new ArrayList<>(); private List<AtlasEntity.AtlasEntityWithExtInfo> entityBuffer = new ArrayList<>();
private List<EntityMutationResponse> localResults = new ArrayList<>(); private List<String> localResults = new ArrayList<>();
public EntityConsumer(AtlasTypeRegistry typeRegistry, public EntityConsumer(AtlasTypeRegistry typeRegistry,
AtlasGraph atlasGraph, AtlasEntityStore entityStore, AtlasGraph atlasGraph, AtlasEntityStore entityStore,
...@@ -119,7 +119,7 @@ public class EntityConsumer extends WorkItemConsumer<AtlasEntity.AtlasEntityWith ...@@ -119,7 +119,7 @@ public class EntityConsumer extends WorkItemConsumer<AtlasEntity.AtlasEntityWith
private void importUsingBulkEntityStore(AtlasEntity.AtlasEntityWithExtInfo entityWithExtInfo) throws AtlasBaseException { private void importUsingBulkEntityStore(AtlasEntity.AtlasEntityWithExtInfo entityWithExtInfo) throws AtlasBaseException {
EntityStream oneEntityStream = new AtlasEntityStreamForImport(entityWithExtInfo, null); EntityStream oneEntityStream = new AtlasEntityStreamForImport(entityWithExtInfo, null);
EntityMutationResponse result = entityStoreBulk.createOrUpdateForImportNoCommit(oneEntityStream); EntityMutationResponse result = entityStoreBulk.createOrUpdateForImportNoCommit(oneEntityStream);
localResults.add(result); localResults.add(entityWithExtInfo.getEntity().getGuid());
entityBuffer.add(entityWithExtInfo); entityBuffer.add(entityWithExtInfo);
} }
...@@ -133,9 +133,9 @@ public class EntityConsumer extends WorkItemConsumer<AtlasEntity.AtlasEntityWith ...@@ -133,9 +133,9 @@ public class EntityConsumer extends WorkItemConsumer<AtlasEntity.AtlasEntityWith
try { try {
LOG.info("Regular: EntityStore: {}: Starting...", this.counter.get()); LOG.info("Regular: EntityStore: {}: Starting...", this.counter.get());
AtlasEntityStreamForImport oneEntityStream = new AtlasEntityStreamForImport(entityWithExtInfo, null); AtlasEntityStreamForImport oneEntityStream = new AtlasEntityStreamForImport(entityWithExtInfo, null);
EntityMutationResponse result = this.entityStore.createOrUpdateForImportNoCommit(oneEntityStream); this.entityStore.createOrUpdateForImportNoCommit(oneEntityStream);
atlasGraph.commit(); atlasGraph.commit();
localResults.add(result); localResults.add(entityWithExtInfo.getEntity().getGuid());
dispatchResults(); dispatchResults();
} catch (Exception e) { } catch (Exception e) {
atlasGraph.rollback(); atlasGraph.rollback();
...@@ -244,12 +244,7 @@ public class EntityConsumer extends WorkItemConsumer<AtlasEntity.AtlasEntityWith ...@@ -244,12 +244,7 @@ public class EntityConsumer extends WorkItemConsumer<AtlasEntity.AtlasEntityWith
} }
private void dispatchResults() { private void dispatchResults() {
localResults.stream().forEach(x -> { localResults.stream().forEach(x -> addResult(x));
addResultsFromResponse(x.getCreatedEntities());
addResultsFromResponse(x.getUpdatedEntities());
addResultsFromResponse(x.getDeletedEntities());
});
clear(); clear();
} }
...@@ -261,16 +256,6 @@ public class EntityConsumer extends WorkItemConsumer<AtlasEntity.AtlasEntityWith ...@@ -261,16 +256,6 @@ public class EntityConsumer extends WorkItemConsumer<AtlasEntity.AtlasEntityWith
} }
} }
private void addResultsFromResponse(List<AtlasEntityHeader> entities) {
if (CollectionUtils.isEmpty(entities)) {
return;
}
for (AtlasEntityHeader eh : entities) {
addResult(eh.getGuid());
}
}
private void clear() { private void clear() {
localResults.clear(); localResults.clear();
entityBuffer.clear(); entityBuffer.clear();
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment