handle errors
Project: http://git-wip-us.apache.org/repos/asf/usergrid/repo Commit: http://git-wip-us.apache.org/repos/asf/usergrid/commit/001a7e38 Tree: http://git-wip-us.apache.org/repos/asf/usergrid/tree/001a7e38 Diff: http://git-wip-us.apache.org/repos/asf/usergrid/diff/001a7e38 Branch: refs/heads/USERGRID-933 Commit: 001a7e385f3e96d9d6a43b99344aa128c64f87c0 Parents: 5680173 Author: Shawn Feldman <sfeld...@apache.org> Authored: Mon Sep 14 13:34:04 2015 -0600 Committer: Shawn Feldman <sfeld...@apache.org> Committed: Mon Sep 14 13:34:04 2015 -0600 ---------------------------------------------------------------------- .../rx/impl/AllApplicationsObservableImpl.java | 36 +++----- .../migration/MvccEntityDataMigrationImpl.java | 94 +++++++++++--------- .../model/entity/MapToEntityConverter.java | 4 +- 3 files changed, 66 insertions(+), 68 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/usergrid/blob/001a7e38/stack/core/src/main/java/org/apache/usergrid/corepersistence/rx/impl/AllApplicationsObservableImpl.java ---------------------------------------------------------------------- diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/rx/impl/AllApplicationsObservableImpl.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/rx/impl/AllApplicationsObservableImpl.java index 9c766a8..e54da0c 100644 --- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/rx/impl/AllApplicationsObservableImpl.java +++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/rx/impl/AllApplicationsObservableImpl.java @@ -93,37 +93,27 @@ public class AllApplicationsObservableImpl implements AllApplicationsObservable //we have app infos. For each of these app infos, we have to load the application itself Observable<ApplicationScope> appIds = gm.loadEdgesFromSource( new SimpleSearchByEdgeType( rootAppId, edgeType, Long.MAX_VALUE, SearchByEdgeType.Order.DESCENDING, - Optional.absent() ) ).flatMap( new Func1<Edge, Observable<ApplicationScope>>() { - @Override - public Observable<ApplicationScope> call( final Edge edge ) { - - //get the app info and load it - final Id appInfo = edge.getTargetNode(); - - return collectionManager.load( appInfo ) - //filter out null entities - .filter( new Func1<Entity, Boolean>() { - @Override - public Boolean call( final Entity entity ) { + Optional.absent() ) ).flatMap(edge -> { + + //get the app info and load it + final Id appInfo = edge.getTargetNode(); + + return collectionManager.load( appInfo ) + //filter out null entities + .filter(entity -> { if ( entity == null ) { logger.warn( "Encountered a null application info for id {}", appInfo ); return false; } return true; - } - } ) - //get the id from the entity - .map( new Func1<Entity, ApplicationScope>() { - - @Override - public ApplicationScope call( final Entity entity ) { + }) + //get the id from the entity + .map(entity -> { final UUID uuid =entity.getId().getUuid(); return getApplicationScope( uuid ); - } - } ); - } - } ); + }); + }); return Observable.merge( systemIds, appIds ); } http://git-wip-us.apache.org/repos/asf/usergrid/blob/001a7e38/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/migration/MvccEntityDataMigrationImpl.java ---------------------------------------------------------------------- diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/migration/MvccEntityDataMigrationImpl.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/migration/MvccEntityDataMigrationImpl.java index e825bbc..86fd701 100644 --- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/migration/MvccEntityDataMigrationImpl.java +++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/migration/MvccEntityDataMigrationImpl.java @@ -142,9 +142,13 @@ public class MvccEntityDataMigrationImpl implements DataMigration<EntityIdScope> EntityToSaveMessage> subscriber ) { while ( allVersions.hasNext() ) { - final EntityToSaveMessage message = - new EntityToSaveMessage( currentScope, allVersions.next() ); - subscriber.onNext( message ); + try { + final EntityToSaveMessage message = + new EntityToSaveMessage(currentScope, allVersions.next()); + subscriber.onNext(message); + }catch (Exception e){ + LOGGER.error("Failed to load entity " +entityIdScope.getId(),e); + } } subscriber.onCompleted(); @@ -159,62 +163,66 @@ public class MvccEntityDataMigrationImpl implements DataMigration<EntityIdScope> for ( EntityToSaveMessage message : entities ) { - final MutationBatch entityRewrite = migration.to.write( message.scope, message.entity ); + try { + final MutationBatch entityRewrite = migration.to.write(message.scope, message.entity); - //add to - // the - // total - // batch - totalBatch.mergeShallow( entityRewrite ); + //add to + // the + // total + // batch + totalBatch.mergeShallow(entityRewrite); - //write - // the - // unique values + //write + // the + // unique values - if ( !message.entity.getEntity().isPresent() ) { - return; - } + if (!message.entity.getEntity().isPresent()) { + return; + } - final Entity entity = message.entity.getEntity().get(); + final Entity entity = message.entity.getEntity().get(); - final Id entityId = entity.getId(); + final Id entityId = entity.getId(); - final UUID version = message.entity.getVersion(); + final UUID version = message.entity.getVersion(); - toSaveIds.add( entityId ); + toSaveIds.add(entityId); - // re-write the unique - // values - // but this - // time with - // no TTL so that cleanup can clean up - // older values - for (final Field field : EntityUtils.getUniqueFields( message.entity.getEntity().get() ) ) { + // re-write the unique + // values + // but this + // time with + // no TTL so that cleanup can clean up + // older values + for (final Field field : EntityUtils.getUniqueFields(message.entity.getEntity().get())) { - final UniqueValue written = new UniqueValueImpl( field, entityId, version ); + final UniqueValue written = new UniqueValueImpl(field, entityId, version); - final MutationBatch mb = uniqueValueSerializationStrategy.write( message.scope, written ); + final MutationBatch mb = uniqueValueSerializationStrategy.write(message.scope, written); - // merge into our - // existing mutation - // batch - totalBatch.mergeShallow( mb ); - } + // merge into our + // existing mutation + // batch + totalBatch.mergeShallow(mb); + } - //add all our log entries - final List<MvccLogEntry> logEntries = mvccLogEntrySerializationStrategy.load( message.scope, - message.entity.getId(), version, 1000 ); + //add all our log entries + final List<MvccLogEntry> logEntries = mvccLogEntrySerializationStrategy.load(message.scope, + message.entity.getId(), version, 1000); - /** - * Migrate the log entry to the new format - */ - for(final MvccLogEntry entry: logEntries){ - final MutationBatch mb = mvccLogEntrySerializationStrategy.write( message.scope, entry ); + /** + * Migrate the log entry to the new format + */ + for (final MvccLogEntry entry : logEntries) { + final MutationBatch mb = mvccLogEntrySerializationStrategy.write(message.scope, entry); - totalBatch.mergeShallow( mb ); + totalBatch.mergeShallow(mb); + } + }catch (Exception e){ + LOGGER.error("Failed to migrate entity "+ message.entity.getId().getUuid()+ " :: " + message.entity.getId().getType(),e); } @@ -232,7 +240,7 @@ public class MvccEntityDataMigrationImpl implements DataMigration<EntityIdScope> } - } ).subscribeOn( Schedulers.io() ); + } ).subscribeOn(Schedulers.io()); }, 10) ); http://git-wip-us.apache.org/repos/asf/usergrid/blob/001a7e38/stack/corepersistence/model/src/main/java/org/apache/usergrid/persistence/model/entity/MapToEntityConverter.java ---------------------------------------------------------------------- diff --git a/stack/corepersistence/model/src/main/java/org/apache/usergrid/persistence/model/entity/MapToEntityConverter.java b/stack/corepersistence/model/src/main/java/org/apache/usergrid/persistence/model/entity/MapToEntityConverter.java index dbd1b2a..c0db81b 100644 --- a/stack/corepersistence/model/src/main/java/org/apache/usergrid/persistence/model/entity/MapToEntityConverter.java +++ b/stack/corepersistence/model/src/main/java/org/apache/usergrid/persistence/model/entity/MapToEntityConverter.java @@ -238,9 +238,9 @@ public class MapToEntityConverter{ } else { //can't process non enties logger.warn( - "entity cannot process location values that don't have valid location{latitude,longitude} values" + "entity cannot process location values that don't have valid location{latitude,longitude} values, changing to generic object" ); - // entity.setField(new EntityObjectField(fieldName,fromMap( value, false))); // recursion + entity.setField(new EntityObjectField(fieldName,fromMap( value, false))); // recursion } } }
