usergrid-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sfeld...@apache.org
Subject [36/50] [abbrv] incubator-usergrid git commit: Merge branch 'two-dot-o' into USERGRID-405
Date Thu, 19 Mar 2015 23:26:53 GMT
Merge branch 'two-dot-o' into USERGRID-405

Conflicts:
	stack/core/src/main/java/org/apache/usergrid/corepersistence/CoreModule.java
	stack/core/src/main/java/org/apache/usergrid/corepersistence/events/EntityVersionDeletedHandler.java
	stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/EntityCollectionManager.java
	stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/guice/CollectionModule.java
	stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityCollectionManagerFactoryImpl.java
	stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityCollectionManagerImpl.java
	stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityDeletedTask.java
	stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityVersionCleanupTask.java
	stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/UniqueValueSerializationStrategy.java
	stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/UniqueValueSerializationStrategyImpl.java
	stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/util/EntityUtils.java
	stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/impl/EntityVersionCleanupTaskTest.java
	stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/GraphManagerImpl.java
	stack/corepersistence/map/src/main/java/org/apache/usergrid/persistence/map/MapManager.java
	stack/corepersistence/queryindex/src/test/java/org/apache/usergrid/persistence/index/guice/TestIndexModule.java


Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/7b5d2224
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/7b5d2224
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/7b5d2224

Branch: refs/heads/USERGRID-480
Commit: 7b5d2224512f0043336acbd9b9323e1e32edf888
Parents: 5c7a5f8 b53cb07
Author: Todd Nine <tnine@apigee.com>
Authored: Tue Mar 17 11:36:08 2015 -0600
Committer: Todd Nine <tnine@apigee.com>
Committed: Tue Mar 17 11:36:08 2015 -0600

----------------------------------------------------------------------
 .../main/dist/init_instance/init_rest_server.sh |   2 +
 .../dist/init_instance/install_elasticsearch.sh |  22 +-
 .../main/groovy/configure_elasticsearch.groovy  |  80 ++--
 .../src/main/groovy/configure_usergrid.groovy   |  12 +-
 stack/awscluster/ugcluster-cf.json              | 466 +++++++++++++------
 stack/core/pom.xml                              |  10 -
 .../batch/service/JobSchedulerService.java      |   5 -
 .../usergrid/corepersistence/CoreModule.java    |  36 +-
 .../corepersistence/CpEntityManager.java        |  64 ++-
 .../corepersistence/CpEntityManagerFactory.java |   4 -
 .../corepersistence/CpRelationManager.java      |  14 -
 .../events/EntityDeletedHandler.java            |  58 ++-
 .../events/EntityVersionCreatedHandler.java     |  60 ++-
 .../events/EntityVersionDeletedHandler.java     |  53 ++-
 .../results/FilteringLoader.java                |   2 +-
 .../usergrid/persistence/EntityManager.java     |  10 +-
 .../persistence/EntityManagerFactory.java       |   2 -
 .../cassandra/EntityManagerFactoryImpl.java     |   3 -
 .../cassandra/EntityManagerImpl.java            |  31 +-
 .../cassandra/RelationManagerImpl.java          |  58 +--
 .../cassandra/index/ConnectedIndexScanner.java  |   2 -
 .../cassandra/index/IndexBucketScanner.java     |   2 -
 .../corepersistence/StaleIndexCleanupTest.java  |  38 +-
 .../usergrid/persistence/EntityManagerIT.java   |   1 +
 .../PerformanceEntityRebuildIndexTest.java      |   2 +
 .../collection/EntityCollectionManager.java     |  20 +-
 .../collection/EntityDeletedFactory.java        |  34 --
 .../collection/EntityVersionCleanupFactory.java |  35 --
 .../collection/EntityVersionCreatedFactory.java |  31 --
 .../persistence/collection/FieldSet.java        |  48 ++
 .../cache/CachedEntityCollectionManager.java    |  11 +-
 .../collection/guice/CollectionModule.java      |  11 +-
 .../EntityCollectionManagerFactoryImpl.java     |  32 +-
 .../impl/EntityCollectionManagerImpl.java       | 251 ++++++++--
 .../collection/impl/EntityDeletedTask.java      |   9 +-
 .../impl/EntityVersionCleanupTask.java          |  34 +-
 .../impl/EntityVersionTaskFactory.java          |  65 +++
 .../mvcc/stage/write/WriteUniqueVerify.java     | 140 +++---
 .../UniqueValueSerializationStrategy.java       |  12 +
 .../serialization/impl/MutableFieldSet.java     |  63 +++
 .../UniqueValueSerializationStrategyImpl.java   |  27 +-
 .../collection/util/EntityUtils.java            |   4 +-
 .../collection/EntityCollectionManagerIT.java   |  68 +++
 .../mvcc/stage/write/WriteUniqueVerifyTest.java |   6 +-
 .../core/astyanax/CassandraConfig.java          |   6 +
 .../core/astyanax/CassandraConfigImpl.java      |   8 +-
 .../persistence/core/astyanax/CassandraFig.java |   9 +-
 .../persistence/core/future/BetterFuture.java   |  43 +-
 .../core/metrics/MetricsFactory.java            |   9 +
 .../core/metrics/MetricsFactoryImpl.java        | 121 +++--
 .../core/astyanax/ColumnNameIteratorTest.java   |   7 +-
 .../MultiKeyColumnNameIteratorTest.java         |   7 +-
 .../astyanax/MultiRowColumnIteratorTest.java    |   7 +-
 stack/corepersistence/graph/pom.xml             |  28 +-
 .../graph/impl/GraphManagerImpl.java            | 288 ++++++++++--
 .../usergrid/persistence/map/MapManager.java    |  14 +-
 .../persistence/map/impl/MapManagerImpl.java    |   8 +
 .../persistence/map/impl/MapSerialization.java  |   9 +
 .../map/impl/MapSerializationImpl.java          | 130 +++++-
 .../persistence/map/MapManagerTest.java         |  49 +-
 stack/corepersistence/queryindex/pom.xml        |  17 +-
 .../usergrid/persistence/index/EntityIndex.java |  26 +-
 .../persistence/index/IndexBufferConsumer.java  |  11 +
 .../persistence/index/IndexBufferProducer.java  |   1 -
 .../usergrid/persistence/index/IndexFig.java    |  73 ++-
 .../index/IndexOperationMessage.java            | 115 ++++-
 .../persistence/index/guice/IndexModule.java    |   7 +
 .../persistence/index/guice/QueueProvider.java  | 116 +++++
 .../persistence/index/impl/BatchRequest.java    |  41 ++
 .../persistence/index/impl/BufferQueue.java     |  68 +++
 .../index/impl/BufferQueueInMemoryImpl.java     | 108 +++++
 .../index/impl/BufferQueueSQSImpl.java          | 307 ++++++++++++
 .../persistence/index/impl/DeIndexRequest.java  | 115 +++++
 .../index/impl/EsEntityIndexBatchImpl.java      |  50 +-
 .../index/impl/EsEntityIndexImpl.java           | 176 ++-----
 .../index/impl/EsIndexBufferConsumerImpl.java   | 286 ++++++++----
 .../index/impl/EsIndexBufferProducerImpl.java   |  16 +-
 .../persistence/index/impl/EsIndexCache.java    | 138 +++---
 .../persistence/index/impl/IndexRequest.java    | 125 +++++
 .../index/guice/TestIndexModule.java            |   3 +-
 .../index/impl/BufferQueueSQSImplTest.java      | 169 +++++++
 .../impl/EntityConnectionIndexImplTest.java     |   5 +-
 .../persistence/index/impl/EntityIndexTest.java |  20 +-
 .../persistence/index/impl/EsTestUtils.java     |  48 --
 .../persistence/queue/QueueManager.java         |   4 +-
 .../usergrid/persistence/queue/QueueScope.java  |   2 +-
 .../persistence/queue/QueueScopeFactory.java    |  34 --
 .../persistence/queue/guice/QueueModule.java    |  17 +-
 .../queue/impl/QueueScopeFactoryImpl.java       |  48 --
 .../persistence/queue/impl/QueueScopeImpl.java  |  27 +-
 .../queue/impl/SQSQueueManagerImpl.java         | 214 +++++----
 .../persistence/queue/NoAWSCredsRule.java       |  98 ++++
 .../persistence/queue/QueueManagerTest.java     |  29 +-
 stack/pom.xml                                   |  43 --
 stack/rest/pom.xml                              |   4 -
 .../org/apache/usergrid/rest/RootResource.java  |   6 -
 stack/services/pom.xml                          |  13 +-
 .../cassandra/ManagementServiceImpl.java        |   9 +-
 .../services/AbstractCollectionService.java     |  40 +-
 .../services/AbstractConnectionsService.java    |  39 +-
 .../notifications/NotificationsService.java     |  50 +-
 .../services/notifications/QueueListener.java   |   5 +-
 .../usergrid/services/queues/QueueListener.java |   5 +-
 .../notifications/NotifiersServiceIT.java       |   6 +
 stack/test-utils/pom.xml                        |   5 -
 105 files changed, 3849 insertions(+), 1571 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/7b5d2224/stack/awscluster/src/main/dist/init_instance/init_rest_server.sh
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/7b5d2224/stack/core/pom.xml
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/7b5d2224/stack/core/src/main/java/org/apache/usergrid/corepersistence/CoreModule.java
----------------------------------------------------------------------
diff --cc stack/core/src/main/java/org/apache/usergrid/corepersistence/CoreModule.java
index cf927ce,161037b..5da7067
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/CoreModule.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/CoreModule.java
@@@ -33,21 -30,17 +30,15 @@@ import org.apache.usergrid.persistence.
  import org.apache.usergrid.persistence.collection.event.EntityVersionCreated;
  import org.apache.usergrid.persistence.collection.event.EntityVersionDeleted;
  import org.apache.usergrid.persistence.collection.guice.CollectionModule;
