From 62280f7a728225aeda0d2ae2f0e1b9c8bc434957 Mon Sep 17 00:00:00 2001 From: hr2904 Date: Thu, 7 Nov 2024 13:25:37 +0530 Subject: [PATCH 01/12] Added logic for updating modificationTimestamp and modificationBy for BM attribute update. --- .../atlas/repository/store/graph/v2/EntityGraphMapper.java | 1 + 1 file changed, 1 insertion(+) diff --git a/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/EntityGraphMapper.java b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/EntityGraphMapper.java index 0b5eb09fca..ad3c0cf11b 100644 --- a/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/EntityGraphMapper.java +++ b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/EntityGraphMapper.java @@ -823,6 +823,7 @@ public void addOrUpdateBusinessAttributes(AtlasVertex entityVertex, AtlasEntityT } if (MapUtils.isNotEmpty(updatedBusinessAttributes)) { + updateModificationMetadata(entityVertex); entityChangeNotifier.onBusinessAttributesUpdated(AtlasGraphUtilsV2.getIdFromVertex(entityVertex), updatedBusinessAttributes); } From 2ddf8fe2e62ce1a2d4ce2c2beb6e46587322dfcc Mon Sep 17 00:00:00 2001 From: hr2904 Date: Thu, 7 Nov 2024 13:29:35 +0530 Subject: [PATCH 02/12] Added logic for updating modificationTimestamp and modificationBy for BM attribute update - 2 --- .../atlas/repository/store/graph/v2/EntityGraphMapper.java | 1 + 1 file changed, 1 insertion(+) diff --git a/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/EntityGraphMapper.java b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/EntityGraphMapper.java index ad3c0cf11b..5152bb2e5d 100644 --- a/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/EntityGraphMapper.java +++ b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/EntityGraphMapper.java @@ -739,6 +739,7 @@ public void setBusinessAttributes(AtlasVertex entityVertex, AtlasEntityType enti } if (MapUtils.isNotEmpty(updatedBusinessAttributes)) { + updateModificationMetadata(entityVertex); entityChangeNotifier.onBusinessAttributesUpdated(AtlasGraphUtilsV2.getIdFromVertex(entityVertex), updatedBusinessAttributes); } From f227a6895a137b0a3892b1d14fce6bc7f5074b76 Mon Sep 17 00:00:00 2001 From: hr2904 Date: Mon, 11 Nov 2024 13:54:04 +0530 Subject: [PATCH 03/12] Added secure in ATLASSESSIONID Cookie. --- .github/workflows/maven.yml | 4 ++-- webapp/src/main/webapp/WEB-INF/web.xml | 1 + 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/.github/workflows/maven.yml b/.github/workflows/maven.yml index f8a09b5589..3af25b1ea5 100644 --- a/.github/workflows/maven.yml +++ b/.github/workflows/maven.yml @@ -25,7 +25,7 @@ on: - beta - development - master - - lineageondemand + - dg1908 jobs: build: @@ -64,7 +64,7 @@ jobs: - name: Build with Maven run: | branch_name=${{ steps.get_branch.outputs.branch }} - if [[ $branch_name == 'main' || $branch_name == 'master' || $branch_name == 'lineageondemand' ]] + if [[ $branch_name == 'main' || $branch_name == 'master' || $branch_name == 'dg1908' ]] then echo "build without dashboard" chmod +x ./build.sh && ./build.sh build_without_dashboard diff --git a/webapp/src/main/webapp/WEB-INF/web.xml b/webapp/src/main/webapp/WEB-INF/web.xml index 07092d62eb..590901b279 100755 --- a/webapp/src/main/webapp/WEB-INF/web.xml +++ b/webapp/src/main/webapp/WEB-INF/web.xml @@ -129,6 +129,7 @@ ATLASSESSIONID true + true From b15906f7e4276ea167521af10b904ffa7e42f1c4 Mon Sep 17 00:00:00 2001 From: suraj5077 Date: Tue, 12 Nov 2024 16:52:19 +0530 Subject: [PATCH 04/12] fix --- .../java/org/apache/atlas/discovery/EntityLineageService.java | 1 + 1 file changed, 1 insertion(+) diff --git a/repository/src/main/java/org/apache/atlas/discovery/EntityLineageService.java b/repository/src/main/java/org/apache/atlas/discovery/EntityLineageService.java index a873105a6d..0c5850f241 100644 --- a/repository/src/main/java/org/apache/atlas/discovery/EntityLineageService.java +++ b/repository/src/main/java/org/apache/atlas/discovery/EntityLineageService.java @@ -543,6 +543,7 @@ private void enqueueNeighbours(AtlasVertex currentVertex, boolean isDataset, RequestContext.get().endMetricRecord(traverseEdgesOnDemandGetEdges); while (edges.hasNext()) { + AtlasEdge currentEdge = edges.next(); if (!lineageListContext.evaluateTraversalFilter(currentEdge)) continue; From 3a21da7046356b9639e7f71ad3a7a8b87f34e051 Mon Sep 17 00:00:00 2001 From: suraj5077 Date: Tue, 12 Nov 2024 17:13:42 +0530 Subject: [PATCH 05/12] ci fix --- .github/workflows/maven.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/maven.yml b/.github/workflows/maven.yml index f8a09b5589..84d3ddfe43 100644 --- a/.github/workflows/maven.yml +++ b/.github/workflows/maven.yml @@ -77,7 +77,7 @@ jobs: shell: bash - name: Get version tag - run: echo "##[set-output name=version;]$(echo `git ls-remote https://${{ secrets.ORG_PAT_GITHUB }}@github.com/atlanhq/${REPOSITORY_NAME}.git ${{ steps.get_branch.outputs.branch }} | awk '{ print $1}' | cut -c1-7`)abcd" + run: echo "##[set-output name=version;]$(echo `git ls-remote https://${{ secrets.ORG_PAT_GITHUB }}@github.com/atlanhq/${REPOSITORY_NAME}.git refs/heads/${{ steps.get_branch.outputs.branch }} | awk '{ print $1}' | cut -c1-7`)abcd" id: get_version - name: Set up Buildx From 34b4f6d7727d1bad638ca673f31529e927c10e3e Mon Sep 17 00:00:00 2001 From: suraj5077 Date: Wed, 13 Nov 2024 00:25:29 +0530 Subject: [PATCH 06/12] Timeout - response clean up handling --- intg/src/main/java/org/apache/atlas/AtlasConfiguration.java | 2 +- .../java/org/apache/atlas/discovery/EntityLineageService.java | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/intg/src/main/java/org/apache/atlas/AtlasConfiguration.java b/intg/src/main/java/org/apache/atlas/AtlasConfiguration.java index 7fd081b9e3..5fe8a5e96f 100644 --- a/intg/src/main/java/org/apache/atlas/AtlasConfiguration.java +++ b/intg/src/main/java/org/apache/atlas/AtlasConfiguration.java @@ -94,7 +94,7 @@ public enum AtlasConfiguration { GRAPH_TRAVERSAL_PARALLELISM("atlas.graph.traverse.bucket.size",10), LINEAGE_ON_DEMAND_ENABLED("atlas.lineage.on.demand.enabled", true), LINEAGE_ON_DEMAND_DEFAULT_NODE_COUNT("atlas.lineage.on.demand.default.node.count", 3), - LINEAGE_MAX_NODE_COUNT("atlas.lineage.max.node.count", 100), + LINEAGE_MAX_NODE_COUNT("atlas.lineage.max.node.count", 1000), SUPPORTED_RELATIONSHIP_EVENTS("atlas.notification.relationships.filter", "asset_readme,asset_links"), diff --git a/repository/src/main/java/org/apache/atlas/discovery/EntityLineageService.java b/repository/src/main/java/org/apache/atlas/discovery/EntityLineageService.java index a881539741..58d2717272 100644 --- a/repository/src/main/java/org/apache/atlas/discovery/EntityLineageService.java +++ b/repository/src/main/java/org/apache/atlas/discovery/EntityLineageService.java @@ -81,7 +81,7 @@ @Service public class EntityLineageService implements AtlasLineageService { private static final Logger LOG = LoggerFactory.getLogger(EntityLineageService.class); - private static final long LINEAGE_TRAVERSAL_TIMEOUT_MILLIS = 15000; + private static final long LINEAGE_TRAVERSAL_TIMEOUT_MILLIS = 4; private static final String PROCESS_INPUTS_EDGE = "__Process.inputs"; private static final String PROCESS_OUTPUTS_EDGE = "__Process.outputs"; private static final String COLUMNS = "columns"; From f31e7ef78dedd9061b99e293407851b6c7712263 Mon Sep 17 00:00:00 2001 From: suraj5077 Date: Wed, 13 Nov 2024 00:26:40 +0530 Subject: [PATCH 07/12] Timeout - response clean up handling --- .../java/org/apache/atlas/discovery/EntityLineageService.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/repository/src/main/java/org/apache/atlas/discovery/EntityLineageService.java b/repository/src/main/java/org/apache/atlas/discovery/EntityLineageService.java index 58d2717272..66591daa9d 100644 --- a/repository/src/main/java/org/apache/atlas/discovery/EntityLineageService.java +++ b/repository/src/main/java/org/apache/atlas/discovery/EntityLineageService.java @@ -81,7 +81,7 @@ @Service public class EntityLineageService implements AtlasLineageService { private static final Logger LOG = LoggerFactory.getLogger(EntityLineageService.class); - private static final long LINEAGE_TRAVERSAL_TIMEOUT_MILLIS = 4; + private static final long LINEAGE_TRAVERSAL_TIMEOUT_MILLIS = 4000; private static final String PROCESS_INPUTS_EDGE = "__Process.inputs"; private static final String PROCESS_OUTPUTS_EDGE = "__Process.outputs"; private static final String COLUMNS = "columns"; From 3202acdc84e87180869847119bb994e369741dd1 Mon Sep 17 00:00:00 2001 From: suraj5077 Date: Wed, 13 Nov 2024 02:33:43 +0530 Subject: [PATCH 08/12] fixes --- .../atlas/discovery/EntityLineageService.java | 63 ++++++++----------- 1 file changed, 27 insertions(+), 36 deletions(-) diff --git a/repository/src/main/java/org/apache/atlas/discovery/EntityLineageService.java b/repository/src/main/java/org/apache/atlas/discovery/EntityLineageService.java index 66591daa9d..c70edb0bde 100644 --- a/repository/src/main/java/org/apache/atlas/discovery/EntityLineageService.java +++ b/repository/src/main/java/org/apache/atlas/discovery/EntityLineageService.java @@ -347,7 +347,7 @@ private void traverseEdgesOnDemand(Iterator processEdges, boolean isI } boolean isInputEdge = processEdge.getLabel().equalsIgnoreCase(PROCESS_INPUTS_EDGE); - if (handleHorizontalAndVerticalPagination(processEdge, isInputEdge, atlasLineageOnDemandContext, ret, depth, entitiesTraversed, new HashSet<>(), timeoutTracker)) { + if (handleHorizontalAndVerticalPagination(processEdge, isInputEdge, atlasLineageOnDemandContext, ret, depth, entitiesTraversed, new HashSet<>(), false)) { break; } else { addEdgeToResult(processEdge, ret, atlasLineageOnDemandContext, nextLevel, traversalOrder); @@ -381,8 +381,14 @@ private void traverseEdgesOnDemand(AtlasVertex datasetVertex, boolean isInput, i while (incomingEdges.hasNext()) { AtlasEdge incomingEdge = incomingEdges.next(); - AtlasVertex processVertex = incomingEdge.getOutVertex(); + boolean isTimedOut = timeoutTracker.hasTimedOut(); + if (isTimedOut) { + handleHorizontalAndVerticalPagination(incomingEdge, !isInput, atlasLineageOnDemandContext, ret, depth, entitiesTraversed, visitedVertices, isTimedOut); + ret.setTraversalTimedOut(true); + break; + } + AtlasVertex processVertex = incomingEdge.getOutVertex(); if (!vertexMatchesEvaluation(processVertex, atlasLineageOnDemandContext) || !edgeMatchesEvaluation(incomingEdge, atlasLineageOnDemandContext)) { continue; } @@ -391,13 +397,8 @@ private void traverseEdgesOnDemand(AtlasVertex datasetVertex, boolean isInput, i continue; } - boolean stopProcessIteration = handleHorizontalAndVerticalPagination(incomingEdge, !isInput, atlasLineageOnDemandContext, ret, depth, entitiesTraversed, visitedVertices, timeoutTracker); + boolean stopProcessIteration = handleHorizontalAndVerticalPagination(incomingEdge, !isInput, atlasLineageOnDemandContext, ret, depth, entitiesTraversed, visitedVertices, isTimedOut); // If timeout occurred or entity limit reached, set pagination flags again for consistency and stop - if (timeoutTracker.hasTimedOut()) { - handleHorizontalAndVerticalPagination(incomingEdge, !isInput, atlasLineageOnDemandContext, ret, depth, entitiesTraversed, visitedVertices, timeoutTracker); - break; - } - if (stopProcessIteration) { LineageInfoOnDemand entityOnDemandInfo = ret.getRelationsOnDemand().get(baseGuid); if (entityOnDemandInfo == null) @@ -416,8 +417,14 @@ private void traverseEdgesOnDemand(AtlasVertex datasetVertex, boolean isInput, i while (outgoingEdges.hasNext()) { AtlasEdge outgoingEdge = outgoingEdges.next(); - AtlasVertex entityVertex = outgoingEdge.getInVertex(); + isTimedOut = timeoutTracker.hasTimedOut(); + if (isTimedOut) { + handleHorizontalAndVerticalPagination(incomingEdge, !isInput, atlasLineageOnDemandContext, ret, depth, entitiesTraversed, visitedVertices, isTimedOut); + ret.setTraversalTimedOut(true); + break; + } + AtlasVertex entityVertex = outgoingEdge.getInVertex(); if (!vertexMatchesEvaluation(entityVertex, atlasLineageOnDemandContext) || !edgeMatchesEvaluation(outgoingEdge, atlasLineageOnDemandContext)) { continue; } @@ -425,14 +432,8 @@ private void traverseEdgesOnDemand(AtlasVertex datasetVertex, boolean isInput, i if (checkForOffset(outgoingEdge, processVertex, atlasLineageOnDemandContext, ret)) { continue; } - boolean stopDatasetIteration = handleHorizontalAndVerticalPagination(outgoingEdge, isInput, atlasLineageOnDemandContext, ret, depth, entitiesTraversed, visitedVertices, timeoutTracker); - // If timeout occurred or entity limit reached, set pagination flags again for consistency - if (timeoutTracker.hasTimedOut()) { - handleHorizontalAndVerticalPagination(outgoingEdge, isInput, atlasLineageOnDemandContext, ret, depth, entitiesTraversed, visitedVertices, timeoutTracker); - ret.setTraversalTimedOut(true); - return; - } + boolean stopDatasetIteration = handleHorizontalAndVerticalPagination(outgoingEdge, isInput, atlasLineageOnDemandContext, ret, depth, entitiesTraversed, visitedVertices, isTimedOut); if (stopDatasetIteration) { String processGuid = AtlasGraphUtilsV2.getIdFromVertex(processVertex); LineageInfoOnDemand entityOnDemandInfo = ret.getRelationsOnDemand().get(processGuid); @@ -450,21 +451,12 @@ private void traverseEdgesOnDemand(AtlasVertex datasetVertex, boolean isInput, i if (isEntityTraversalLimitReached(entitiesTraversed)) setEntityLimitReachedFlag(isInput, ret); - if (timeoutTracker.hasTimedOut()) { - handleHorizontalAndVerticalPagination(outgoingEdge, isInput, atlasLineageOnDemandContext, ret, depth, entitiesTraversed, visitedVertices, timeoutTracker); - ret.setTraversalTimedOut(true); - return; - } } if (entityVertex != null && !visitedVertices.contains(getId(entityVertex))) { traverseEdgesOnDemand(entityVertex, isInput, depth - 1, nextLevel, visitedVertices, atlasLineageOnDemandContext, ret, baseGuid, entitiesTraversed, traversalOrder, timeoutTracker); // execute inner depth AtlasEntityHeader traversedEntity = ret.getGuidEntityMap().get(AtlasGraphUtilsV2.getIdFromVertex(entityVertex)); if (traversedEntity != null) traversedEntity.setFinishTime(traversalOrder.get()); - if (timeoutTracker.hasTimedOut()) { - ret.setTraversalTimedOut(true); - return; - } } } } @@ -715,7 +707,7 @@ private static String getId(AtlasVertex vertex) { return vertex.getIdForDisplay(); } - private boolean handleHorizontalAndVerticalPagination(AtlasEdge atlasEdge, boolean isInput, AtlasLineageOnDemandContext atlasLineageOnDemandContext, AtlasLineageOnDemandInfo ret, int depth, AtomicInteger entitiesTraversed, Set visitedVertices, TimeoutTracker timeoutTracker) { + private boolean handleHorizontalAndVerticalPagination(AtlasEdge atlasEdge, boolean isInput, AtlasLineageOnDemandContext atlasLineageOnDemandContext, AtlasLineageOnDemandInfo ret, int depth, AtomicInteger entitiesTraversed, Set visitedVertices, boolean isTimedOut) { AtlasPerfMetrics.MetricRecorder metricRecorder = RequestContext.get().startMetricRecord("incrementAndCheckIfRelationsLimitReached"); AtlasVertex inVertex = isInput ? atlasEdge.getOutVertex() : atlasEdge.getInVertex(); @@ -729,10 +721,9 @@ private boolean handleHorizontalAndVerticalPagination(AtlasEdge atlasEdge, boole LineageInfoOnDemand inLineageInfo = ret.getRelationsOnDemand().containsKey(inGuid) ? ret.getRelationsOnDemand().get(inGuid) : new LineageInfoOnDemand(inGuidLineageConstraints); LineageInfoOnDemand outLineageInfo = ret.getRelationsOnDemand().containsKey(outGuid) ? ret.getRelationsOnDemand().get(outGuid) : new LineageInfoOnDemand(outGuidLineageConstraints); + setHorizontalPaginationFlags(isInput, atlasLineageOnDemandContext, depth, entitiesTraversed, inVertex, outVertex, inLineageInfo, outLineageInfo, visitedVertices, isTimedOut); - setHorizontalPaginationFlags(isInput, atlasLineageOnDemandContext, depth, entitiesTraversed, inVertex, outVertex, inLineageInfo, outLineageInfo, visitedVertices, timeoutTracker); - - boolean hasRelationsLimitReached = setVerticalPaginationFlags(entitiesTraversed, inLineageInfo, outLineageInfo, timeoutTracker); + boolean hasRelationsLimitReached = setVerticalPaginationFlags(inLineageInfo, outLineageInfo, isTimedOut); if (!hasRelationsLimitReached) { ret.getRelationsOnDemand().put(inGuid, inLineageInfo); ret.getRelationsOnDemand().put(outGuid, outLineageInfo); @@ -742,14 +733,14 @@ private boolean handleHorizontalAndVerticalPagination(AtlasEdge atlasEdge, boole return hasRelationsLimitReached; } - private boolean setVerticalPaginationFlags(AtomicInteger entitiesTraversed, LineageInfoOnDemand inLineageInfo, LineageInfoOnDemand outLineageInfo, TimeoutTracker timeoutTracker) { - boolean hasRelationsLimitReached = false; - if (timeoutTracker.hasTimedOut()) { + private boolean setVerticalPaginationFlags(LineageInfoOnDemand inLineageInfo, LineageInfoOnDemand outLineageInfo, boolean isTimedOut) { + if (isTimedOut) { inLineageInfo.setHasMoreInputs(true); outLineageInfo.setHasMoreOutputs(true); + return true; } - - if (inLineageInfo.isInputRelationsReachedLimit() || outLineageInfo.isOutputRelationsReachedLimit() || isEntityTraversalLimitReached(entitiesTraversed)) { + boolean hasRelationsLimitReached = false; + if (inLineageInfo.isInputRelationsReachedLimit() || outLineageInfo.isOutputRelationsReachedLimit()) { inLineageInfo.setHasMoreInputs(true); outLineageInfo.setHasMoreOutputs(true); hasRelationsLimitReached = true; @@ -762,10 +753,10 @@ private boolean setVerticalPaginationFlags(AtomicInteger entitiesTraversed, Line return hasRelationsLimitReached; } - private void setHorizontalPaginationFlags(boolean isInput, AtlasLineageOnDemandContext atlasLineageOnDemandContext, int depth, AtomicInteger entitiesTraversed, AtlasVertex inVertex, AtlasVertex outVertex, LineageInfoOnDemand inLineageInfo, LineageInfoOnDemand outLineageInfo, Set visitedVertices, TimeoutTracker timeoutTracker) { + private void setHorizontalPaginationFlags(boolean isInput, AtlasLineageOnDemandContext atlasLineageOnDemandContext, int depth, AtomicInteger entitiesTraversed, AtlasVertex inVertex, AtlasVertex outVertex, LineageInfoOnDemand inLineageInfo, LineageInfoOnDemand outLineageInfo, Set visitedVertices, boolean isTimedOut) { boolean isOutVertexVisited = visitedVertices.contains(getId(outVertex)); boolean isInVertexVisited = visitedVertices.contains(getId(inVertex)); - if (depth == 1 || entitiesTraversed.get() == getLineageMaxNodeAllowedCount()-1 || timeoutTracker.hasTimedOut()) { // If traversal has to stop, set pagination flags + if (depth == 1 || entitiesTraversed.get() == getLineageMaxNodeAllowedCount()-1 || isTimedOut) { // If traversal has to stop, set pagination flags if (isInput && ! isOutVertexVisited) setHasUpstream(atlasLineageOnDemandContext, outVertex, outLineageInfo); else if (!isInput && ! isInVertexVisited) From 0587eb1f5b64ba1d004950698c52535088245cb2 Mon Sep 17 00:00:00 2001 From: suraj5077 Date: Wed, 13 Nov 2024 02:48:35 +0530 Subject: [PATCH 09/12] moved pagination logic before timeouts --- .../atlas/discovery/EntityLineageService.java | 18 ++++++++++-------- 1 file changed, 10 insertions(+), 8 deletions(-) diff --git a/repository/src/main/java/org/apache/atlas/discovery/EntityLineageService.java b/repository/src/main/java/org/apache/atlas/discovery/EntityLineageService.java index c70edb0bde..9872d8103c 100644 --- a/repository/src/main/java/org/apache/atlas/discovery/EntityLineageService.java +++ b/repository/src/main/java/org/apache/atlas/discovery/EntityLineageService.java @@ -381,6 +381,11 @@ private void traverseEdgesOnDemand(AtlasVertex datasetVertex, boolean isInput, i while (incomingEdges.hasNext()) { AtlasEdge incomingEdge = incomingEdges.next(); + + if (checkForOffset(incomingEdge, datasetVertex, atlasLineageOnDemandContext, ret)) { + continue; + } + boolean isTimedOut = timeoutTracker.hasTimedOut(); if (isTimedOut) { handleHorizontalAndVerticalPagination(incomingEdge, !isInput, atlasLineageOnDemandContext, ret, depth, entitiesTraversed, visitedVertices, isTimedOut); @@ -393,10 +398,6 @@ private void traverseEdgesOnDemand(AtlasVertex datasetVertex, boolean isInput, i continue; } - if (checkForOffset(incomingEdge, datasetVertex, atlasLineageOnDemandContext, ret)) { - continue; - } - boolean stopProcessIteration = handleHorizontalAndVerticalPagination(incomingEdge, !isInput, atlasLineageOnDemandContext, ret, depth, entitiesTraversed, visitedVertices, isTimedOut); // If timeout occurred or entity limit reached, set pagination flags again for consistency and stop if (stopProcessIteration) { @@ -417,6 +418,11 @@ private void traverseEdgesOnDemand(AtlasVertex datasetVertex, boolean isInput, i while (outgoingEdges.hasNext()) { AtlasEdge outgoingEdge = outgoingEdges.next(); + + if (checkForOffset(outgoingEdge, processVertex, atlasLineageOnDemandContext, ret)) { + continue; + } + isTimedOut = timeoutTracker.hasTimedOut(); if (isTimedOut) { handleHorizontalAndVerticalPagination(incomingEdge, !isInput, atlasLineageOnDemandContext, ret, depth, entitiesTraversed, visitedVertices, isTimedOut); @@ -429,10 +435,6 @@ private void traverseEdgesOnDemand(AtlasVertex datasetVertex, boolean isInput, i continue; } - if (checkForOffset(outgoingEdge, processVertex, atlasLineageOnDemandContext, ret)) { - continue; - } - boolean stopDatasetIteration = handleHorizontalAndVerticalPagination(outgoingEdge, isInput, atlasLineageOnDemandContext, ret, depth, entitiesTraversed, visitedVertices, isTimedOut); if (stopDatasetIteration) { String processGuid = AtlasGraphUtilsV2.getIdFromVertex(processVertex); From 2e12c7fb25b23a2a5e1c55156f3bdf7d39be97ac Mon Sep 17 00:00:00 2001 From: aarshi Date: Wed, 13 Nov 2024 11:28:39 +0530 Subject: [PATCH 10/12] Configure async indexsearch --- .../atlas/repository/graphdb/janus/AtlasElasticsearchQuery.java | 2 +- intg/src/main/java/org/apache/atlas/AtlasConfiguration.java | 1 + 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/graphdb/janus/src/main/java/org/apache/atlas/repository/graphdb/janus/AtlasElasticsearchQuery.java b/graphdb/janus/src/main/java/org/apache/atlas/repository/graphdb/janus/AtlasElasticsearchQuery.java index 9aa7cfe8ba..f862b643c4 100644 --- a/graphdb/janus/src/main/java/org/apache/atlas/repository/graphdb/janus/AtlasElasticsearchQuery.java +++ b/graphdb/janus/src/main/java/org/apache/atlas/repository/graphdb/janus/AtlasElasticsearchQuery.java @@ -132,7 +132,7 @@ private DirectIndexQueryResult runQueryWithLowLevelClient(SearchParams searchPar DirectIndexQueryResult result = null; try { - if(searchParams.isCallAsync()) { + if(searchParams.isCallAsync() || AtlasConfiguration.ENABLE_ASYNC_INDEXSEARCH.getBoolean()) { return performAsyncDirectIndexQuery(searchParams); } else{ String responseString = performDirectIndexQuery(searchParams.getQuery(), false); diff --git a/intg/src/main/java/org/apache/atlas/AtlasConfiguration.java b/intg/src/main/java/org/apache/atlas/AtlasConfiguration.java index 7fd081b9e3..df2bca7860 100644 --- a/intg/src/main/java/org/apache/atlas/AtlasConfiguration.java +++ b/intg/src/main/java/org/apache/atlas/AtlasConfiguration.java @@ -113,6 +113,7 @@ public enum AtlasConfiguration { HERACLES_API_SERVER_URL("atlas.heracles.api.service.url", "http://heracles-service.heracles.svc.cluster.local"), INDEXSEARCH_ASYNC_SEARCH_KEEP_ALIVE_TIME_IN_SECONDS("atlas.indexsearch.async.search.keep.alive.time.in.seconds", 300), + ENABLE_ASYNC_INDEXSEARCH("atlas.indexsearch.async.enable", false), ATLAS_INDEXSEARCH_QUERY_SIZE_MAX_LIMIT("atlas.indexsearch.query.size.max.limit", 100000), ATLAS_INDEXSEARCH_LIMIT_UTM_TAGS("atlas.indexsearch.limit.ignore.utm.tags", ""), ATLAS_INDEXSEARCH_ENABLE_API_LIMIT("atlas.indexsearch.enable.api.limit", false), From 7bba0eb8d17bc9ba5c7a4a5bf4e1b4d1e9ebc13c Mon Sep 17 00:00:00 2001 From: aarshi Date: Wed, 13 Nov 2024 12:07:11 +0530 Subject: [PATCH 11/12] enable async calls --- intg/src/main/java/org/apache/atlas/AtlasConfiguration.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/intg/src/main/java/org/apache/atlas/AtlasConfiguration.java b/intg/src/main/java/org/apache/atlas/AtlasConfiguration.java index df2bca7860..e4886f9812 100644 --- a/intg/src/main/java/org/apache/atlas/AtlasConfiguration.java +++ b/intg/src/main/java/org/apache/atlas/AtlasConfiguration.java @@ -113,7 +113,7 @@ public enum AtlasConfiguration { HERACLES_API_SERVER_URL("atlas.heracles.api.service.url", "http://heracles-service.heracles.svc.cluster.local"), INDEXSEARCH_ASYNC_SEARCH_KEEP_ALIVE_TIME_IN_SECONDS("atlas.indexsearch.async.search.keep.alive.time.in.seconds", 300), - ENABLE_ASYNC_INDEXSEARCH("atlas.indexsearch.async.enable", false), + ENABLE_ASYNC_INDEXSEARCH("atlas.indexsearch.async.enable", true), ATLAS_INDEXSEARCH_QUERY_SIZE_MAX_LIMIT("atlas.indexsearch.query.size.max.limit", 100000), ATLAS_INDEXSEARCH_LIMIT_UTM_TAGS("atlas.indexsearch.limit.ignore.utm.tags", ""), ATLAS_INDEXSEARCH_ENABLE_API_LIMIT("atlas.indexsearch.enable.api.limit", false), From 6cae188db004be84cd169cd42c3aff8677f0c94a Mon Sep 17 00:00:00 2001 From: hr2904 Date: Wed, 13 Nov 2024 16:23:14 +0530 Subject: [PATCH 12/12] ADded updateTime in delete flow for updatesBM --- .../atlas/repository/store/graph/v2/EntityGraphMapper.java | 1 + 1 file changed, 1 insertion(+) diff --git a/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/EntityGraphMapper.java b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/EntityGraphMapper.java index 5152bb2e5d..76f0be44cb 100644 --- a/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/EntityGraphMapper.java +++ b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/EntityGraphMapper.java @@ -877,6 +877,7 @@ public void removeBusinessAttributes(AtlasVertex entityVertex, AtlasEntityType e } if (MapUtils.isNotEmpty(updatedBusinessAttributes)) { + updateModificationMetadata(entityVertex); entityChangeNotifier.onBusinessAttributesUpdated(AtlasGraphUtilsV2.getIdFromVertex(entityVertex), updatedBusinessAttributes); }