diff --git a/stack/config/src/main/resources/usergrid-default.properties b/stack/config/src/main/resources/usergrid-default.properties
index 6dd3a24..39ae84e 100644
--- a/stack/config/src/main/resources/usergrid-default.properties
+++ b/stack/config/src/main/resources/usergrid-default.properties
@@ -405,6 +405,13 @@
 #
 #usergrid.queue.deliveryLimit=5
 
+# Set the number of async workers used to publish messages to SNS
+#
+#usergrid.queue.publish.threads=100
+
+# Set the queue size for the number of messages that can be queued during async publishing to SNS
+#
+#usergrid.queue.publish.queuesize=850000
 
 
 
diff --git a/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/executor/TaskExecutorFactory.java b/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/executor/TaskExecutorFactory.java
index c653458..3c6a750 100644
--- a/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/executor/TaskExecutorFactory.java
+++ b/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/executor/TaskExecutorFactory.java
@@ -20,11 +20,10 @@
 package org.apache.usergrid.persistence.core.executor;
 
 
-import java.util.concurrent.ArrayBlockingQueue;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.ThreadFactory;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicLong;
 
 
@@ -33,7 +32,12 @@
  */
 public class TaskExecutorFactory {
 
+    private static final Logger log = LoggerFactory.getLogger(TaskExecutorFactory.class);
 
+    public enum RejectionAction {
+        ABORT,
+        CALLERRUNS
+    }
     /**
      * Create a task executor
      * @param schedulerName
@@ -42,19 +46,28 @@
      * @return
      */
     public static ThreadPoolExecutor createTaskExecutor( final String schedulerName, final int maxThreadCount,
-                                                         final int maxQueueSize ) {
+                                                         final int maxQueueSize, RejectionAction rejectionAction ) {
 
 
         final BlockingQueue<Runnable> queue = new ArrayBlockingQueue<Runnable>( maxQueueSize );
 
 
-        final MaxSizeThreadPool threadPool = new MaxSizeThreadPool( queue, schedulerName, maxThreadCount );
+        if(rejectionAction.equals(RejectionAction.ABORT)){
 
+            return new MaxSizeThreadPool( queue, schedulerName, maxThreadCount );
 
-        return threadPool;
+        }
+        else if(rejectionAction.equals(RejectionAction.CALLERRUNS)){
+
+            return new MaxSizeThreadPoolCallerRuns( queue, schedulerName, maxThreadCount );
+
+        }else{
+            //default to the thread pool with ABORT policy
+            return new MaxSizeThreadPool( queue, schedulerName, maxThreadCount );
+        }
+
     }
 
-
     /**
      * Create a thread pool that will reject work if our audit tasks become overwhelmed
      */
@@ -65,6 +78,17 @@
         }
     }
 
+    /**
+     * Create a thread pool that will implement CallerRunsPolicy if our tasks become overwhelmed
+     */
+    private static final class MaxSizeThreadPoolCallerRuns extends ThreadPoolExecutor {
+
+        public MaxSizeThreadPoolCallerRuns( final BlockingQueue<Runnable> queue, final String poolName, final int maxPoolSize ) {
+            super( maxPoolSize, maxPoolSize, 30, TimeUnit.SECONDS, queue,
+                new CountingThreadFactory( poolName ), new RejectedHandler(poolName) );
+        }
+    }
+
 
     /**
      * Thread factory that will name and count threads for easier debugging
@@ -87,9 +111,29 @@
             Thread t = new Thread( r, threadName );
 
             //set it to be a daemon thread so it doesn't block shutdown
-            t.setDaemon( true );
+            t.setDaemon(true);
 
             return t;
         }
     }
+
+    /**
+     * The handler that will handle rejected executions and signal the interface
+     */
+    private static final class RejectedHandler implements RejectedExecutionHandler {
+
+        private final String poolName;
+
+        private RejectedHandler (final String poolName) {this.poolName = poolName;}
+
+        @Override
+        public void rejectedExecution( final Runnable r, final ThreadPoolExecutor executor ) {
+            log.warn( "{} task queue full, rejecting task {} and running in thread {}", poolName, r, Thread.currentThread().getName() );
+
+            //We've decided we want to have a "caller runs" policy, to just invoke the task when rejected
+
+            r.run();
+        }
+
+    }
 }
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/impl/ShardGroupCompactionImpl.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/impl/ShardGroupCompactionImpl.java
index 2194400..c9e389f 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/impl/ShardGroupCompactionImpl.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/impl/ShardGroupCompactionImpl.java
@@ -126,7 +126,7 @@
 
         this.taskExecutor = MoreExecutors.listeningDecorator( TaskExecutorFactory
             .createTaskExecutor( "ShardCompaction", graphFig.getShardAuditWorkerCount(),
-                graphFig.getShardAuditWorkerQueueSize() ) );
+                graphFig.getShardAuditWorkerQueueSize(), TaskExecutorFactory.RejectionAction.ABORT ) );
     }
 
 
