You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@usergrid.apache.org by sn...@apache.org on 2015/02/10 16:48:38 UTC
[2/2] incubator-usergrid git commit: Change ImportServicesImpl to use
searchConnectedEntities() instead of getConnectedEntities() as a work around
to make ImportCollectionIT#testImportWithMultipleFiles pass.
Change ImportServicesImpl to use searchConnectedEntities() instead of getConnectedEntities() as a work around to make ImportCollectionIT#testImportWithMultipleFiles pass.
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/b8da17c1
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/b8da17c1
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/b8da17c1
Branch: refs/heads/two-dot-o-import
Commit: b8da17c1849d5c94c1bb86b476f3ae72ddb60a51
Parents: 73be355
Author: Dave Johnson <dm...@apigee.com>
Authored: Tue Feb 10 10:48:26 2015 -0500
Committer: Dave Johnson <dm...@apigee.com>
Committed: Tue Feb 10 10:48:26 2015 -0500
----------------------------------------------------------------------
.../management/importer/ImportServiceImpl.java | 95 +++++++-------------
.../importer/ImportConnectionsTest.java | 85 +++++++++++++++---
2 files changed, 103 insertions(+), 77 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/b8da17c1/stack/services/src/main/java/org/apache/usergrid/management/importer/ImportServiceImpl.java
----------------------------------------------------------------------
diff --git a/stack/services/src/main/java/org/apache/usergrid/management/importer/ImportServiceImpl.java b/stack/services/src/main/java/org/apache/usergrid/management/importer/ImportServiceImpl.java
index 11c1648..1f93ded 100644
--- a/stack/services/src/main/java/org/apache/usergrid/management/importer/ImportServiceImpl.java
+++ b/stack/services/src/main/java/org/apache/usergrid/management/importer/ImportServiceImpl.java
@@ -28,6 +28,7 @@ import org.apache.usergrid.persistence.*;
import org.apache.usergrid.persistence.entities.FileImport;
import org.apache.usergrid.persistence.entities.Import;
import org.apache.usergrid.persistence.entities.JobData;
+import org.apache.usergrid.persistence.index.query.Query;
import org.apache.usergrid.persistence.index.query.Query.Level;
import org.apache.usergrid.persistence.queue.QueueManager;
import org.apache.usergrid.persistence.queue.QueueManagerFactory;
@@ -63,42 +64,24 @@ public class ImportServiceImpl implements ImportService {
public static final String FILE_IMPORT_JOB_NAME = "fileImportJob";
public static final int HEARTBEAT_COUNT = 50;
-
private static final Logger logger = LoggerFactory.getLogger(ImportServiceImpl.class);
- //injected the Entity Manager Factory
+ int MAX_FILE_IMPORTS = 1000; // max number of file import jobs / import job
+
protected EntityManagerFactory emf;
- //dependency injection
private SchedulerService sch;
- private ServiceManagerFactory smf;
-
- //Dependency injection through spring
- private QueueManager qm;
-
- private QueueManagerFactory queueManagerFactory;
-
- //inject Management Service to access Organization Data
private ManagementService managementService;
+
private JsonFactory jsonFactory = new JsonFactory();
@PostConstruct
public void init(){
-
- //TODO: move this to a before or initialization method.
-
- //TODO: made queueName clearly defined.
- //smf = getApplicationContext().getBean(ServiceManagerFactory.class);
-
- String name = ImportQueueListener.QUEUE_NAME;
- QueueScopeFactory queueScopeFactory = CpSetup.getInjector().getInstance(QueueScopeFactory.class);
- QueueScope queueScope = queueScopeFactory.getScope(CpNamingUtils.MANAGEMENT_APPLICATION_ID, name);
- queueManagerFactory = CpSetup.getInjector().getInstance(QueueManagerFactory.class);
- qm = queueManagerFactory.getQueueManager(queueScope);
}
+
/**
* This schedules the main import Job.
*
@@ -217,19 +200,25 @@ public class ImportServiceImpl implements ImportService {
private int getConnectionCount( final Import importRoot ) {
try {
- EntityManager rootEm = emf.getEntityManager( CpNamingUtils.MANAGEMENT_APPLICATION_ID );
-
- Results entities = rootEm.getConnectedEntities( importRoot, "includes", null, Level.ALL_PROPERTIES );
- PagingResultsIterator itr = new PagingResultsIterator( entities );
-
- int count = 0;
- while ( itr.hasNext() ) {
- itr.next();
- count++;
- }
-
- return count;
+ EntityManager emMgmtApp = emf.getEntityManager( CpNamingUtils.MANAGEMENT_APPLICATION_ID );
+ Query query = Query.fromQL("select *");
+ query.setEntityType("file_import");
+ query.setConnectionType("includes");
+ query.setLimit(MAX_FILE_IMPORTS);
+
+ Results entities = emMgmtApp.searchConnectedEntities( importRoot, query );
+ return entities.size();
+
+ // see ImportConnectsTest()
+// Results entities = emMgmtApp.getConnectedEntities( importRoot, "includes", null, Level.ALL_PROPERTIES );
+// PagingResultsIterator itr = new PagingResultsIterator( entities );
+// int count = 0;
+// while ( itr.hasNext() ) {
+// itr.next();
+// count++;
+// }
+// return count;
}
catch ( Exception e ) {
logger.error( "application doesn't exist within the current context" );
@@ -485,8 +474,10 @@ public class ImportServiceImpl implements ImportService {
final int count = getConnectionCount(importEntity);
if ( count == fileJobs.size() ) {
+ logger.debug("Got ALL {} of {} expected connections", count, fileJobs.size());
done = true;
} else {
+ logger.debug("Got {} of {} expected connections. Waiting...", count, fileJobs.size());
Thread.sleep(1000);
}
}
@@ -591,36 +582,12 @@ public class ImportServiceImpl implements ImportService {
String randTag = RandomStringUtils.randomAlphanumeric(4);
logger.debug("{} Got importEntity {}", randTag, importEntity.getUuid() );
- Results entities = emManagementApp.getConnectedEntities(
- importEntity, "includes", "file_import", Level.ALL_PROPERTIES);
-
-
-// int retries = 0;
-// int maxRetries = 60;
-// Results entities = null;
-// boolean done = false;
-// while ( !done && retries++ < maxRetries ) {
-//
-// // get all file import job siblings of the current job we're working now
-// entities = emManagementApp.getConnectedEntities(
-// importEntity, "includes", "file_import", Level.ALL_PROPERTIES);
-//
-// if ( entities.size() == importEntity.getFileCount() ) {
-// logger.debug("{} got {} file_import entities, expected {} DONE!",
-// new Object[] { randTag, entities.size(), importEntity.getFileCount() });
-// done = true;
-//
-// } else {
-// logger.debug("{} got {} file_import entities, expected {} waiting... ",
-// new Object[] { randTag, entities.size(), importEntity.getFileCount() });
-// Thread.sleep(1000);
-// }
-// }
-//
-// if ( retries >= maxRetries ) {
-// throw new RuntimeException("Max retries was reached");
-// }
-
+ EntityManager emMgmtApp = emf.getEntityManager( CpNamingUtils.MANAGEMENT_APPLICATION_ID );
+ Query query = Query.fromQL("select *");
+ query.setEntityType("file_import");
+ query.setConnectionType("includes");
+ query.setLimit(MAX_FILE_IMPORTS);
+ Results entities = emMgmtApp.searchConnectedEntities(importEntity, query);
PagingResultsIterator itr = new PagingResultsIterator( entities );
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/b8da17c1/stack/services/src/test/java/org/apache/usergrid/management/importer/ImportConnectionsTest.java
----------------------------------------------------------------------
diff --git a/stack/services/src/test/java/org/apache/usergrid/management/importer/ImportConnectionsTest.java b/stack/services/src/test/java/org/apache/usergrid/management/importer/ImportConnectionsTest.java
index 8ce7361..dda5f51 100644
--- a/stack/services/src/test/java/org/apache/usergrid/management/importer/ImportConnectionsTest.java
+++ b/stack/services/src/test/java/org/apache/usergrid/management/importer/ImportConnectionsTest.java
@@ -55,10 +55,41 @@ public class ImportConnectionsTest {
@Rule
public NewOrgAppAdminRule newOrgAppAdminRule = new NewOrgAppAdminRule( setup );
+
+ @Test
+ @Ignore("Because getConnectedEntities() is broken")
+ public void testCreateAndCountConnectionsViaGet() throws Exception {
+
+ doTestCreateAndCountConnections(new ConnectionCounter() {
+ @Override
+ public int count(Import importEntity) {
+ return getConnectionCountViaGet(importEntity);
+ }
+ });
+ }
+
+
@Test
- public void testCreateAndSearchConnections() throws Exception {
+ public void testCreateAndCountConnectionsViaSearch() throws Exception {
+
+ doTestCreateAndCountConnections(new ConnectionCounter() {
+ @Override
+ public int count(Import importEntity) {
+ return getConnectionCountViaSearch(importEntity);
+ }
+ });
+ }
+
+
+ interface ConnectionCounter {
+ int count( Import importEntity );
+ }
+
- final int connectionCount = 10;
+ public void doTestCreateAndCountConnections(
+ ConnectionCounter counter) throws Exception {
+
+ final int connectionCount = 15;
EntityManager emMgmtApp = setup.getEmf()
.getEntityManager(CpNamingUtils.MANAGEMENT_APPLICATION_ID);
@@ -75,16 +106,17 @@ public class ImportConnectionsTest {
}
int retries = 0;
- int maxRetries = 60;
+ int maxRetries = 20;
boolean done = false;
+ int count = 0;
while ( !done && retries++ < maxRetries ) {
- final int count = getConnectionCount(importEntity);
+ count = counter.count( importEntity );
if ( count == connectionCount ) {
logger.debug("Count good!");
done = true;
} else {
- logger.debug("Waiting...");
+ logger.debug("Got {} of {} Waiting...", count, connectionCount );
Thread.sleep(1000);
}
}
@@ -92,27 +124,25 @@ public class ImportConnectionsTest {
throw new RuntimeException("Max retries was reached");
}
- assertEquals("did not get all connections",
- connectionCount, getConnectionCount( importEntity ));
+ assertEquals("did not get all connections", connectionCount, count);
}
- private int getConnectionCount( final Import importRoot ) {
+
+ private int getConnectionCountViaGet( final Import importRoot ) {
try {
- EntityManager rootEm = setup.getEmf()
+ EntityManager emMgmtApp = setup.getEmf()
.getEntityManager(CpNamingUtils.MANAGEMENT_APPLICATION_ID );
- Results entities = rootEm.getConnectedEntities(
+ Results entities = emMgmtApp.getConnectedEntities(
importRoot, "includes", null, Query.Level.ALL_PROPERTIES );
- PagingResultsIterator itr = new PagingResultsIterator( entities );
+ PagingResultsIterator itr = new PagingResultsIterator( entities );
int count = 0;
-
while ( itr.hasNext() ) {
itr.next();
count++;
}
-
return count;
}
catch ( Exception e ) {
@@ -120,5 +150,34 @@ public class ImportConnectionsTest {
throw new RuntimeException( e );
}
}
+
+
+ private int getConnectionCountViaSearch( final Import importRoot ) {
+
+ try {
+ EntityManager emMgmtApp = setup.getEmf()
+ .getEntityManager(CpNamingUtils.MANAGEMENT_APPLICATION_ID );
+
+ Query query = Query.fromQL("select *");
+ query.setEntityType("file_import");
+ query.setConnectionType("includes");
+ query.setLimit(10000);
+
+ Results entities = emMgmtApp.searchConnectedEntities( importRoot, query );
+ return entities.size();
+
+// PagingResultsIterator itr = new PagingResultsIterator( entities );
+// int count = 0;
+// while ( itr.hasNext() ) {
+// itr.next();
+// count++;
+// }
+// return count;
+ }
+ catch ( Exception e ) {
+ logger.error( "application doesn't exist within the current context" );
+ throw new RuntimeException( e );
+ }
+ }
}