You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@usergrid.apache.org by sf...@apache.org on 2015/02/16 23:01:23 UTC
[16/50] incubator-usergrid git commit: More improvements to error
handling and reporting and using the tracker correctly,
plus improvements to ImportCollectionIT.testImportBadJson().
More improvements to error handling and reporting and using the tracker correctly, plus improvements to ImportCollectionIT.testImportBadJson().
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/ee864092
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/ee864092
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/ee864092
Branch: refs/heads/USERGRID-365
Commit: ee8640928dc05c473bd31c286280d4aa4af194a5
Parents: a00af86
Author: Dave Johnson <dm...@apigee.com>
Authored: Thu Feb 12 10:25:01 2015 -0500
Committer: Dave Johnson <dm...@apigee.com>
Committed: Thu Feb 12 10:25:01 2015 -0500
----------------------------------------------------------------------
.../management/importer/ImportServiceImpl.java | 94 ++++++++++----------
.../management/importer/ImportCollectionIT.java | 61 +++++++++----
2 files changed, 88 insertions(+), 67 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/ee864092/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 434d62f..bba12f1 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
@@ -684,7 +684,7 @@ public class ImportServiceImpl implements ImportService {
EntityManager emManagementApp = emf.getEntityManager( CpNamingUtils.MANAGEMENT_APPLICATION_ID );
- // get the file import entity
+ // get the file import entity
FileImport fileImport;
try {
@@ -694,6 +694,9 @@ public class ImportServiceImpl implements ImportService {
return;
}
+ // tracker flushes every 100 entities
+ final FileImportTracker tracker = new FileImportTracker( emf, fileImport, 100 );
+
String fileName = jobExecution.getJobData().getProperty("File").toString();
UUID targetAppId = (UUID) jobExecution.getJobData().getProperty("applicationId");
@@ -712,20 +715,12 @@ public class ImportServiceImpl implements ImportService {
emManagementApp.update(fileImport);
if ( emManagementApp.get( targetAppId ) == null ) {
- fileImport.setState( FileImport.State.FAILED );
- fileImport.setErrorMessage("Application " + targetAppId + " does not exist");
- emManagementApp.update(fileImport);
+ tracker.fatal("Application " + targetAppId + " does not exist");
return;
}
} catch (Exception e) {
- fileImport.setState( FileImport.State.FAILED );
- fileImport.setErrorMessage("Application " + targetAppId + " does not exist");
- try {
- emManagementApp.update( fileImport );
- } catch (Exception e1) {
- logger.error("Error updating fileImport to set state of file import", e1);
- }
+ tracker.fatal("Application " + targetAppId + " does not exist");
}
EntityManager targetEm = emf.getEntityManager( targetAppId );
@@ -741,14 +736,7 @@ public class ImportServiceImpl implements ImportService {
s3Import = new S3ImportImpl();
}
} catch (Exception e) {
- logger.error("doImport(): Error creating S3Import", e);
- fileImport.setErrorMessage(e.getMessage());
- fileImport.setState( FileImport.State.FAILED );
- try {
- emManagementApp.update(fileImport);
- } catch (Exception e1) {
- logger.error("Error updating file import with error information", e1);
- }
+ tracker.fatal("Unable to connect to S3 bucket, error: " + e.getMessage());
return;
}
@@ -756,31 +744,17 @@ public class ImportServiceImpl implements ImportService {
downloadedFile = s3Import.copyFileFromBucket(
fileName, bucketName, accessId, secretKey );
} catch (Exception e) {
- logger.error("Error downloading file " + fileName, e);
- fileImport.setErrorMessage("Error downloading file " + fileName + ": " + e.getMessage());
- fileImport.setState(FileImport.State.FAILED);
- try {
- emManagementApp.update(fileImport);
- } catch (Exception e1) {
- logger.error("Error updating file import with error information", e1);
- }
+ tracker.fatal("Error downloading file " + fileName + ": " + e.getMessage());
}
// parse JSON data, create Entities and Connections from import data
try {
parseEntitiesAndConnectionsFromJson(
- jobExecution, downloadedFile, targetEm, emManagementApp, fileImport);
+ jobExecution, downloadedFile, targetEm, emManagementApp, fileImport, tracker);
} catch (Exception e) {
- logger.error("Error importing file " + fileName, e);
- fileImport.setErrorMessage("Error downloading file " + fileName + ": " + e.getMessage());
- fileImport.setState( FileImport.State.FAILED );
- try {
- emManagementApp.update(fileImport);
- } catch (Exception e1) {
- logger.error("Error updating file import with error information", e1);
- }
+ tracker.fatal("Error importing file " + fileName + ": " + e.getMessage());
}
// mark ImportJob FINISHED but only if all other FileImportJobs are complete
@@ -804,12 +778,14 @@ public class ImportServiceImpl implements ImportService {
EntityManager emMgmtApp = emf.getEntityManager( CpNamingUtils.MANAGEMENT_APPLICATION_ID );
-
Query query = new Query();
query.setEntityType(Schema.getDefaultSchema().getEntityType( FileImport.class ));
query.setConnectionType( IMPORT_FILE_INCLUDES_CONNECTION );
query.setLimit(MAX_FILE_IMPORTS);
+ // TODO: better way to wait for ES indexes to catch up
+ try { Thread.sleep(5000); } catch ( Exception intentionallyIgnored ) {}
+
Results entities = emMgmtApp.searchConnectedEntities(importEntity, query);
PagingResultsIterator itr = new PagingResultsIterator(entities);
@@ -885,11 +861,12 @@ public class ImportServiceImpl implements ImportService {
final File file,
final EntityManager em,
final EntityManager rootEm,
- final FileImport fileImport) throws Exception {
+ final FileImport fileImport,
+ final FileImportTracker tracker) throws Exception {
// tracker flushes every 100 entities
- final FileImportTracker tracker = new FileImportTracker( emf, fileImport, 100 );
+ //final FileImportTracker tracker = new FileImportTracker( emf, fileImport, 100 );
// function to execute for each write event
final Action1<WriteEvent> doWork = new Action1<WriteEvent>() {
@@ -945,7 +922,12 @@ public class ImportServiceImpl implements ImportService {
jp.close();
- logger.debug("\n\nparseEntitiesAndConnectionsFromJson(): Wrote entities\n");
+ if ( FileImport.State.FAILED.equals( fileImport.getState() ) ) {
+ logger.debug("\n\nFailed to completely write entities, skipping second phase. File: {}\n",
+ fileImport.getFileName());
+ return;
+ }
+ logger.debug("\n\nWrote entities. File: {}\n", fileImport.getFileName() );
// SECOND PASS: import all connections and dictionaries
@@ -984,9 +966,20 @@ public class ImportServiceImpl implements ImportService {
logger.debug("\n\nparseEntitiesAndConnectionsFromJson(): Wrote others for file {}\n",
fileImport.getFileName());
+ if ( FileImport.State.FAILED.equals( fileImport.getState() ) ) {
+ logger.debug("\n\nparseEntitiesAndConnectionsFromJson(): failed to completely write entities\n");
+ return;
+ }
// flush the job statistics
tracker.complete();
+
+ if ( FileImport.State.FAILED.equals( fileImport.getState() ) ) {
+ logger.debug("\n\nFailed to completely wrote connections and dictionaries. File: {}\n",
+ fileImport.getFileName());
+ return;
+ }
+ logger.debug("\n\nWrote connections and dictionaries. File: {}\n", fileImport.getFileName() );
}
@@ -1249,15 +1242,17 @@ public class ImportServiceImpl implements ImportService {
logger.debug("process(): done parsing JSON");
} catch (Exception e) {
- // skip illegal entity UUID and go to next one
- fileImport.setErrorMessage(e.getMessage());
- try {
- rootEm.update(fileImport);
- } catch (Exception ex) {
- logger.error("Error updating file import record", ex);
- }
+
+ tracker.fatal("Failed to import file" + fileImport.getFileName() + " error " + e.getMessage());
+
if ( subscriber != null ) {
- subscriber.onError(e);
+
+ // don't need to blow up here, we handled the problem
+ // if we blow up we may prevent in-flight entities from being written
+ // subscriber.onError(e);
+
+ // but we are done reading entities
+ subscriber.onCompleted();
}
}
}
@@ -1265,8 +1260,11 @@ public class ImportServiceImpl implements ImportService {
private void processWriteEvent( final Subscriber<? super WriteEvent> subscriber, WriteEvent writeEvent ) {
if ( subscriber == null ) {
+
+ // this logic makes it easy to remove Rx for debugging purposes
// no Rx, just do it
writeEvent.doWrite(em, fileImport, tracker);
+
} else {
subscriber.onNext( writeEvent );
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/ee864092/stack/services/src/test/java/org/apache/usergrid/management/importer/ImportCollectionIT.java
----------------------------------------------------------------------
diff --git a/stack/services/src/test/java/org/apache/usergrid/management/importer/ImportCollectionIT.java b/stack/services/src/test/java/org/apache/usergrid/management/importer/ImportCollectionIT.java
index 41d264c..fde0d8b 100644
--- a/stack/services/src/test/java/org/apache/usergrid/management/importer/ImportCollectionIT.java
+++ b/stack/services/src/test/java/org/apache/usergrid/management/importer/ImportCollectionIT.java
@@ -35,6 +35,7 @@ import org.apache.usergrid.management.OrganizationInfo;
import org.apache.usergrid.management.UserInfo;
import org.apache.usergrid.management.export.ExportService;
import org.apache.usergrid.persistence.*;
+import org.apache.usergrid.persistence.entities.FileImport;
import org.apache.usergrid.persistence.entities.Import;
import org.apache.usergrid.persistence.index.impl.ElasticSearchResource;
import org.apache.usergrid.persistence.index.query.Query;
@@ -342,37 +343,49 @@ public class ImportCollectionIT {
* TODO: Test that importing bad JSON will result in an informative error message.
*/
@Test
- public void testImportBadJson() throws Exception{
- // import from a bad JSON file
+ public void testImportBadJson() throws Exception {
deleteBucket();
- //list out all the files in the resource directory you want uploaded
- List<String> filenames = new ArrayList<>( 1 );
+ // export and upload a bad JSON file to the S3 bucket
+ List<String> filenames = new ArrayList<>( 1 );
filenames.add( "testimport-bad-json.json");
- // create 10 applications each with collection of 10 things, export all to S3
S3Upload s3Upload = new S3Upload();
s3Upload.copyToS3(
System.getProperty(SDKGlobalConfiguration.ACCESS_KEY_ENV_VAR),
System.getProperty(SDKGlobalConfiguration.SECRET_KEY_ENV_VAR),
bucketName, filenames );
- // import all those exports from S3 into the default test application
+ // import bad JSON from from the S3 bucket
final EntityManager emDefaultApp = setup.getEmf().getEntityManager( applicationId );
- importCollection( emDefaultApp );
+ UUID importId = importCollection( emDefaultApp );
- // we should now have 100 Entities in the default app
+
+ // check that we got an informative error message back
List<Entity> importedThings = emDefaultApp.getCollection(
emDefaultApp.getApplicationId(), "things", null, Level.ALL_PROPERTIES).getEntities();
- assertTrue( importedThings.isEmpty() );
- // assertEquals( , importedThings.size() );
+ assertTrue("No entities should have been imported", importedThings.isEmpty());
+
+ ImportService importService = setup.getImportService();
+ Results results = importService.getFileImports( applicationId, importId, null, null );
+
+ assertEquals( "There is one", 1, results.size() );
+
+ assertEquals( "Entity is FileImport object",
+ FileImport.class, results.getEntity().getClass() );
- // check that error message indicates JSON parsing error
+ FileImport fileImport = (FileImport)results.getEntity();
+
+ assertEquals( "File name is correct",
+ "testimport-bad-json.json", fileImport.getFileName());
+
+ assertTrue( "Error message is correct",
+ fileImport.getErrorMessage().startsWith("Unexpected character ('<' (code 60))"));
}
@Test
@@ -432,7 +445,7 @@ public class ImportCollectionIT {
//---------------------------------------------------------------------------------------------
- private void importCollection(final EntityManager em ) throws Exception {
+ private UUID importCollection(final EntityManager em ) throws Exception {
logger.debug("\n\nImport into new app {}\n", em.getApplication().getName() );
@@ -454,19 +467,25 @@ public class ImportCollectionIT {
}} );
}} );
-
-
- int maxRetries = 60;
+ int maxRetries = 30;
int retries = 0;
- while ( !importService.getState( importEntity.getUuid() ).equals( "FINISHED" )
- && !importService.getState( importEntity.getUuid() ).equals( "FAILED" )
+ Import.State state = importService.getState(importEntity.getUuid());
+ while ( !state.equals( Import.State.FINISHED )
+ && !state.equals( Import.State.FAILED )
&& retries++ < maxRetries ) {
- logger.debug("Waiting for import...");
+ logger.debug("Waiting for import ({}) ...", state.toString());
Thread.sleep(1000);
+
+ state = importService.getState(importEntity.getUuid());
}
+ if ( retries >= maxRetries ) {
+ throw new RuntimeException("Max retries reached");
+ }
em.refreshIndex();
+
+ return importEntity.getUuid();
}
@@ -499,12 +518,16 @@ public class ImportCollectionIT {
}});
}});
- int maxRetries = 60;
+ int maxRetries = 30;
int retries = 0;
while ( !exportService.getState( exportUUID ).equals( "FINISHED" ) && retries++ < maxRetries ) {
logger.debug("Waiting for export...");
Thread.sleep(1000);
}
+
+ if ( retries >= maxRetries ) {
+ throw new RuntimeException("Max retries reached");
+ }
}