diff --git a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/QueueFig.java b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/QueueFig.java
index 2a9f321..66f8af4 100644
--- a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/QueueFig.java
+++ b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/QueueFig.java
@@ -20,7 +20,7 @@
      */
     @Key( "usergrid.queue.region" )
     @Default("us-east-1")
-    public String getRegion();
+    String getRegion();
 
     /**
      * Flag to determine if Usergrid should use a multi-region Amazon queue
@@ -28,7 +28,7 @@
      */
     @Key( "usergrid.queue.multiregion" )
     @Default("false")
-    public boolean isMultiRegion();
+    boolean isMultiRegion();
 
     /**
      * Comma-separated list of one or more Amazon regions to use if multiregion
@@ -36,7 +36,7 @@
      */
     @Key( "usergrid.queue.regionList" )
     @Default("us-east-1")
-    public String getRegionList();
+    String getRegionList();
 
 
     /**
@@ -45,7 +45,7 @@
      */
     @Key( "usergrid.queue.retention" )
     @Default("1209600")
-    public String getRetentionPeriod();
+    String getRetentionPeriod();
 
     /**
      * Set the amount of time (in minutes) to retain messages in a dead letter queue.
@@ -53,16 +53,25 @@
      */
     @Key( "usergrid.queue.deadletter.retention" )
     @Default("1209600")
-    public String getDeadletterRetentionPeriod();
+    String getDeadletterRetentionPeriod();
 
     /**
      * The maximum number of messages to deliver to a dead letter queue.
      */
     @Key( "usergrid.queue.deliveryLimit" )
     @Default("5")
-    public String getQueueDeliveryLimit();
+    String getQueueDeliveryLimit();
 
     @Key("usergrid.use.default.queue")
     @Default("false")
-    public boolean overrideQueueForDefault();
+    boolean overrideQueueForDefault();
+
+    @Key("usergrid.queue.publish.threads")
+    @Default("100")
+    int getAsyncMaxThreads();
+
+    // current msg size 1.2kb * 850000 = 1.02 GB (let this default be the most we'll queue in heap)
+    @Key("usergrid.queue.publish.queuesize")
+    @Default("850000")
+    int getAsyncQueueSize();
 }
diff --git a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/impl/SNSQueueManagerImpl.java b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/impl/SNSQueueManagerImpl.java
index bcf4499..8fb0f52 100644
--- a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/impl/SNSQueueManagerImpl.java
+++ b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/impl/SNSQueueManagerImpl.java
@@ -19,8 +19,10 @@
 
 
 import com.amazonaws.AmazonServiceException;
+import com.amazonaws.handlers.AsyncHandler;
 import com.amazonaws.regions.Region;
 import com.amazonaws.regions.Regions;
+import com.amazonaws.services.sns.AmazonSNSAsyncClient;
 import com.amazonaws.services.sns.AmazonSNSClient;
 import com.amazonaws.services.sns.model.*;
 import com.amazonaws.services.sqs.AmazonSQSClient;
@@ -39,15 +41,15 @@
 import org.apache.usergrid.persistence.queue.*;
 import org.apache.usergrid.persistence.queue.Queue;
 import org.apache.usergrid.persistence.queue.util.AmazonNotificationUtils;
-import org.apache.usergrid.persistence.core.rx.RxTaskScheduler;
-
+import org.apache.usergrid.persistence.core.executor.TaskExecutorFactory;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
 import java.util.*;
 import java.util.concurrent.ExecutionException;
