Unique value serialization completely converted to use CQL. Complete …

…Astyanax removal still needs to be completed.

stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityCollectionManagerFactoryImpl.java

@@ -23,6 +23,7 @@
 
 import java.util.concurrent.ExecutionException;
 
+import com.datastax.driver.core.Session;
 import org.apache.usergrid.persistence.collection.EntityCollectionManager;
 import org.apache.usergrid.persistence.collection.EntityCollectionManagerFactory;
 import org.apache.usergrid.persistence.collection.cache.EntityCacheFig;
@@ -75,6 +76,7 @@ public class EntityCollectionManagerFactoryImpl implements EntityCollectionManag
     private final UniqueValueSerializationStrategy uniqueValueSerializationStrategy;
     private final MvccLogEntrySerializationStrategy mvccLogEntrySerializationStrategy;
     private final Keyspace keyspace;
+    private final Session session;
     private final MetricsFactory metricsFactory;
     private final RxTaskScheduler rxTaskScheduler;
 
@@ -89,7 +91,7 @@ public EntityCollectionManager load( ApplicationScope scope ) {
                                 entitySerializationStrategy, uniqueValueSerializationStrategy,
                                 mvccLogEntrySerializationStrategy, keyspace,
                                 metricsFactory, serializationFig,
-                                rxTaskScheduler, scope );
+                                rxTaskScheduler, scope, session );
 
                             return target;
                         }
