filter some ops out, check for queue overflow

stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AmazonAsyncEventService.java

@@ -239,7 +239,7 @@ private Observable<IndexEventResult> handleMessages( final List<QueueMessage> me
 
             if (event == null) {
                 logger.error("AsyncEvent type or event is null!");
-                return Observable.just(new IndexEventResult(message, Optional.<IndexOperationMessage>absent(), false));
+                return Observable.just(new IndexEventResult(message, Optional.<IndexOperationMessage>absent(), true));
             }
             try {
                 //merge each operation to a master observable;
@@ -254,10 +254,10 @@ private Observable<IndexEventResult> handleMessages( final List<QueueMessage> me
                 } else if (event instanceof InitializeApplicationIndexEvent) {
                     //does not return observable
                     handleInitializeApplicationIndex(message);
-                    return Observable.just(new IndexEventResult(message, Optional.<IndexOperationMessage>absent(), true));
+                    return Observable.just(new IndexEventResult(message, Optional.<IndexOperationMessage>absent(), false));
                 } else {
                     logger.error("Unknown EventType: {}", event);
-                    return Observable.just(new IndexEventResult(message, Optional.<IndexOperationMessage>absent(), false));
+                    return Observable.just(new IndexEventResult(message, Optional.<IndexOperationMessage>absent(), true));
                 }
             }catch (Exception e){
                 logger.error("Failed to index entity", e,message);
@@ -270,20 +270,25 @@ private Observable<IndexEventResult> handleMessages( final List<QueueMessage> me
 
         return masterObservable
             //remove unsuccessful
-            .filter( indexEventResult -> indexEventResult.success() && indexEventResult.getIndexOperationMessage()
-                                                                                       .isPresent() )
+            .filter( indexEventResult -> indexEventResult.shouldProcess()  )
             //take the max
             .buffer( MAX_TAKE )
             //map them to index results and return them
             .flatMap( indexEventResults -> {
                 IndexOperationMessage combined = new IndexOperationMessage();
                 indexEventResults.stream().forEach(
-                    indexEventResult -> combined.ingest( indexEventResult.getIndexOperationMessage().get() ) );
+                    indexEventResult ->{
+                        if(indexEventResult.getIndexOperationMessage().isPresent()) {
+                            combined.ingest(indexEventResult.getIndexOperationMessage().get());
+                        }
+                    } );
+
 
                 //ack after successful completion of the operation.
-                return indexProducer.put( combined ).flatMap( operationResult -> Observable.from( indexEventResults ) )
+                return indexProducer.put( combined )
+                    .flatMap( operationResult -> Observable.from( indexEventResults ) )
                     //ack each message, but only if we didn't error.  If we did, we'll want to log it and
-                                    .map( indexEventResult -> {
+                    .map( indexEventResult -> {
                                         ack( indexEventResult.queueMessage );
                                         return indexEventResult;
                                     } );
@@ -562,21 +567,21 @@ public void indexBatch(final List<EdgeScope> edges, final long updatedSince) {
     public class IndexEventResult{
         private final QueueMessage queueMessage;
         private final Optional<IndexOperationMessage> indexOperationMessage;
-        private final boolean success;
+        private final boolean shouldProcess;
 
-        public IndexEventResult(QueueMessage queueMessage, Optional<IndexOperationMessage> indexOperationMessage ,boolean success){
+        public IndexEventResult(QueueMessage queueMessage, Optional<IndexOperationMessage> indexOperationMessage ,boolean shouldProcess){
 
             this.queueMessage = queueMessage;
             this.indexOperationMessage = indexOperationMessage;
-            this.success = success;
+            this.shouldProcess = shouldProcess;
         }
 
         public QueueMessage getQueueMessage() {
             return queueMessage;
         }
 
-        public boolean success() {
-            return success;
+        public boolean shouldProcess() {
+            return shouldProcess;
         }
 
         public Optional<IndexOperationMessage> getIndexOperationMessage() {

stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/IndexFig.java

@@ -196,4 +196,7 @@ public interface IndexFig extends GuicyFig {
     long getWriteTimeout();
 
 
+    @Default("1000")
+    @Key( "elasticsearch_queue_error_sleep_ms" )
+    long getSleepTimeForQueueError();
 }

stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsIndexProducerImpl.java

@@ -205,6 +205,13 @@ private void sendRequest( BulkRequestBuilder bulkRequest ) {
         }
 
         if ( error ) {
+            if(errorString.lastIndexOf("rejected execution (queue capacity")>=0){
+                try{
+                    Thread.sleep(indexFig.getSleepTimeForQueueError());
+                }catch (InterruptedException ie){
+                    //move on
+                }
+            }
             throw new RuntimeException(
                 "Error during processing of bulk index operations one of the responses failed. \n" + errorString);
         }

stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/impl/QueueManagerFactoryImpl.java

@@ -47,7 +47,7 @@ public QueueManagerFactoryImpl(final QueueFig queueFig, final QueueManagerIntern
     }
     @Override
     public QueueManager getQueueManager(QueueScope scope) {
-        if(queueFig.overrideQueueForDefault()){
+        if(true==false){
             QueueManager manager = defaultManager.get(scope.getName());
             if(manager==null){
                 manager = new DefaultQueueManager();