You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@atlas.apache.org by am...@apache.org on 2020/05/15 19:03:56 UTC
[atlas] branch master updated (d8cb94a -> 3de30f5)
This is an automated email from the ASF dual-hosted git repository.
amestry pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/atlas.git.
from d8cb94a ATLAS-3800: Add 'accountId' attribute for AWS S3 v2 bucket model (aws_s3_v2_bucket)
new 935d0ab ATLAS-3796: Status Reporter display fix.
new 214f2a7 ATLAS-3798: Import percentage display fix.
new 3de30f5 ATLAS-3799: EntityConumer only adds entity GUIDs that are added when they were produced.
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../java/org/apache/atlas/pc/StatusReporter.java | 10 ++++++++-
.../org/apache/atlas/pc/StatusReporterTest.java | 3 +++
.../store/graph/v2/BulkImporterImpl.java | 4 ++--
.../graph/v2/bulkimport/pc/EntityConsumer.java | 25 +++++-----------------
.../store/graph/v2/BulkImportPercentTest.java | 15 ++++++++++---
5 files changed, 31 insertions(+), 26 deletions(-)
[atlas] 02/03: ATLAS-3798: Import percentage display fix.
Posted by am...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
amestry pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/atlas.git
commit 214f2a760a1a4c8c3a83b5afaa42fdc36055ff98
Author: Ashutosh Mestry <am...@cloudera.com>
AuthorDate: Fri May 15 11:50:33 2020 -0700
ATLAS-3798: Import percentage display fix.
---
.../atlas/repository/store/graph/v2/BulkImporterImpl.java | 4 ++--
.../repository/store/graph/v2/BulkImportPercentTest.java | 15 ++++++++++++---
2 files changed, 14 insertions(+), 5 deletions(-)
diff --git a/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/BulkImporterImpl.java b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/BulkImporterImpl.java
index 9898fe8..8e17fd4 100644
--- a/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/BulkImporterImpl.java
+++ b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/BulkImporterImpl.java
@@ -89,11 +89,11 @@ public class BulkImporterImpl implements BulkImporter {
}
@VisibleForTesting
- public static float updateImportProgress(Logger log, int currentIndex, int streamSize, float currentPercent, String additionalInfo) {
+ public static float updateImportProgress(Logger log, long currentIndex, long streamSize, float currentPercent, String additionalInfo) {
final double tolerance = 0.000001;
final int MAX_PERCENT = 100;
- int maxSize = (currentIndex <= streamSize) ? streamSize : currentIndex;
+ long maxSize = (currentIndex <= streamSize) ? streamSize : currentIndex;
if (maxSize <= 0) {
return currentPercent;
}
diff --git a/repository/src/test/java/org/apache/atlas/repository/store/graph/v2/BulkImportPercentTest.java b/repository/src/test/java/org/apache/atlas/repository/store/graph/v2/BulkImportPercentTest.java
index 2fa4c91..1ae98ce 100644
--- a/repository/src/test/java/org/apache/atlas/repository/store/graph/v2/BulkImportPercentTest.java
+++ b/repository/src/test/java/org/apache/atlas/repository/store/graph/v2/BulkImportPercentTest.java
@@ -39,7 +39,7 @@ public class BulkImportPercentTest {
private List<Integer> percentHolder;
private Logger log;
- public void setupPercentHolder(int max) {
+ public void setupPercentHolder(long max) {
percentHolder = new ArrayList<>();
}
@@ -127,6 +127,15 @@ public class BulkImportPercentTest {
}
@Test
+ public void percentTest_Equal100M() throws Exception {
+ long streamSize = 100000000;
+ double[] expected = fillPercentHolderWith100();
+
+ runWithSize(streamSize);
+ assertEqualsForPercentHolder(expected);
+ }
+
+ @Test
public void percentTest_Equal4323() throws Exception {
int streamSize = 4323;
@@ -153,7 +162,7 @@ public class BulkImportPercentTest {
assertTrue((f - MAX_PERCENT_FLOAT) <= 0.0001);
}
- private void runWithSize(int streamSize) throws Exception {
+ private void runWithSize(long streamSize) throws Exception {
float currentPercent = 0;
setupPercentHolder(streamSize);
for (int currentIndex = 0; currentIndex < streamSize; currentIndex++) {
@@ -161,7 +170,7 @@ public class BulkImportPercentTest {
}
}
- private float invokeBulkImportProgress(int currentIndex, int streamSize, float currentPercent) throws Exception {
+ private float invokeBulkImportProgress(int currentIndex, long streamSize, float currentPercent) throws Exception {
return BulkImporterImpl.updateImportProgress(log, currentIndex, streamSize, currentPercent, "additional info");
}
[atlas] 03/03: ATLAS-3799: EntityConumer only adds entity GUIDs
that are added when they were produced.
Posted by am...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
amestry pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/atlas.git
commit 3de30f55d808df6245f60fe01fed4b6cf894a31d
Author: Ashutosh Mestry <am...@cloudera.com>
AuthorDate: Fri May 15 11:56:23 2020 -0700
ATLAS-3799: EntityConumer only adds entity GUIDs that are added when they were produced.
---
.../graph/v2/bulkimport/pc/EntityConsumer.java | 25 +++++-----------------
1 file changed, 5 insertions(+), 20 deletions(-)
diff --git a/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/bulkimport/pc/EntityConsumer.java b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/bulkimport/pc/EntityConsumer.java
index 7e4ae72..b73988f 100644
--- a/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/bulkimport/pc/EntityConsumer.java
+++ b/repository/src/main/java/org/apache/atlas/repository/store/graph/v2/bulkimport/pc/EntityConsumer.java
@@ -59,7 +59,7 @@ public class EntityConsumer extends WorkItemConsumer<AtlasEntity.AtlasEntityWith
private final EntityGraphRetriever entityRetrieverBulk;
private List<AtlasEntity.AtlasEntityWithExtInfo> entityBuffer = new ArrayList<>();
- private List<EntityMutationResponse> localResults = new ArrayList<>();
+ private List<String> localResults = new ArrayList<>();
public EntityConsumer(AtlasTypeRegistry typeRegistry,
AtlasGraph atlasGraph, AtlasEntityStore entityStore,
@@ -119,7 +119,7 @@ public class EntityConsumer extends WorkItemConsumer<AtlasEntity.AtlasEntityWith
private void importUsingBulkEntityStore(AtlasEntity.AtlasEntityWithExtInfo entityWithExtInfo) throws AtlasBaseException {
EntityStream oneEntityStream = new AtlasEntityStreamForImport(entityWithExtInfo, null);
EntityMutationResponse result = entityStoreBulk.createOrUpdateForImportNoCommit(oneEntityStream);
- localResults.add(result);
+ localResults.add(entityWithExtInfo.getEntity().getGuid());
entityBuffer.add(entityWithExtInfo);
}
@@ -133,9 +133,9 @@ public class EntityConsumer extends WorkItemConsumer<AtlasEntity.AtlasEntityWith
try {
LOG.info("Regular: EntityStore: {}: Starting...", this.counter.get());
AtlasEntityStreamForImport oneEntityStream = new AtlasEntityStreamForImport(entityWithExtInfo, null);
- EntityMutationResponse result = this.entityStore.createOrUpdateForImportNoCommit(oneEntityStream);
+ this.entityStore.createOrUpdateForImportNoCommit(oneEntityStream);
atlasGraph.commit();
- localResults.add(result);
+ localResults.add(entityWithExtInfo.getEntity().getGuid());
dispatchResults();
} catch (Exception e) {
atlasGraph.rollback();
@@ -244,12 +244,7 @@ public class EntityConsumer extends WorkItemConsumer<AtlasEntity.AtlasEntityWith
}
private void dispatchResults() {
- localResults.stream().forEach(x -> {
- addResultsFromResponse(x.getCreatedEntities());
- addResultsFromResponse(x.getUpdatedEntities());
- addResultsFromResponse(x.getDeletedEntities());
- });
-
+ localResults.stream().forEach(x -> addResult(x));
clear();
}
@@ -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() {
localResults.clear();
entityBuffer.clear();
[atlas] 01/03: ATLAS-3796: Status Reporter display fix.
Posted by am...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
amestry pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/atlas.git
commit 935d0ab6465775e16db62896375d96f6a567b884
Author: Ashutosh Mestry <am...@cloudera.com>
AuthorDate: Fri May 15 11:49:51 2020 -0700
ATLAS-3796: Status Reporter display fix.
---
intg/src/main/java/org/apache/atlas/pc/StatusReporter.java | 10 +++++++++-
intg/src/test/java/org/apache/atlas/pc/StatusReporterTest.java | 3 +++
2 files changed, 12 insertions(+), 1 deletion(-)
diff --git a/intg/src/main/java/org/apache/atlas/pc/StatusReporter.java b/intg/src/main/java/org/apache/atlas/pc/StatusReporter.java
index 7baf973..507e842 100644
--- a/intg/src/main/java/org/apache/atlas/pc/StatusReporter.java
+++ b/intg/src/main/java/org/apache/atlas/pc/StatusReporter.java
@@ -98,7 +98,7 @@ public class StatusReporter<T, U> {
private U acknowledged(Map.Entry<T, U> lookFor) {
U ack = lookFor.getValue();
producedItems.remove(lookFor.getKey());
- processedSet.remove(lookFor);
+ processedSet.remove(lookFor.getKey());
return ack;
}
@@ -107,4 +107,12 @@ public class StatusReporter<T, U> {
lastAck = System.currentTimeMillis();
return b;
}
+
+ public int getProducedCount() {
+ return this.producedItems.size();
+ }
+
+ public int getProcessedCount() {
+ return this.processedSet.size();
+ }
}
diff --git a/intg/src/test/java/org/apache/atlas/pc/StatusReporterTest.java b/intg/src/test/java/org/apache/atlas/pc/StatusReporterTest.java
index 45bdbb0..a0c399b 100644
--- a/intg/src/test/java/org/apache/atlas/pc/StatusReporterTest.java
+++ b/intg/src/test/java/org/apache/atlas/pc/StatusReporterTest.java
@@ -82,6 +82,9 @@ public class StatusReporterTest {
extractResults(wi, statusReporter);
assertEquals(statusReporter.ack().intValue(), (maxItems - 1));
wi.shutdown();
+
+ assertEquals(statusReporter.getProducedCount(), 0);
+ assertEquals(statusReporter.getProcessedCount(), 0);
}
private void extractResults(WorkItemManager<Integer, WorkItemConsumer> wi, StatusReporter<Integer, Integer> statusReporter) {