@@ -107,7 +109,9 @@ public EntityCollectionManagerFactoryImpl( final WriteStart writeStart, final Wr
                                                final UniqueValueSerializationStrategy uniqueValueSerializationStrategy,
                                                final MvccLogEntrySerializationStrategy mvccLogEntrySerializationStrategy,
                                                final Keyspace keyspace, final EntityCacheFig entityCacheFig,
-                                               final MetricsFactory metricsFactory, @CollectionExecutorScheduler  final RxTaskScheduler rxTaskScheduler ) {
+                                               final MetricsFactory metricsFactory,
+                                               @CollectionExecutorScheduler  final RxTaskScheduler rxTaskScheduler,
+                                               final Session session ) {
 
         this.writeStart = writeStart;
         this.writeVerifyUnique = writeVerifyUnique;
@@ -125,6 +129,7 @@ public EntityCollectionManagerFactoryImpl( final WriteStart writeStart, final Wr
         this.keyspace = keyspace;
         this.metricsFactory = metricsFactory;
         this.rxTaskScheduler = rxTaskScheduler;
+        this.session = session;
     }
     @Override
     public EntityCollectionManager createCollectionManager(ApplicationScope applicationScope) {

stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityCollectionManagerImpl.java

@@ -26,6 +26,8 @@
 import java.util.List;
 import java.util.UUID;
 
+import com.datastax.driver.core.BatchStatement;
+import com.datastax.driver.core.Session;
 import com.netflix.astyanax.model.ConsistencyLevel;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -64,15 +66,13 @@
 import org.apache.usergrid.persistence.model.entity.Entity;
 import org.apache.usergrid.persistence.model.entity.Id;
 import org.apache.usergrid.persistence.model.field.Field;
-import org.apache.usergrid.persistence.model.util.EntityUtils;
 import org.apache.usergrid.persistence.model.util.UUIDGenerator;
 
 import com.codahale.metrics.Timer;
 import com.google.common.base.Preconditions;
 import com.google.inject.Inject;
 import com.google.inject.assistedinject.Assisted;
 import com.netflix.astyanax.Keyspace;
-import com.netflix.astyanax.MutationBatch;
 import com.netflix.astyanax.connectionpool.OperationResult;
 import com.netflix.astyanax.connectionpool.exceptions.ConnectionException;
 import com.netflix.astyanax.model.ColumnFamily;
@@ -81,7 +81,6 @@
 
 import rx.Observable;
 import rx.Subscriber;
-import rx.functions.Action0;
 
 
 /**
@@ -114,6 +113,7 @@ public class EntityCollectionManagerImpl implements EntityCollectionManager {
 
 
     private final Keyspace keyspace;
+    private final Session session;
     private final Timer writeTimer;
     private final Timer deleteTimer;
     private final Timer fieldIdTimer;
@@ -136,7 +136,8 @@ public EntityCollectionManagerImpl( final WriteStart writeStart, final WriteUniq
                                         final MvccLogEntrySerializationStrategy mvccLogEntrySerializationStrategy,
                                         final Keyspace keyspace, final MetricsFactory metricsFactory,
                                         final SerializationFig serializationFig, final RxTaskScheduler rxTaskScheduler,
-                                        @Assisted final ApplicationScope applicationScope ) {
+                                        @Assisted final ApplicationScope applicationScope,
+                                        final Session session) {
         this.uniqueValueSerializationStrategy = uniqueValueSerializationStrategy;
         this.entitySerializationStrategy = entitySerializationStrategy;
         this.uniqueCleanup = uniqueCleanup;
@@ -157,6 +158,7 @@ public EntityCollectionManagerImpl( final WriteStart writeStart, final WriteUniq
         this.markCommit = markCommit;
 
         this.keyspace = keyspace;
+        this.session = session;
 
 
         this.applicationScope = applicationScope;
@@ -347,8 +349,7 @@ public Observable<FieldSet> getEntitiesFromFields( final String type, final Coll
                 //Load a entity for each entityId we retrieved.
                 final EntitySet entitySet = entitySerializationStrategy.load( applicationScope, entityIds, startTime );
 
-                //now loop through and ensure the entities are there.
-                final MutationBatch deleteBatch = keyspace.prepareMutationBatch();
+                final BatchStatement uniqueDeleteBatch = new BatchStatement();
 
                 final MutableFieldSet response = new MutableFieldSet( fields1.size() );
 
@@ -357,9 +358,8 @@ public Observable<FieldSet> getEntitiesFromFields( final String type, final Coll
 
                     //bad unique value, delete this, it's inconsistent
                     if ( entity == null || !entity.getEntity().isPresent() ) {
-                        final MutationBatch valueDelete =
-                            uniqueValueSerializationStrategy.delete( applicationScope, expectedUnique );
-                        deleteBatch.mergeShallow( valueDelete );
+                        uniqueDeleteBatch.add(
+                            uniqueValueSerializationStrategy.deleteCQL( applicationScope, expectedUnique ));
                         continue;
                     }
 
@@ -371,8 +371,7 @@ public Observable<FieldSet> getEntitiesFromFields( final String type, final Coll
                 }
 
                 //TODO: explore making this an Async process
-                //We'll repair it again if we have to
-                deleteBatch.execute();
+                session.execute(uniqueDeleteBatch);
 
                 return response;
             }

stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/stage/delete/UniqueCleanup.java

@@ -26,6 +26,8 @@
 import java.util.List;
 import java.util.UUID;
 
+import com.datastax.driver.core.BatchStatement;
+import com.datastax.driver.core.Session;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -63,19 +65,22 @@ public class UniqueCleanup
 
     private final UniqueValueSerializationStrategy uniqueValueSerializationStrategy;
     private final Keyspace keyspace;
+    private final Session session;
 
     private final SerializationFig serializationFig;
 
 
     @Inject
     public UniqueCleanup( final SerializationFig serializationFig,
                           final UniqueValueSerializationStrategy uniqueValueSerializationStrategy,
-                          final Keyspace keyspace, final MetricsFactory metricsFactory ) {
+                          final Keyspace keyspace, final MetricsFactory metricsFactory,
+                          final Session session ) {
 
         this.serializationFig = serializationFig;
         this.uniqueValueSerializationStrategy = uniqueValueSerializationStrategy;
         this.keyspace = keyspace;
         this.uniqueCleanupTimer = metricsFactory.getTimer( UniqueCleanup.class, "uniquecleanup.base" );
+        this.session = session;
     }
 
 
@@ -127,22 +132,20 @@ protected Iterator<UniqueValue> getIterator() {
                             //roll them up
 
                         .doOnNext( uniqueValues -> {
-                            final MutationBatch uniqueCleanupBatch = keyspace.prepareMutationBatch();
+
+                            final BatchStatement uniqueCleanupBatch = new BatchStatement();
 
 
                             for ( UniqueValue value : uniqueValues ) {
                                 logger
                                     .debug( "Deleting value:{} from application scope: {} ", value, applicationScope );
                                 uniqueCleanupBatch
-                                    .mergeShallow( uniqueValueSerializationStrategy.delete( applicationScope, value ) );
+                                    .add( uniqueValueSerializationStrategy.deleteCQL( applicationScope, value ) );
                             }
 
-                            try {
-                                uniqueCleanupBatch.execute();
-                            }
-                            catch ( ConnectionException e ) {
-                                throw new RuntimeException( "Unable to execute batch mutation", e );
-                            }
+
+                            session.execute(uniqueCleanupBatch);
+
                         } ).lastOrDefault( Collections.emptyList() ).map( list -> mvccEntityCollectionIoEvent );
 
                 return ObservableTimer.time( uniqueValueCleanup, uniqueCleanupTimer );

stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/stage/write/RollbackAction.java

@@ -18,6 +18,8 @@
 package org.apache.usergrid.persistence.collection.mvcc.stage.write;
 
 
+import com.datastax.driver.core.BatchStatement;
+import com.datastax.driver.core.Session;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -51,14 +53,17 @@ public class RollbackAction implements Action1<Throwable> {
 
     private final UniqueValueSerializationStrategy uniqueValueStrat;
     private final MvccLogEntrySerializationStrategy logEntryStrat;
+    private final Session session;
 
 
     @Inject
-    public RollbackAction(MvccLogEntrySerializationStrategy logEntryStrat,
-                           UniqueValueSerializationStrategy uniqueValueStrat ) {
+    public RollbackAction( final MvccLogEntrySerializationStrategy logEntryStrat,
+                           final UniqueValueSerializationStrategy uniqueValueStrat,
+                           final Session session ) {
 
         this.uniqueValueStrat = uniqueValueStrat;
         this.logEntryStrat = logEntryStrat;
+        this.session = session;
     }
 
 
@@ -72,6 +77,7 @@ public void call( final Throwable t ) {
 
             // one batch to handle rollback
             MutationBatch rollbackMb = null;
+            final BatchStatement uniqueDeleteBatch = new BatchStatement();
             final Optional<Entity> entity = mvccEntity.getEntity();
 
             if ( entity.isPresent() ) {
@@ -83,45 +89,17 @@ public void call( final Throwable t ) {
                         UniqueValue toDelete =
                                 new UniqueValueImpl( field, entity.get().getId(), mvccEntity.getVersion() );
 
-                        MutationBatch deleteMb = uniqueValueStrat.delete(scope,  toDelete );
+                        uniqueDeleteBatch.add(uniqueValueStrat.deleteCQL(scope,  toDelete ));
 
-                        if ( rollbackMb == null ) {
-                            rollbackMb = deleteMb;
-                        }
-                        else {
-                            rollbackMb.mergeShallow( deleteMb );
-                        }
                     }
                 }
 
-
-                if ( rollbackMb != null ) {
-                    try {
-                        rollbackMb.execute();
-                    }
-                    catch ( ConnectionException ex ) {
-                        throw new RuntimeException( "Error rolling back changes", ex );
-                    }
-                }
+                // execute the batch statements for deleting unique field entries
+                session.execute(uniqueDeleteBatch);
 
                 logEntryStrat.delete( scope, entity.get().getId(), mvccEntity.getVersion() );
             }
         }
     }
 
-
-    class FieldDeleteResult {
-
-        private final String name;
-
-
-        public FieldDeleteResult( String name ) {
-            this.name = name;
-        }
-
-
-        public String getName() {
-            return this.name;
-        }
-    }
 }

stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/stage/write/WriteUniqueVerify.java

@@ -25,6 +25,7 @@
 
 import com.datastax.driver.core.BatchStatement;
 import com.datastax.driver.core.Session;
+import com.netflix.hystrix.HystrixCommandProperties;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -68,7 +69,9 @@ public class WriteUniqueVerify implements Action1<CollectionIoEvent<MvccEntity>>
 
     private final UniqueValueSerializationStrategy uniqueValueStrat;
 
-    public static int uniqueVerifyPoolSize = 100;
+    private static int uniqueVerifyPoolSize = 100;
+
+    private static int uniqueVerifyTimeoutMillis= 5000;
 
     protected final SerializationFig serializationFig;
 
@@ -224,8 +227,10 @@ public Map<String, Field> executeStrategy(ConsistencyLevel consistencyLevel){
     /**
      * Command group used for realtime user commands
      */
-    public static final HystrixCommand.Setter
-        REPLAY_GROUP = HystrixCommand.Setter.withGroupKey(
-            HystrixCommandGroupKey.Factory.asKey( "uniqueVerify" ) ).andThreadPoolPropertiesDefaults(
-                HystrixThreadPoolProperties.Setter().withCoreSize( uniqueVerifyPoolSize ) );
+    private static final HystrixCommand.Setter
+        REPLAY_GROUP = HystrixCommand.Setter.withGroupKey( HystrixCommandGroupKey.Factory.asKey( "uniqueVerify" ) )
+        .andThreadPoolPropertiesDefaults(
+            HystrixThreadPoolProperties.Setter().withCoreSize( uniqueVerifyPoolSize ) )
+        .andCommandPropertiesDefaults(
+            HystrixCommandProperties.Setter().withExecutionTimeoutInMilliseconds(uniqueVerifyTimeoutMillis));
 }

stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/UniqueValueSerializationStrategy.java

@@ -41,23 +41,14 @@ public interface UniqueValueSerializationStrategy extends Migration, VersionedDa
 
 
     /**
-     * Write the specified UniqueValue to Cassandra with optional timeToLive in milliseconds.
-     *
-     * @param applicationScope scope
-     * @param uniqueValue Object to be written
-     *
-     * @return MutatationBatch that encapsulates operation, caller may or may not execute.
-     */
-
-    /**
-     * Write the specified UniqueValue to Cassandra with optional timeToLive in milliseconds.
+     * Write the specified UniqueValue to Cassandra with optional timeToLive in milliseconds. -1 is the same as no ttl
+     * (lives forever)
      *
      * @param applicationScope scope
      * @param uniqueValue Object to be written
      * @param timeToLive How long object should live in seconds.  -1 implies store forever
-     * @return MutatationBatch that encapsulates operation, caller may or may not execute.
+     * @return BatchStatement that encapsulates CQL statements, caller may or may not execute.
      */
-
     BatchStatement writeCQL(ApplicationScope applicationScope, UniqueValue uniqueValue, int timeToLive );
 
     /**
@@ -103,9 +94,9 @@ UniqueValueSet load( ApplicationScope applicationScope, ConsistencyLevel consist
      *
      * @param applicationScope The scope of the application
      * @param uniqueValue Object to be deleted.
-     * @return MutatationBatch that encapsulates operation, caller may or may not execute.
+     * @return BatchStatement that encapsulates the CQL statements, caller may or may not execute.
      */
-    MutationBatch delete( ApplicationScope applicationScope, UniqueValue uniqueValue );
+    BatchStatement deleteCQL( ApplicationScope applicationScope, UniqueValue uniqueValue);
 
 
 }

stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/AllUniqueFieldsIterator.java

@@ -0,0 +1,29 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.usergrid.persistence.collection.serialization.impl;
+
+
+import com.datastax.driver.core.Row;
+import com.datastax.driver.core.Session;
+import com.datastax.driver.core.Statement;
+import org.apache.usergrid.persistence.collection.serialization.UniqueValue;
+
+import java.util.Iterator;
+
+