-import rx.Observable;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
 
 public class SNSQueueManagerImpl implements QueueManager {
 
@@ -57,9 +59,10 @@
     private final QueueFig fig;
     private final ClusterFig clusterFig;
     private final CassandraFig cassandraFig;
+    private final QueueFig queueFig;
     private final AmazonSQSClient sqs;
     private final AmazonSNSClient sns;
-    private final RxTaskScheduler rxTaskScheduler;
+    private final AmazonSNSAsyncClient snsAsync;
 
 
     private final JsonFactory JSON_FACTORY = new JsonFactory();
@@ -67,57 +70,59 @@
 
 
     private final LoadingCache<String, String> writeTopicArnMap = CacheBuilder.newBuilder()
-            .maximumSize(1000)
-            .build(new CacheLoader<String, String>() {
-                @Override
-                public String load(String queueName)
-                        throws Exception {
+        .maximumSize(1000)
+        .build(new CacheLoader<String, String>() {
+            @Override
+            public String load(String queueName)
+                throws Exception {
 
-                    return setupTopics(queueName);
-                }
-            });
+                return setupTopics(queueName);
+            }
+        });
 
     private final LoadingCache<String, Queue> readQueueUrlMap = CacheBuilder.newBuilder()
-            .maximumSize(1000)
-            .build(new CacheLoader<String, Queue>() {
-                @Override
-                public Queue load(String queueName) throws Exception {
+        .maximumSize(1000)
+        .build(new CacheLoader<String, Queue>() {
+            @Override
+            public Queue load(String queueName) throws Exception {
 
-                    Queue queue = null;
+                Queue queue = null;
 
-                    try {
-                        GetQueueUrlResult result = sqs.getQueueUrl(queueName);
-                        queue = new Queue(result.getQueueUrl());
-                    } catch (QueueDoesNotExistException queueDoesNotExistException) {
-                        logger.error("Queue {} does not exist, will create", queueName);
-                    } catch (Exception e) {
-                        logger.error("failed to get queue from service", e);
-                        throw e;
-                    }
-
-                    if (queue == null) {
-                        String url = AmazonNotificationUtils.createQueue(sqs, queueName, fig);
-                        queue = new Queue(url);
-                    }
-
-                    setupTopics(queueName);
-
-                    return queue;
+                try {
+                    GetQueueUrlResult result = sqs.getQueueUrl(queueName);
+                    queue = new Queue(result.getQueueUrl());
+                } catch (QueueDoesNotExistException queueDoesNotExistException) {
+                    logger.error("Queue {} does not exist, will create", queueName);
+                } catch (Exception e) {
+                    logger.error("failed to get queue from service", e);
+                    throw e;
                 }
-            });
+
+                if (queue == null) {
+                    String url = AmazonNotificationUtils.createQueue(sqs, queueName, fig);
+                    queue = new Queue(url);
+                }
+
+                setupTopics(queueName);
+
+                return queue;
+            }
+        });
 
 
     @Inject
-    public SNSQueueManagerImpl(@Assisted QueueScope scope, QueueFig fig, ClusterFig clusterFig, CassandraFig cassandraFig, final RxTaskScheduler rxTaskScheduler) {
+    public SNSQueueManagerImpl(@Assisted QueueScope scope, QueueFig fig, ClusterFig clusterFig,
+                               CassandraFig cassandraFig, QueueFig queueFig) {
         this.scope = scope;
         this.fig = fig;
         this.clusterFig = clusterFig;
         this.cassandraFig = cassandraFig;
-        this.rxTaskScheduler = rxTaskScheduler;
+        this.queueFig = queueFig;
 
         try {
             sqs = createSQSClient(getRegion());
             sns = createSNSClient(getRegion());
+            snsAsync = createAsyncSNSClient(getRegion());
 
         } catch (Exception e) {
             throw new RuntimeException("Error setting up mapper", e);
@@ -125,7 +130,7 @@
     }
 
     private String setupTopics(final String queueName)
-            throws Exception {
+        throws Exception {
 
         logger.info("Setting up setupTopics SNS/SQS...");
 
@@ -247,10 +252,10 @@
 
                     } catch (Exception e) {
                         logger.error(String.format("ERROR Subscribing Queue ARN/Region=[%s / %s] and Topic ARN/Region=[%s / %s]",
-                                queueARN,
-                                strSqsRegion,
-                                topicARN,
-                                strSnsRegion), e);
+                            queueARN,
+                            strSqsRegion,
+                            topicARN,
+                            strSnsRegion), e);
 
 
                     }
@@ -267,6 +272,27 @@
     }
 
     /**
+     * The Asynchronous SNS client is used for publishing events to SNS.
+     *
+     */
+
+    private AmazonSNSAsyncClient createAsyncSNSClient(final Region region) {
+        final UsergridAwsCredentialsProvider ugProvider = new UsergridAwsCredentialsProvider();
+
+
+        // create our own executor which has a bounded queue w/ caller runs policy for rejected tasks
+        final Executor executor = TaskExecutorFactory
+            .createTaskExecutor("amazon-async-io", queueFig.getAsyncMaxThreads(), queueFig.getAsyncQueueSize(),
+                TaskExecutorFactory.RejectionAction.CALLERRUNS);
+
+        final AmazonSNSAsyncClient sns = new AmazonSNSAsyncClient(ugProvider.getCredentials(), (ExecutorService) executor);
+
+        sns.setRegion(region);
+
+        return sns;
+    }
+
+    /**
      * The Synchronous SNS client is used for creating topics and subscribing queues.
      *
      */
@@ -380,7 +406,7 @@
     @Override
     public void sendMessages(final List bodies) throws IOException {
 
-        if (sns == null) {
+        if (snsAsync == null) {
             logger.error("SNS client is null, perhaps it failed to initialize successfully");
             return;
         }
@@ -393,40 +419,33 @@
 
     @Override
     public void sendMessage(final Object body) throws IOException {
-        Observable.just(body).doOnNext(message->{
 
-            if (sns == null) {
-                logger.error("SNS client is null, perhaps it failed to initialize successfully");
-                return;
+        if (snsAsync == null) {
+            logger.error("SNS client is null, perhaps it failed to initialize successfully");
+            return;
+        }
+
+        final String stringBody = toString(body);
+
+        String topicArn = getWriteTopicArn();
+
+        if (logger.isDebugEnabled()) logger.debug("Publishing Message...{} to arn: {}", stringBody, topicArn);
+
+        PublishRequest publishRequest = new PublishRequest(topicArn, stringBody);
+
+        snsAsync.publishAsync(publishRequest, new AsyncHandler<PublishRequest, PublishResult>() {
+            @Override
+            public void onError(Exception e) {
+                logger.error("Error publishing message... {}", e);
             }
 
-            final String stringBody;
-            try {
+            @Override
+            public void onSuccess(PublishRequest request, PublishResult result) {
+                if (logger.isDebugEnabled())
+                    logger.debug("Successfully published... messageID=[{}],  arn=[{}]", result.getMessageId(), request.getTopicArn());
 
-                stringBody = toString(body);
-                String topicArn = getWriteTopicArn();
-
-                if (logger.isDebugEnabled()){
-                    logger.debug("Publishing Message...{} to arn: {}", stringBody, topicArn);
-                }
-
-                PublishRequest publishRequest = new PublishRequest(topicArn, stringBody);
-
-                // publish message to SNS
-                PublishResult publishResult = sns.publish(publishRequest);
-
-                if(logger.isDebugEnabled()){
-                    logger.debug("Successfully published... messageID=[{}],  arn=[{}]",
-                        publishResult.getMessageId(), publishRequest.getTopicArn());
-                }
-
-            } catch (IOException e) {
-                logger.error("Unable to convert queue object to a string message body", e);
             }
-
-        }).doOnError(e ->{
-            logger.error("Error while publishing SNS message: ", e);
-        }).subscribeOn(rxTaskScheduler.getAsyncIOScheduler() ).subscribe();
+        });
 
     }
 
@@ -438,8 +457,8 @@
             logger.debug("Commit message {} to queue {}", queueMessage.getMessageId(), url);
 
         sqs.deleteMessage(new DeleteMessageRequest()
-                .withQueueUrl(url)
-                .withReceiptHandle(queueMessage.getHandle()));
+            .withQueueUrl(url)
+            .withReceiptHandle(queueMessage.getHandle()));
     }
 
 
@@ -473,7 +492,7 @@
      */
 
     private Object fromString(final String s, final Class klass)
-            throws IOException, ClassNotFoundException {
+        throws IOException, ClassNotFoundException {
 
         Object o = mapper.readValue(s, klass);
         return o;