- import org.apache.usergrid.persistence.collection.serialization.impl.migration.EntityIdScope;
  import org.apache.usergrid.persistence.core.guice.CommonModule;
  import org.apache.usergrid.persistence.core.migration.data.DataMigration;
- import org.apache.usergrid.persistence.core.migration.data.MigrationDataProvider;
- import org.apache.usergrid.persistence.core.migration.data.MigrationPlugin;
  import org.apache.usergrid.persistence.graph.guice.GraphModule;
- import org.apache.usergrid.persistence.graph.serialization.impl.migration.GraphNode;
  import org.apache.usergrid.persistence.index.guice.IndexModule;
 -import org.apache.usergrid.persistence.index.impl.BufferQueue;
 -import org.apache.usergrid.persistence.index.impl.BufferQueueSQSImpl;
  import org.apache.usergrid.persistence.map.guice.MapModule;
  import org.apache.usergrid.persistence.queue.guice.QueueModule;
- 
- import com.google.inject.AbstractModule;
- import com.google.inject.Provider;
- import com.google.inject.TypeLiteral;
- import com.google.inject.multibindings.Multibinder;
+ import org.slf4j.Logger;
+ import org.slf4j.LoggerFactory;
+ import org.springframework.context.ApplicationContext;
  
  
  /**
@@@ -76,31 -69,11 +67,31 @@@ public class CoreModule  extends Abstra
          bind(EntityManagerFactory.class).toProvider( lazyEntityManagerFactoryProvider );
  
          install( new CommonModule());
 -        install(new CollectionModule());
 -        install(new GraphModule());
 -        install( new IndexModule() );
 -//        install(new MapModule());   TODO, re-enable when index module doesn't depend on queue
 -//        install(new QueueModule());
 +        install( new CollectionModule() {
 +            /**
 +             * configure our migration data provider for all entities in the system
 +             */
 +            @Override
 +           public void configureMigrationProvider() {
 +
 +                bind(new TypeLiteral< MigrationDataProvider<EntityIdScope>>(){}).to(
 +                    AllEntitiesInSystemImpl.class );
 +           }
 +        } );
 +        install( new GraphModule() {
 +
 +            /**
 +             * Override the observable that needs to be used for migration
 +             */
 +            @Override
 +            public void configureMigrationProvider() {
 +                bind( new TypeLiteral<MigrationDataProvider<GraphNode>>() {} ).to(
 +                    AllNodesInGraphImpl.class );
 +            }
 +        } );
 +        install(new IndexModule());
-         install(new MapModule());
-         install(new QueueModule());
++       //        install(new MapModule());   TODO, re-enable when index module doesn't depend on queue
++       //        install(new QueueModule());
  
          bind(ManagerCache.class).to( CpManagerCache.class );
  

http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/7b5d2224/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpEntityManagerFactory.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/7b5d2224/stack/core/src/main/java/org/apache/usergrid/corepersistence/events/EntityVersionDeletedHandler.java
----------------------------------------------------------------------
diff --cc stack/core/src/main/java/org/apache/usergrid/corepersistence/events/EntityVersionDeletedHandler.java
index 65dacee,01848cb..95ff9dc
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/events/EntityVersionDeletedHandler.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/events/EntityVersionDeletedHandler.java
@@@ -47,18 -53,18 +54,20 @@@ import rx.schedulers.Schedulers
  public class EntityVersionDeletedHandler implements EntityVersionDeleted {
      private static final Logger logger = LoggerFactory.getLogger(EntityVersionDeletedHandler.class );
  
-     @Inject
-     private SerializationFig serializationFig;
+ 
+ 
+ 
+     private final EntityManagerFactory emf;
  
      @Inject
-     private EntityManagerFactory emf;
+     public EntityVersionDeletedHandler( final EntityManagerFactory emf ) {this.emf = emf;}
  
  
 +
      @Override
 -    public void versionDeleted(
 -            final CollectionScope scope, final Id entityId, final List<MvccEntity> entityVersions) {
 +    public void versionDeleted( final CollectionScope scope, final Id entityId,
 +                                final List<MvccLogEntry> entityVersions ) {
 +
  
          // This check is for testing purposes and for a test that to be able to dynamically turn
          // off and on delete previous versions so that it can test clean-up on read.
@@@ -87,19 -89,18 +92,19 @@@
                  scope.getName()
          );
  
-         rx.Observable.from( entityVersions )
-             .buffer(serializationFig.getBufferSize())
-             .map(new Func1<List<MvccLogEntry>, List<MvccLogEntry>>() {
+         Observable.from( entityVersions )
+             .collect( ei.createBatch(), new Action2<EntityIndexBatch, MvccEntity>() {
                  @Override
-                 public List<MvccLogEntry> call(List<MvccLogEntry> entityList) {
-                     for (MvccLogEntry entity : entityList) {
-                         eibatch.deindex(indexScope, entityId, entity.getVersion());
-                     }
-                     eibatch.execute();
-                     return entityList;
+                 public void call( final EntityIndexBatch entityIndexBatch, final MvccEntity mvccEntity ) {
+                     entityIndexBatch.deindex( indexScope, mvccEntity.getId(), mvccEntity.getVersion() );
                  }
-             }).toBlocking().last();
+             } ).doOnNext( new Action1<EntityIndexBatch>() {
+             @Override
+             public void call( final EntityIndexBatch entityIndexBatch ) {
+                 entityIndexBatch.execute();
+             }
+         } ).toBlocking().last();
      }
  
 +
  }

http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/7b5d2224/stack/core/src/main/java/org/apache/usergrid/persistence/EntityManagerFactory.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/7b5d2224/stack/core/src/main/java/org/apache/usergrid/persistence/cassandra/EntityManagerFactoryImpl.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/7b5d2224/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/cache/CachedEntityCollectionManager.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/7b5d2224/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/guice/CollectionModule.java
----------------------------------------------------------------------
diff --cc stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/guice/CollectionModule.java
index d5478d4,a73d7a7..68ddab5
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/guice/CollectionModule.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/guice/CollectionModule.java
@@@ -21,19 -21,17 +21,14 @@@ package org.apache.usergrid.persistence
  import org.safehaus.guicyfig.GuicyFigModule;
  
  import org.apache.usergrid.persistence.collection.EntityCollectionManagerFactory;
- import org.apache.usergrid.persistence.collection.EntityDeletedFactory;
- import org.apache.usergrid.persistence.collection.EntityVersionCleanupFactory;
- import org.apache.usergrid.persistence.collection.EntityVersionCreatedFactory;
- import org.apache.usergrid.persistence.collection.MvccEntity;
+ import org.apache.usergrid.persistence.collection.impl.EntityVersionTaskFactory;
 -import org.apache.usergrid.persistence.collection.MvccEntity;
  import org.apache.usergrid.persistence.collection.cache.EntityCacheFig;
  import org.apache.usergrid.persistence.collection.event.EntityDeleted;
  import org.apache.usergrid.persistence.collection.event.EntityVersionCreated;
  import org.apache.usergrid.persistence.collection.event.EntityVersionDeleted;
  import org.apache.usergrid.persistence.collection.impl.EntityCollectionManagerFactoryImpl;
--import org.apache.usergrid.persistence.collection.mvcc.MvccLogEntrySerializationStrategy;
  import org.apache.usergrid.persistence.collection.mvcc.changelog.ChangeLogGenerator;
  import org.apache.usergrid.persistence.collection.mvcc.changelog.ChangeLogGeneratorImpl;
--import org.apache.usergrid.persistence.collection.mvcc.stage.write.WriteStart;
  import org.apache.usergrid.persistence.collection.serialization.SerializationFig;
  import org.apache.usergrid.persistence.collection.serialization.UniqueValueSerializationStrategy;
  import org.apache.usergrid.persistence.collection.serialization.impl.SerializationModule;

http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/7b5d2224/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityCollectionManagerFactoryImpl.java
----------------------------------------------------------------------
diff --cc stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityCollectionManagerFactoryImpl.java
index 10d85f8,23e375d..ac82181
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityCollectionManagerFactoryImpl.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityCollectionManagerFactoryImpl.java
@@@ -26,25 -25,25 +26,18 @@@ import java.util.concurrent.ExecutionEx
  import org.apache.usergrid.persistence.collection.CollectionScope;
  import org.apache.usergrid.persistence.collection.EntityCollectionManager;
  import org.apache.usergrid.persistence.collection.EntityCollectionManagerFactory;
+ import org.apache.usergrid.persistence.collection.EntityCollectionManagerSync;
 +import org.apache.usergrid.persistence.collection.EntityDeletedFactory;
 +import org.apache.usergrid.persistence.collection.EntityVersionCleanupFactory;
 +import org.apache.usergrid.persistence.collection.EntityVersionCreatedFactory;
  import org.apache.usergrid.persistence.collection.cache.CachedEntityCollectionManager;
  import org.apache.usergrid.persistence.collection.cache.EntityCacheFig;
--import org.apache.usergrid.persistence.collection.guice.CollectionTaskExecutor;
- import org.apache.usergrid.persistence.collection.mvcc.MvccLogEntrySerializationStrategy;
- import org.apache.usergrid.persistence.collection.mvcc.stage.delete.MarkCommit;
- import org.apache.usergrid.persistence.collection.mvcc.stage.delete.MarkStart;
 -import org.apache.usergrid.persistence.collection.guice.Write;
 -import org.apache.usergrid.persistence.collection.guice.WriteUpdate;
+ import org.apache.usergrid.persistence.collection.mvcc.MvccEntitySerializationStrategy;
 -import org.apache.usergrid.persistence.collection.mvcc.MvccLogEntrySerializationStrategy;
 -import org.apache.usergrid.persistence.collection.mvcc.stage.delete.MarkCommit;
 -import org.apache.usergrid.persistence.collection.mvcc.stage.delete.MarkStart;
  import org.apache.usergrid.persistence.collection.mvcc.stage.write.RollbackAction;
  import org.apache.usergrid.persistence.collection.mvcc.stage.write.WriteCommit;
  import org.apache.usergrid.persistence.collection.mvcc.stage.write.WriteOptimisticVerify;
  import org.apache.usergrid.persistence.collection.mvcc.stage.write.WriteStart;
  import org.apache.usergrid.persistence.collection.mvcc.stage.write.WriteUniqueVerify;
- import org.apache.usergrid.persistence.collection.serialization.MvccEntitySerializationStrategy;
- import org.apache.usergrid.persistence.collection.serialization.SerializationFig;
--import org.apache.usergrid.persistence.collection.serialization.UniqueValueSerializationStrategy;
--import org.apache.usergrid.persistence.core.guice.ProxyImpl;
 -import org.apache.usergrid.persistence.core.metrics.MetricsFactory;
--import org.apache.usergrid.persistence.core.task.TaskExecutor;
  
  import com.google.common.base.Preconditions;
  import com.google.common.cache.CacheBuilder;
@@@ -74,25 -74,22 +67,24 @@@ public class EntityCollectionManagerFac
      private final UniqueValueSerializationStrategy uniqueValueSerializationStrategy;
      private final MvccLogEntrySerializationStrategy mvccLogEntrySerializationStrategy;
      private final Keyspace keyspace;
-     private final EntityVersionCleanupFactory entityVersionCleanupFactory;
-     private final EntityVersionCreatedFactory entityVersionCreatedFactory;
-     private final EntityDeletedFactory entityDeletedFactory;
+     private final EntityVersionTaskFactory entityVersionTaskFactory;
      private final TaskExecutor taskExecutor;
+     private final EntityCacheFig entityCacheFig;
+     private final MetricsFactory metricsFactory;
  
 +    private EntityCacheFig entityCacheFig;
 +    private SerializationFig serializationFig;
      private LoadingCache<CollectionScope, EntityCollectionManager> ecmCache =
          CacheBuilder.newBuilder().maximumSize( 1000 )
                      .build( new CacheLoader<CollectionScope, EntityCollectionManager>() {
                          public EntityCollectionManager load( CollectionScope scope ) {
  
                                    //create the target EM that will perform logic
 -                            final EntityCollectionManager target = new EntityCollectionManagerImpl( writeStart, writeUpdate, writeVerifyUnique,
 +                            final EntityCollectionManager target = new EntityCollectionManagerImpl(
 +                                writeStart, writeVerifyUnique,
                                  writeOptimisticVerify, writeCommit, rollback, markStart, markCommit,
                                  entitySerializationStrategy, uniqueValueSerializationStrategy,
-                                 mvccLogEntrySerializationStrategy, keyspace, serializationFig,entityVersionCleanupFactory,
-                                 entityVersionCreatedFactory, entityDeletedFactory, taskExecutor, scope );
 -                                mvccLogEntrySerializationStrategy, keyspace, entityVersionTaskFactory,
 -                                taskExecutor, scope, metricsFactory );
++                                mvccLogEntrySerializationStrategy, keyspace, serializationFig,entityVersionTaskFactory, taskExecutor, scope, metricsFactory );
  
  
                              final EntityCollectionManager proxy = new CachedEntityCollectionManager(entityCacheFig, target  );
@@@ -113,14 -111,13 +105,12 @@@
                                                 final UniqueValueSerializationStrategy uniqueValueSerializationStrategy,
                                                 final MvccLogEntrySerializationStrategy mvccLogEntrySerializationStrategy,
                                                 final Keyspace keyspace,
-                                                final EntityVersionCleanupFactory entityVersionCleanupFactory,
-                                                final EntityVersionCreatedFactory entityVersionCreatedFactory,
-                                                final EntityDeletedFactory entityDeletedFactory,
+                                                final EntityVersionTaskFactory entityVersionTaskFactory,
                                                 @CollectionTaskExecutor final TaskExecutor taskExecutor,
                                                final EntityCacheFig entityCacheFig,
-                                                final SerializationFig serializationFig) {
+                                                MetricsFactory metricsFactory) {
  
          this.writeStart = writeStart;
 -        this.writeUpdate = writeUpdate;
          this.writeVerifyUnique = writeVerifyUnique;
          this.writeOptimisticVerify = writeOptimisticVerify;
          this.writeCommit = writeCommit;
@@@ -131,20 -128,21 +121,18 @@@
          this.uniqueValueSerializationStrategy = uniqueValueSerializationStrategy;
          this.mvccLogEntrySerializationStrategy = mvccLogEntrySerializationStrategy;
          this.keyspace = keyspace;
-         this.entityVersionCleanupFactory = entityVersionCleanupFactory;
-         this.entityVersionCreatedFactory = entityVersionCreatedFactory;
-         this.entityDeletedFactory = entityDeletedFactory;
+         this.entityVersionTaskFactory = entityVersionTaskFactory;
          this.taskExecutor = taskExecutor;
          this.entityCacheFig = entityCacheFig;
-         this.serializationFig = serializationFig;
+         this.metricsFactory = metricsFactory;
      }
 -
 -
      @Override
 -    public EntityCollectionManager createCollectionManager( CollectionScope collectionScope ) {
 -        Preconditions.checkNotNull( collectionScope );
 -        try {
 -            return ecmCache.get( collectionScope );
 -        }
 -        catch ( ExecutionException ee ) {
 -            throw new RuntimeException( ee );
 +    public EntityCollectionManager createCollectionManager(CollectionScope collectionScope) {
 +        Preconditions.checkNotNull(collectionScope);
 +        try{
 +            return ecmCache.get(collectionScope);
 +        }catch (ExecutionException ee){
 +            throw new RuntimeException(ee);
          }
      }
  

http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/7b5d2224/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityCollectionManagerImpl.java
----------------------------------------------------------------------
diff --cc stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityCollectionManagerImpl.java
index 19e8937,6ba4513..001e0db
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityCollectionManagerImpl.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityCollectionManagerImpl.java
@@@ -35,6 -34,9 +37,10 @@@ import org.apache.usergrid.persistence.
  import org.apache.usergrid.persistence.collection.MvccEntity;
  import org.apache.usergrid.persistence.collection.VersionSet;
  import org.apache.usergrid.persistence.collection.guice.CollectionTaskExecutor;
++import org.apache.usergrid.persistence.collection.guice.CollectionTaskExecutor;
+ import org.apache.usergrid.persistence.collection.guice.Write;
+ import org.apache.usergrid.persistence.collection.guice.WriteUpdate;
+ import org.apache.usergrid.persistence.collection.mvcc.MvccEntitySerializationStrategy;
  import org.apache.usergrid.persistence.collection.mvcc.MvccLogEntrySerializationStrategy;
  import org.apache.usergrid.persistence.collection.mvcc.entity.MvccValidationUtils;
  import org.apache.usergrid.persistence.collection.mvcc.stage.CollectionIoEvent;
@@@ -51,8 -51,9 +57,11 @@@ import org.apache.usergrid.persistence.
  import org.apache.usergrid.persistence.collection.serialization.UniqueValueSerializationStrategy;
  import org.apache.usergrid.persistence.collection.serialization.UniqueValueSet;
  import org.apache.usergrid.persistence.core.guice.ProxyImpl;
 +import org.apache.usergrid.persistence.core.task.Task;
 +import org.apache.usergrid.persistence.core.task.TaskExecutor;
+ import org.apache.usergrid.persistence.core.metrics.MetricsFactory;
+ import org.apache.usergrid.persistence.core.task.Task;
+ import org.apache.usergrid.persistence.core.task.TaskExecutor;
  import org.apache.usergrid.persistence.core.util.Health;
  import org.apache.usergrid.persistence.core.util.ValidationUtils;
  import org.apache.usergrid.persistence.model.entity.Entity;
@@@ -69,9 -72,14 +80,17 @@@ import com.netflix.astyanax.connectionp
  import com.netflix.astyanax.model.ColumnFamily;
  import com.netflix.astyanax.model.CqlResult;
  import com.netflix.astyanax.serializers.StringSerializer;
++import org.apache.usergrid.persistence.collection.EntityDeletedFactory;
++import org.apache.usergrid.persistence.collection.EntityVersionCleanupFactory;
++import org.apache.usergrid.persistence.collection.EntityVersionCreatedFactory;
+ import org.apache.usergrid.persistence.collection.guice.CollectionTaskExecutor;
+ import org.apache.usergrid.persistence.core.task.Task;
+ import org.apache.usergrid.persistence.core.task.TaskExecutor;
  
+ import rx.Notification;
  import rx.Observable;
  import rx.Subscriber;
+ import rx.functions.Action0;
  import rx.functions.Action1;
  import rx.functions.Func1;
  import rx.schedulers.Schedulers;
@@@ -288,6 -368,140 +375,89 @@@ public class EntityCollectionManagerImp
          } );
      }
  
+ 
+     /**
+      * Retrieves all entities that correspond to each field given in the Collection.
+      * @param fields
+      * @return
+      */
+     @Override
+     public Observable<FieldSet> getEntitiesFromFields( final Collection<Field> fields ) {
+         return rx.Observable.just(fields).map( new Func1<Collection<Field>, FieldSet>() {
+             @Override
+             public FieldSet call( Collection<Field> fields ) {
+                 try {
+ 
+                     final UUID startTime = UUIDGenerator.newTimeUUID();
+ 
+                     //Get back set of unique values that correspond to collection of fields
+                     UniqueValueSet set = uniqueValueSerializationStrategy.load( collectionScope, fields );
+ 
+                     //Short circut if we don't have any uniqueValues from the given fields.
+                     if(!set.iterator().hasNext()){
+                         return new MutableFieldSet( 0 );
+                     }
+ 
+ 
+                     //loop through each field, and construct an entity load
+                     List<Id> entityIds = new ArrayList<>(fields.size());
+                     List<UniqueValue> uniqueValues = new ArrayList<>(fields.size());
+ 
+                     for(final Field expectedField: fields) {
+ 
+                         UniqueValue value = set.getValue(expectedField.getName());
+ 
+                         if(value ==null){
+                             logger.debug( "Field does not correspond to a unique value" );
+                         }
+ 
+                         entityIds.add(value.getEntityId());
+                         uniqueValues.add(value);
+                     }
+ 
+                     //Load a entity for each entityId we retrieved.
+                     final EntitySet entitySet = entitySerializationStrategy.load(collectionScope, entityIds, startTime);
+ 
+                     //now loop through and ensure the entities are there.
+                     final MutationBatch deleteBatch = keyspace.prepareMutationBatch();
+ 
+                     final MutableFieldSet response = new MutableFieldSet(fields.size());
+ 
+                     for(final UniqueValue expectedUnique: uniqueValues) {
+                         final MvccEntity entity = entitySet.getEntity(expectedUnique.getEntityId());
+ 
+                         //bad unique value, delete this, it's inconsistent
+                         if(entity == null || !entity.getEntity().isPresent()){
+                             final MutationBatch valueDelete = uniqueValueSerializationStrategy.delete(collectionScope, expectedUnique);
+                             deleteBatch.mergeShallow(valueDelete);
+                             continue;
+                         }
+ 
+ 
+                         //else add it to our result set
+                         response.addEntity(expectedUnique.getField(),entity);
+ 
+                     }
+ 
+                     //TODO: explore making this an Async process
+                     //We'll repair it again if we have to
+                     deleteBatch.execute();
+ 
+                     return response;
+ 
+ 
+                 }
+                 catch ( ConnectionException e ) {
+                     logger.error( "Failed to getIdField", e );
+                     throw new RuntimeException( e );
+                 }
+             }
+         } );
+     }
+ 
+ 
+ 
 -    @Override
 -    public Observable<Entity> update( final Entity entity ) {
 -
 -        logger.debug( "Starting update process" );
 -
 -        //do our input validation
 -        Preconditions.checkNotNull( entity, "Entity is required in the new stage of the mvcc write" );
 -
 -        final Id entityId = entity.getId();
 -
 -
 -        ValidationUtils.verifyIdentity( entityId );
 -
 -        // create our observable and start the write
 -        CollectionIoEvent<Entity> writeData = new CollectionIoEvent<Entity>( collectionScope, entity );
 -
 -
 -        Observable<CollectionIoEvent<MvccEntity>> observable = stageRunner( writeData, writeUpdate );
 -
 -
 -        final Timer.Context timer = updateTimer.time();
 -        return observable.map( writeCommit ).doOnNext(new Action1<Entity>() {
 -            @Override
 -            public void call(final Entity entity) {
 -                logger.debug("sending entity to the queue");
 -
 -                //we an update, signal the fix
 -                taskExecutor.submit( entityVersionTaskFactory.getCreatedTask( collectionScope, entity ));
 -
 -                taskExecutor.submit( entityVersionTaskFactory.getCleanupTask( collectionScope, entity.getId(), entity.getVersion(), false) );
 -                //TODO T.N Change this to fire a task
 -                //                Observable.from( new CollectionIoEvent<Id>(collectionScope,
 -                // entityId ) ).map( load ).subscribeOn( Schedulers.io() ).subscribe();
 -
 -
 -            }
 -        }).doOnError(rollback)
 -            .doOnNext(new Action1<Entity>() {
 -                @Override
 -                public void call(Entity entity) {
 -                    updateMeter.mark();
 -                }
 -            })
 -            .doOnCompleted(new Action0() {
 -                @Override
 -                public void call() {
 -                    timer.stop();
 -                }
 -            });
 -    }
 -
+ 
      // fire the stages
      public Observable<CollectionIoEvent<MvccEntity>> stageRunner( CollectionIoEvent<Entity> writeData,
                                                                    WriteStart writeState ) {

http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/7b5d2224/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityDeletedTask.java
----------------------------------------------------------------------
diff --cc stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityDeletedTask.java
index dbe58d4,83f165d..5472645
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityDeletedTask.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityDeletedTask.java
@@@ -22,9 -22,8 +22,8 @@@ import com.google.inject.Inject
  import com.google.inject.assistedinject.Assisted;
  import com.netflix.astyanax.MutationBatch;
  import org.apache.usergrid.persistence.collection.CollectionScope;
- import org.apache.usergrid.persistence.collection.EntityVersionCleanupFactory;
  import org.apache.usergrid.persistence.collection.event.EntityDeleted;
 -import org.apache.usergrid.persistence.collection.mvcc.MvccEntitySerializationStrategy;
 +import org.apache.usergrid.persistence.collection.serialization.MvccEntitySerializationStrategy;
  import org.apache.usergrid.persistence.collection.mvcc.MvccLogEntrySerializationStrategy;
  import org.apache.usergrid.persistence.core.task.Task;
  import org.apache.usergrid.persistence.model.entity.Id;

http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/7b5d2224/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityVersionCleanupTask.java
----------------------------------------------------------------------
diff --cc stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityVersionCleanupTask.java
index 7c39ef4,65506ec..27c3db8
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityVersionCleanupTask.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityVersionCleanupTask.java
@@@ -20,9 -20,17 +20,18 @@@ package org.apache.usergrid.persistence
  
  import java.util.Iterator;
  import java.util.List;
 +import java.util.Set;
  import java.util.UUID;
  
+ import com.google.inject.Inject;
+ import com.google.inject.assistedinject.Assisted;
+ import org.apache.usergrid.persistence.collection.MvccEntity;
+ import org.apache.usergrid.persistence.collection.serialization.UniqueValue;
+ import org.apache.usergrid.persistence.collection.serialization.UniqueValueSerializationStrategy;
+ import org.apache.usergrid.persistence.collection.serialization.impl.UniqueValueImpl;
+ import org.apache.usergrid.persistence.collection.util.EntityUtils;
+ import org.apache.usergrid.persistence.model.entity.Entity;
+ import org.apache.usergrid.persistence.model.field.Field;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
@@@ -70,6 -75,7 +79,7 @@@ public class EntityVersionCleanupTask i
      private final CollectionScope scope;
      private final Id entityId;
      private final UUID version;
 -    private final int numToSkip;
++    private final boolean includeVersion;
  
  
      @Inject
@@@ -89,6 -100,8 +103,8 @@@
          this.scope = scope;
          this.entityId = entityId;
          this.version = version;
+ 
 -        numToSkip = includeVersion? 0: 1;
++        includeVersion = includeVersion;
      }
  
  
@@@ -117,94 -130,79 +133,92 @@@
      @Override
      public Void call() throws Exception {
          //TODO Refactor this logic into a a class that can be invoked from anywhere
 -        //load every entity we have history of
 -        Observable<List<MvccEntity>> deleteFieldsObservable =
 -            Observable.create(new ObservableIterator<MvccEntity>("deleteColumns") {
 -                @Override
 -                protected Iterator<MvccEntity> getIterator() {
 -                    Iterator<MvccEntity> entities =  entitySerializationStrategy.loadDescendingHistory(
 -                        scope, entityId, version, 1000); // TODO: what fetchsize should we use here?
 -                    return entities;
 -                }
 -            })
 -            //buffer them for efficiency
 -            .skip(numToSkip)
 -            .buffer(serializationFig.getBufferSize()).doOnNext(
 -            new Action1<List<MvccEntity>>() {
 -                @Override
 -                public void call(final List<MvccEntity> mvccEntities) {
 -                    final MutationBatch batch = keyspace.prepareMutationBatch();
 -                    final MutationBatch entityBatch = keyspace.prepareMutationBatch();
 -                    final MutationBatch logBatch = keyspace.prepareMutationBatch();
--
 -                    for (MvccEntity mvccEntity : mvccEntities) {
 -                        final UUID entityVersion = mvccEntity.getVersion();
--
-         //iterate all unique values
 -
 -                        //if the entity is present process the fields
 -                        if(mvccEntity.getEntity().isPresent()) {
 -                            final Entity entity = mvccEntity.getEntity().get();
 -
 -                            //remove all unique fields from the index
 -                            for ( final Field field : EntityUtils.getUniqueFields(entity )) {
 -
 -                                final UniqueValue unique = new UniqueValueImpl( field, entityId, entityVersion );
 -                                final MutationBatch deleteMutation =
 -                                    uniqueValueSerializationStrategy.delete( scope, unique );
 -                                batch.mergeShallow( deleteMutation );
++       //iterate all unique values
 +        final BlockingObservable<Long> uniqueValueCleanup =
 +                Observable.create( new ObservableIterator<UniqueValue>( "Unique value load" ) {
 +                    @Override
 +                    protected Iterator<UniqueValue> getIterator() {
 +                        return uniqueValueSerializationStrategy.getAllUniqueFields( scope, entityId );
 +                    }
 +                } )
 +
 +                        //skip current versions
 +                        .skipWhile( new Func1<UniqueValue, Boolean>() {
 +                            @Override
 +                            public Boolean call( final UniqueValue uniqueValue ) {
 +                                return version.equals( uniqueValue.getEntityVersion() );
                              }
 +                        } )
 +                                //buffer our buffer size, then roll them all up in a single batch mutation
 +                        .buffer( serializationFig.getBufferSize() ).doOnNext( new Action1<List<UniqueValue>>() {
 +                    @Override
 +                    public void call( final List<UniqueValue> uniqueValues ) {
 +                        final MutationBatch uniqueCleanupBatch = keyspace.prepareMutationBatch();
 +
 +
 +                        for ( UniqueValue value : uniqueValues ) {
 +                            uniqueCleanupBatch.mergeShallow( uniqueValueSerializationStrategy.delete( scope, value ) );
                          }
  
 -                        final MutationBatch entityDelete = entitySerializationStrategy
 -                                .delete(scope, entityId, mvccEntity.getVersion());
 -                        entityBatch.mergeShallow( entityDelete );
 -                        final MutationBatch logDelete = logEntrySerializationStrategy
 -                                .delete(scope, entityId, version);
 -                        logBatch.mergeShallow(logDelete);
 +                        try {
 +                            uniqueCleanupBatch.execute();
 +                        }
 +                        catch ( ConnectionException e ) {
 +                            throw new RuntimeException( "Unable to execute batch mutation", e );
 +                        }
                      }
 +                } ).subscribeOn( Schedulers.io() ).longCount().toBlocking();
  
 -                    try {
 -                        batch.execute();
 -                    } catch (ConnectionException e1) {
 -                        throw new RuntimeException("Unable to execute " +
 -                                "unique value " +
 -                                "delete", e1);
 -                    }
 -                    fireEvents(mvccEntities);
 -                    try {
 -                        entityBatch.execute();
 -                    } catch (ConnectionException e) {
 -                        throw new RuntimeException("Unable to delete entities in cleanup", e);
 +
 +        //start calling the listeners for remove log entries
 +        BlockingObservable<Long> versionsDeletedObservable =
 +
 +                Observable.create( new ObservableIterator<MvccLogEntry>( "Log entry iterator" ) {
 +                    @Override
 +                    protected Iterator<MvccLogEntry> getIterator() {
 +
 +                        return new LogEntryIterator( logEntrySerializationStrategy, scope, entityId, version,
 +                                serializationFig.getBufferSize() );
                      }
 +                } )
 +                        //skip current version
 +                        .skipWhile( new Func1<MvccLogEntry, Boolean>() {
 +                            @Override
 +                            public Boolean call( final MvccLogEntry mvccLogEntry ) {
 +                                return version.equals( mvccLogEntry.getVersion() );
 +                            }
 +                        } )
 +                                //buffer them for efficiency
 +                        .buffer( serializationFig.getBufferSize() ).doOnNext( new Action1<List<MvccLogEntry>>() {
 +                    @Override
 +                    public void call( final List<MvccLogEntry> mvccEntities ) {
 +
 +                        fireEvents( mvccEntities );
 +
 +                        final MutationBatch logCleanupBatch = keyspace.prepareMutationBatch();
 +
 +
 +                        for ( MvccLogEntry entry : mvccEntities ) {
 +                            logCleanupBatch.mergeShallow( logEntrySerializationStrategy.delete( scope, entityId, entry.getVersion() ));
 +                        }
  
 -                    try {
 -                        logBatch.execute();
 -                    } catch (ConnectionException e) {
 -                        throw new RuntimeException("Unable to delete entities from the log", e);
 +                        try {
 +                            logCleanupBatch.execute();
 +                        }
 +                        catch ( ConnectionException e ) {
 +                            throw new RuntimeException( "Unable to execute batch mutation", e );
 +                        }
                      }
 +                } ).subscribeOn( Schedulers.io() ).longCount().toBlocking();
 +
 +        //wait or this to complete
 +        final Long removedCount = uniqueValueCleanup.last();
  
 -                }
 -            }
 -        );
 +        logger.debug( "Removed unique values for {} entities of entity {}", removedCount, entityId );
  
 -        final int removedCount = deleteFieldsObservable.count().toBlocking().last();
 +        final Long versionCleanupCount = versionsDeletedObservable.last();
  
 -        logger.debug("Removed unique values for {} entities of entity {}",removedCount,entityId);
 +        logger.debug( "Removed {} previous entity versions of entity {}", versionCleanupCount, entityId );
  
          return null;
      }

http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/7b5d2224/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/UniqueValueSerializationStrategy.java
----------------------------------------------------------------------
diff --cc stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/UniqueValueSerializationStrategy.java
index 5b7898e,60275d9..d7a6407
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/UniqueValueSerializationStrategy.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/UniqueValueSerializationStrategy.java
@@@ -23,9 -22,9 +23,10 @@@ import java.util.Iterator
  
  import com.netflix.astyanax.MutationBatch;
  import com.netflix.astyanax.connectionpool.exceptions.ConnectionException;
+ import com.netflix.astyanax.model.ConsistencyLevel;
  import org.apache.usergrid.persistence.collection.CollectionScope;
  import org.apache.usergrid.persistence.core.migration.schema.Migration;
 +import org.apache.usergrid.persistence.model.entity.Id;
  import org.apache.usergrid.persistence.model.field.Field;
  
  
@@@ -55,27 -53,23 +56,38 @@@ public interface UniqueValueSerializati
      /**
       * Load UniqueValue that matches field from collection or null if that value does not exist.
       *
 -     * @param colScope Collection scope in which to look for field name/value
 +     * @param collectionScope scope in which to look for field name/value
       * @param fields Field name/value to search for
 +     *
       * @return UniqueValueSet containing fields from the collection that exist in cassandra
 +     *
       * @throws ConnectionException on error connecting to Cassandra
       */
 -    public UniqueValueSet load( CollectionScope colScope, Collection<Field> fields ) throws ConnectionException;
 +    public UniqueValueSet load( CollectionScope collectionScope, Collection<Field> fields ) throws ConnectionException;
  
+     /**
 -     * Load UniqueValue that matches field from collection or null if that value does not exist.
 -     *
 -     * @param colScope Collection scope in which to look for field name/value
 -     * @param consistencyLevel Consistency level of query
 -     * @param fields Field name/value to search for
 -     * @return UniqueValueSet containing fields from the collection that exist in cassandra
 -     * @throws ConnectionException on error connecting to Cassandra
 -     */
++    * Load UniqueValue that matches field from collection or null if that value does not exist.
++    *
++    * @param colScope Collection scope in which to look for field name/value
++    * @param consistencyLevel Consistency level of query
++    * @param fields Field name/value to search for
++    * @return UniqueValueSet containing fields from the collection that exist in cassandra
++    * @throws ConnectionException on error connecting to Cassandra
++    */
+     public UniqueValueSet load( CollectionScope colScope, ConsistencyLevel consistencyLevel, Collection<Field> fields ) throws ConnectionException;
++
 +
 +    /**
 +     * Loads the currently persisted history of every unique value the entity has held.  This will
 +     * start from the max version and return values in descending version order.  Note that for entities
 +     * with more than one unique field, sequential fields can be returned with the same version.
 +     * @param collectionScope The scope the entity is stored in
 +     * @param entityId
 +     * @return
 +     */
 +    public Iterator<UniqueValue> getAllUniqueFields(CollectionScope collectionScope, Id entityId);
 +
 +
      /**
       * Delete the specified Unique Value from Cassandra.
       *

http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/7b5d2224/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/UniqueValueSerializationStrategyImpl.java
----------------------------------------------------------------------
diff --cc stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/UniqueValueSerializationStrategyImpl.java
index 7aed8db,47372ae..108f2e8
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/UniqueValueSerializationStrategyImpl.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/UniqueValueSerializationStrategyImpl.java
@@@ -38,16 -36,8 +40,17 @@@ import org.apache.usergrid.persistence.
  import org.apache.usergrid.persistence.collection.serialization.UniqueValue;
  import org.apache.usergrid.persistence.collection.serialization.UniqueValueSerializationStrategy;
  import org.apache.usergrid.persistence.collection.serialization.UniqueValueSet;
 +import org.apache.usergrid.persistence.collection.util.EntityUtils;
 +import org.apache.usergrid.persistence.core.astyanax.ColumnNameIterator;
 +import org.apache.usergrid.persistence.core.astyanax.ColumnParser;
 +import org.apache.usergrid.persistence.core.astyanax.ColumnTypes;
 +import org.apache.usergrid.persistence.core.astyanax.IdRowCompositeSerializer;
 +import org.apache.usergrid.persistence.core.astyanax.MultiTennantColumnFamily;
 +import org.apache.usergrid.persistence.core.astyanax.MultiTennantColumnFamilyDefinition;
 +import org.apache.usergrid.persistence.core.astyanax.ScopedRowKey;
+ import org.apache.usergrid.persistence.core.migration.schema.Migration;
  import org.apache.usergrid.persistence.core.util.ValidationUtils;
 +import org.apache.usergrid.persistence.model.entity.Entity;
  import org.apache.usergrid.persistence.model.entity.Id;
  import org.apache.usergrid.persistence.model.field.Field;
  
@@@ -78,40 -65,23 +81,43 @@@ public class UniqueValueSerializationSt
  
      private static final EntityVersionSerializer ENTITY_VERSION_SER = new EntityVersionSerializer();
  
 -    private static final MultiTennantColumnFamily<ScopedRowKey<CollectionPrefixedKey<Field>>, EntityVersion> CF_UNIQUE_VALUES =
 -            new MultiTennantColumnFamily<>( "Unique_Values", ROW_KEY_SER,
 -                    ENTITY_VERSION_SER );
 +    private static final MultiTennantColumnFamily<ScopedRowKey<CollectionPrefixedKey<Field>>, EntityVersion>
 +        CF_UNIQUE_VALUES = new MultiTennantColumnFamily<>( "Unique_Values", ROW_KEY_SER, ENTITY_VERSION_SER );
 +
 +
++
 +    private static final IdRowCompositeSerializer ID_SER = IdRowCompositeSerializer.get();
 +
 +
 +    private static final CollectionScopedRowKeySerializer<Id> ENTITY_ROW_KEY_SER =
 +        new CollectionScopedRowKeySerializer<>( ID_SER );
 +
  
 +
 +    private static final MultiTennantColumnFamily<ScopedRowKey<CollectionPrefixedKey<Id>>, UniqueFieldEntry>
 +        CF_ENTITY_UNIQUE_VALUES =
 +        new MultiTennantColumnFamily<>( "Entity_Unique_Values", ENTITY_ROW_KEY_SER, UniqueFieldEntrySerializer.get() );
 +
 +    public static final int COL_VALUE = 0x0;
 +
 +
-     private final Keyspace keyspace;
 +    private final SerializationFig serializationFig;
+     protected final Keyspace keyspace;
 -    private final CassandraFig cassandraFig;
++       private final CassandraFig cassandraFig;
 +
  
  
      /**
       * Construct serialization strategy for keyspace.
       *
       * @param keyspace Keyspace in which to store Unique Values.
 +     * @param serializationFig
       */
      @Inject
-     public UniqueValueSerializationStrategyImpl( final Keyspace keyspace, final SerializationFig serializationFig ) {
 -    public UniqueValueSerializationStrategyImpl( final Keyspace keyspace, final CassandraFig cassandraFig) {
 -        this.cassandraFig = cassandraFig;
++    public UniqueValueSerializationStrategyImpl( final Keyspace keyspace, final CassandraFig cassandraFig, final  SerializationFig serializationFig ) {
          this.keyspace = keyspace;
++        this.cassandraFig = cassandraFig;
 +        this.serializationFig = serializationFig;
      }
  
  

http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/7b5d2224/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/util/EntityUtils.java
----------------------------------------------------------------------
diff --cc stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/util/EntityUtils.java
index beff3bd,cf964a3..20edb66
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/util/EntityUtils.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/util/EntityUtils.java
@@@ -2,9 -2,10 +2,10 @@@ package org.apache.usergrid.persistence
  
  
  import java.util.ArrayList;
- import java.util.Collection;
  import java.util.List;
 -import java.util.Set;
++import java.util.Collection;
  import java.util.UUID;
+ import org.apache.usergrid.persistence.model.field.Field;
  
  import org.apache.commons.lang3.reflect.FieldUtils;
  

http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/7b5d2224/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/EntityCollectionManagerIT.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/7b5d2224/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/GraphManagerImpl.java
----------------------------------------------------------------------
diff --cc stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/GraphManagerImpl.java
index 8f78799,f19d613..246bbca
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/GraphManagerImpl.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/GraphManagerImpl.java
@@@ -96,7 -126,8 +126,8 @@@ public class GraphManagerImpl implement
                               final GraphFig graphFig,
                               final EdgeDeleteListener edgeDeleteListener,
                               final NodeDeleteListener nodeDeleteListener,
-                              final ApplicationScope scope) {
 -                             @Assisted final ApplicationScope scope,
++                             final ApplicationScope scope,
+                              MetricsFactory metricsFactory) {
  
  
          ValidationUtils.validateApplicationScope( scope );

http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/7b5d2224/stack/corepersistence/map/src/main/java/org/apache/usergrid/persistence/map/MapManager.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/7b5d2224/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/EntityIndex.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/7b5d2224/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsEntityIndexBatchImpl.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/7b5d2224/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsEntityIndexImpl.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/7b5d2224/stack/corepersistence/queryindex/src/test/java/org/apache/usergrid/persistence/index/guice/TestIndexModule.java
----------------------------------------------------------------------
diff --cc stack/corepersistence/queryindex/src/test/java/org/apache/usergrid/persistence/index/guice/TestIndexModule.java
index 7fccd11,50b994d..27fe705
--- a/stack/corepersistence/queryindex/src/test/java/org/apache/usergrid/persistence/index/guice/TestIndexModule.java
+++ b/stack/corepersistence/queryindex/src/test/java/org/apache/usergrid/persistence/index/guice/TestIndexModule.java
@@@ -19,8 -19,12 +19,9 @@@
  package org.apache.usergrid.persistence.index.guice;
  
  
- import org.apache.usergrid.persistence.core.guice.CommonModule;
+ import org.apache.usergrid.persistence.collection.guice.CollectionModule;
  import org.apache.usergrid.persistence.core.guice.TestModule;
+ import org.apache.usergrid.persistence.core.guice.CommonModule;
 -import org.apache.usergrid.persistence.index.impl.BufferQueue;
 -import org.apache.usergrid.persistence.index.impl.BufferQueueInMemoryImpl;
 -import org.apache.usergrid.persistence.index.impl.BufferQueueSQSImpl;
  
  
  public class TestIndexModule extends TestModule {

http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/7b5d2224/stack/corepersistence/queryindex/src/test/java/org/apache/usergrid/persistence/index/impl/EntityIndexTest.java
----------------------------------------------------------------------
diff --cc stack/corepersistence/queryindex/src/test/java/org/apache/usergrid/persistence/index/impl/EntityIndexTest.java
index e4043d2,ca9bf79..9cb5297
--- a/stack/corepersistence/queryindex/src/test/java/org/apache/usergrid/persistence/index/impl/EntityIndexTest.java
+++ b/stack/corepersistence/queryindex/src/test/java/org/apache/usergrid/persistence/index/impl/EntityIndexTest.java
@@@ -24,10 -24,14 +24,12 @@@ import java.util.*
  import java.util.concurrent.CountDownLatch;
  import java.util.concurrent.atomic.AtomicLong;
  
+ import org.apache.usergrid.persistence.core.guice.MigrationManagerRule;
  import org.apache.usergrid.persistence.index.*;
 -import org.apache.usergrid.persistence.model.field.ArrayField;
 -import org.apache.usergrid.persistence.model.field.EntityObjectField;
 -import org.apache.usergrid.persistence.model.field.UUIDField;
 +import org.apache.usergrid.persistence.model.field.*;
  import org.apache.usergrid.persistence.model.field.value.EntityObject;
  import org.junit.Ignore;
+ import org.junit.Rule;
  import org.junit.Test;
  import org.junit.runner.RunWith;
  import org.slf4j.Logger;

http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/7b5d2224/stack/pom.xml
----------------------------------------------------------------------


Mime
View raw message