You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@usergrid.apache.org by sn...@apache.org on 2016/11/03 21:56:43 UTC
[1/3] usergrid git commit: Logging changes & import cleanups.
Repository: usergrid
Updated Branches:
refs/heads/usergrid-1318-queue 198056227 -> 522a5515b
Logging changes & import cleanups.
Project: http://git-wip-us.apache.org/repos/asf/usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/usergrid/commit/1d5e407f
Tree: http://git-wip-us.apache.org/repos/asf/usergrid/tree/1d5e407f
Diff: http://git-wip-us.apache.org/repos/asf/usergrid/diff/1d5e407f
Branch: refs/heads/usergrid-1318-queue
Commit: 1d5e407fd4ab961886eb7cf4d32a1f3d194dbe1a
Parents: 1980562
Author: Dave Johnson <sn...@apache.org>
Authored: Thu Nov 3 17:20:36 2016 -0400
Committer: Dave Johnson <sn...@apache.org>
Committed: Thu Nov 3 17:20:36 2016 -0400
----------------------------------------------------------------------
.../actorsystem/ActorSystemManagerImpl.java | 9 ++++++++-
.../persistence/qakka/core/impl/InMemoryQueue.java | 5 ++++-
.../qakka/core/impl/QueueMessageManagerImpl.java | 9 ++++++---
.../qakka/distributed/actors/QueueSender.java | 4 ++++
.../impl/QueueMessageSerializationImpl.java | 8 ++++----
.../sharding/impl/ShardSerializationImpl.java | 4 ++++
.../persistence/queue/impl/QakkaQueueManager.java | 2 ++
.../usergrid/rest/system/QueueSystemResource.java | 14 +++++++++-----
8 files changed, 41 insertions(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/usergrid/blob/1d5e407f/stack/corepersistence/actorsystem/src/main/java/org/apache/usergrid/persistence/actorsystem/ActorSystemManagerImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/actorsystem/src/main/java/org/apache/usergrid/persistence/actorsystem/ActorSystemManagerImpl.java b/stack/corepersistence/actorsystem/src/main/java/org/apache/usergrid/persistence/actorsystem/ActorSystemManagerImpl.java
index 687715a..96ebe69 100644
--- a/stack/corepersistence/actorsystem/src/main/java/org/apache/usergrid/persistence/actorsystem/ActorSystemManagerImpl.java
+++ b/stack/corepersistence/actorsystem/src/main/java/org/apache/usergrid/persistence/actorsystem/ActorSystemManagerImpl.java
@@ -313,7 +313,10 @@ public class ActorSystemManagerImpl implements ActorSystemManager {
}} );
put( "cluster", new HashMap<String, Object>() {{
- put( "max-nr-of-instances-per-node", numInstancesPerNode); // this sets default if router does not set
+
+ // this sets default if router does not set
+ put( "max-nr-of-instances-per-node", numInstancesPerNode);
+
put( "roles", Collections.singletonList("io") );
put( "seed-nodes", new ArrayList<String>() {{
for (String seed : seeds) {
@@ -337,6 +340,8 @@ public class ActorSystemManagerImpl implements ActorSystemManager {
routerProducer.addConfiguration( configMap );
}
+ logger.debug("Actor system configMap: " + configMap );
+
config = ConfigFactory.parseMap( configMap )
.withFallback( ConfigFactory.load( "application.conf" ) );
@@ -409,6 +414,8 @@ public class ActorSystemManagerImpl implements ActorSystemManager {
} else {
+ logger.info( "Creating clusterClient for region [{}]", region );
+
Set<ActorPath> seedPaths = new HashSet<>(20);
for ( String seed : getSeedsByRegion().get( region ) ) {
seedPaths.add( ActorPaths.fromString( seed + "/system/receptionist") );
http://git-wip-us.apache.org/repos/asf/usergrid/blob/1d5e407f/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/core/impl/InMemoryQueue.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/core/impl/InMemoryQueue.java b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/core/impl/InMemoryQueue.java
index 6a26483..4315df0 100644
--- a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/core/impl/InMemoryQueue.java
+++ b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/core/impl/InMemoryQueue.java
@@ -26,7 +26,10 @@ import org.apache.usergrid.persistence.qakka.serialization.queuemessages.Databas
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.*;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Queue;
+import java.util.UUID;
import java.util.concurrent.ConcurrentLinkedQueue;
http://git-wip-us.apache.org/repos/asf/usergrid/blob/1d5e407f/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/core/impl/QueueMessageManagerImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/core/impl/QueueMessageManagerImpl.java b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/core/impl/QueueMessageManagerImpl.java
index 59a14bd..ac2857f 100644
--- a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/core/impl/QueueMessageManagerImpl.java
+++ b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/core/impl/QueueMessageManagerImpl.java
@@ -32,8 +32,6 @@ import org.apache.usergrid.persistence.qakka.serialization.queuemessages.Databas
import org.apache.usergrid.persistence.qakka.serialization.queuemessages.DatabaseQueueMessageBody;
import org.apache.usergrid.persistence.qakka.serialization.queuemessages.MessageCounterSerialization;
import org.apache.usergrid.persistence.qakka.serialization.queuemessages.QueueMessageSerialization;
-import org.apache.usergrid.persistence.qakka.serialization.sharding.Shard;
-import org.apache.usergrid.persistence.qakka.serialization.sharding.ShardIterator;
import org.apache.usergrid.persistence.qakka.serialization.sharding.ShardSerialization;
import org.apache.usergrid.persistence.qakka.serialization.transferlog.TransferLogSerialization;
import org.slf4j.Logger;
@@ -42,7 +40,10 @@ import org.slf4j.LoggerFactory;
import java.io.UnsupportedEncodingException;
import java.net.URISyntaxException;
import java.nio.ByteBuffer;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.UUID;
@Singleton
@@ -92,6 +93,8 @@ public class QueueMessageManagerImpl implements QueueMessageManager {
throw new NotFoundException( "Queue " + queueName + " not found" );
}
+ logger.trace("Sending message to queue {} regions {}", queueName, destinationRegions);
+
// TODO: implement delay and expiration
// Preconditions.checkArgument(delayMs == null || delayMs > 0L,
http://git-wip-us.apache.org/repos/asf/usergrid/blob/1d5e407f/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/distributed/actors/QueueSender.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/distributed/actors/QueueSender.java b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/distributed/actors/QueueSender.java
index ccc39f5..461c28f 100644
--- a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/distributed/actors/QueueSender.java
+++ b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/distributed/actors/QueueSender.java
@@ -129,12 +129,16 @@ public class QueueSender extends UntypedActor {
if (actorSystemManager.getCurrentRegion().equals( destRegion )) {
+ logger.trace("Sending queue {} message to local region {}", queueName, destRegion );
+
// send to current region via local clientActor
ActorRef clientActor = actorSystemManager.getClientActor();
fut = Patterns.ask( clientActor, request, t );
} else {
+ logger.trace("Sending queue {} message to remote region {}", queueName, destRegion );
+
// send to remote region via cluster client for that region
ActorRef clusterClient = actorSystemManager.getClusterClient( destRegion );
fut = Patterns.ask(
http://git-wip-us.apache.org/repos/asf/usergrid/blob/1d5e407f/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/serialization/queuemessages/impl/QueueMessageSerializationImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/serialization/queuemessages/impl/QueueMessageSerializationImpl.java b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/serialization/queuemessages/impl/QueueMessageSerializationImpl.java
index b1b57ae..fb49825 100644
--- a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/serialization/queuemessages/impl/QueueMessageSerializationImpl.java
+++ b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/serialization/queuemessages/impl/QueueMessageSerializationImpl.java
@@ -347,15 +347,15 @@ public class QueueMessageSerializationImpl implements QueueMessageSerialization
while (defaultShardIterator.hasNext()) {
Shard shard = defaultShardIterator.next();
- deleteAllBatch.add( createDeleteAllMessagesStatement( shard ) );
+ Statement deleteAll = createDeleteAllMessagesStatement( shard );
+ deleteAllBatch.add( deleteAll );
- logger.trace("added queueName {} type {} shard {}",
- queueName, shardType, shard.getShardId() );
+ logger.trace("Deleting messages for queue {} shardType {} shard {} query {}",
+ queueName, shardType, shard.getShardId(), deleteAll.toString() );
}
}
cassandraClient.getQueueMessageSession().execute( deleteAllBatch );
- logger.trace("deleted messages in queue: " + queueName);
// clear counters, we only want to this to happen after successful deletion
for ( Shard.Type shardType : shardTypes ) {
http://git-wip-us.apache.org/repos/asf/usergrid/blob/1d5e407f/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/serialization/sharding/impl/ShardSerializationImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/serialization/sharding/impl/ShardSerializationImpl.java b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/serialization/sharding/impl/ShardSerializationImpl.java
index 501a6ca..4f59057 100644
--- a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/serialization/sharding/impl/ShardSerializationImpl.java
+++ b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/serialization/sharding/impl/ShardSerializationImpl.java
@@ -164,9 +164,13 @@ public class ShardSerializationImpl implements ShardSerialization {
.where( QueryBuilder.eq(COLUMN_QUEUE_NAME, queueName) )
.and( QueryBuilder.eq(COLUMN_REGION, region) );
+ logger.trace("Removing shards for queue {} region {} shardType {} query {}",
+ queueName, region, shardType, batch.toString());
+
batch.add( delete );
}
+
cassandraClient.getQueueMessageSession().execute( batch );
}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/1d5e407f/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/impl/QakkaQueueManager.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/impl/QakkaQueueManager.java b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/impl/QakkaQueueManager.java
index b81e888..9d7a341 100644
--- a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/impl/QakkaQueueManager.java
+++ b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/impl/QakkaQueueManager.java
@@ -80,6 +80,8 @@ public class QakkaQueueManager implements LegacyQueueManager {
@Override
public <T extends Serializable> void sendMessage(T body) throws IOException {
+ logger.debug( "Sending message to queue {} region {}", this.scope.getRegionImplementation().name() );
+
createQueueIfNecessary();
ByteArrayOutputStream bos = new ByteArrayOutputStream();
http://git-wip-us.apache.org/repos/asf/usergrid/blob/1d5e407f/stack/rest/src/main/java/org/apache/usergrid/rest/system/QueueSystemResource.java
----------------------------------------------------------------------
diff --git a/stack/rest/src/main/java/org/apache/usergrid/rest/system/QueueSystemResource.java b/stack/rest/src/main/java/org/apache/usergrid/rest/system/QueueSystemResource.java
index a9d3e1b..16e77d0 100644
--- a/stack/rest/src/main/java/org/apache/usergrid/rest/system/QueueSystemResource.java
+++ b/stack/rest/src/main/java/org/apache/usergrid/rest/system/QueueSystemResource.java
@@ -153,23 +153,27 @@ public class QueueSystemResource extends AbstractContextResource {
final List<String> listOfQueues = queueManager.getListOfQueues();
for ( String queueName : listOfQueues ) {
+ Map<String, Object> queueInfo = new HashMap<>();
try {
- Map<String, Object> queueInfo = new HashMap<>();
queueInfo.put( "name", queueName );
+
+ queueInfo.put( "inmemory", inMemoryQueue.size( queueName ) );
+
+ UUID newest = inMemoryQueue.getNewest( queueName );
+ queueInfo.put( "since", newest == null ? "null" : newest.timestamp() );
+
queueInfo.put( "depth",
queueMessageManager.getQueueDepth( queueName, DatabaseQueueMessage.Type.DEFAULT ) );
queueInfo.put( "inflight",
queueMessageManager.getQueueDepth( queueName, DatabaseQueueMessage.Type.INFLIGHT ) );
- queueInfo.put( "inmemory", inMemoryQueue.size( queueName ) );
- UUID newest = inMemoryQueue.getNewest( queueName );
- queueInfo.put( "since", newest == null ? "null" : newest.timestamp() );
- queues.add( queueInfo );
} catch ( Exception e ) {
logger.error("Error getting queue info for queue: " + queueName, e);
}
+
+ queues.add( queueInfo );
}
info.put("queues", queues);
[2/3] usergrid git commit: Add timeout to shard and message counters,
in addition to existing "number of changes before write" logic.
Posted by sn...@apache.org.
Add timeout to shard and message counters, in addition to existing "number of changes before write" logic.
Project: http://git-wip-us.apache.org/repos/asf/usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/usergrid/commit/521047c1
Tree: http://git-wip-us.apache.org/repos/asf/usergrid/tree/521047c1
Diff: http://git-wip-us.apache.org/repos/asf/usergrid/diff/521047c1
Branch: refs/heads/usergrid-1318-queue
Commit: 521047c19504c3b1722dd27a2f3b808c612b4a34
Parents: 1d5e407
Author: Dave Johnson <sn...@apache.org>
Authored: Thu Nov 3 17:50:08 2016 -0400
Committer: Dave Johnson <sn...@apache.org>
Committed: Thu Nov 3 17:50:08 2016 -0400
----------------------------------------------------------------------
.../usergrid/persistence/qakka/QakkaFig.java | 17 +++++++++
.../impl/MessageCounterSerializationImpl.java | 37 ++++++++++++++------
.../impl/ShardCounterSerializationImpl.java | 22 ++++++++++++
3 files changed, 66 insertions(+), 10 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/usergrid/blob/521047c1/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/QakkaFig.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/QakkaFig.java b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/QakkaFig.java
index 87b5c83..105c193 100644
--- a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/QakkaFig.java
+++ b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/QakkaFig.java
@@ -53,8 +53,10 @@ public interface QakkaFig extends GuicyFig, Serializable {
String QUEUE_GET_TIMEOUT = "queue.get.timeout.seconds";
String QUEUE_SHARD_COUNTER_MAX_IN_MEMORY = "queue.shard.counter.max-in-memory";
+ String QUEUE_SHARD_COUNTER_WRITE_TIMEOUT = "queue.shard.counter.write-timeout";
String QUEUE_MESSAGE_COUNTER_MAX_IN_MEMORY = "queue.message.counter.max-in-memory";
+ String QUEUE_MESSAGE_COUNTER_WRITE_TIMEOUT = "queue.message.counter.write-timeout";
String QUEUE_SHARD_ALLOCATION_CHECK_FREQUENCY = "queue.shard.allocation.check.frequency.millis";
@@ -66,6 +68,9 @@ public interface QakkaFig extends GuicyFig, Serializable {
String QUEUE_MAX_TTL = "queue.max.ttl";
+ String QUEUE_IN_MEMORY = "queue.in-memory.cache";
+
+
/** True if Qakka is running standlone */
@Key(QUEUE_STANDALONE)
@@ -127,11 +132,19 @@ public interface QakkaFig extends GuicyFig, Serializable {
@Default("100")
long getShardCounterMaxInMemory();
+ @Key(QUEUE_SHARD_COUNTER_WRITE_TIMEOUT)
+ @Default("5000")
+ long getShardCounterWriteTimeoutMillis();
+
/** Once counter reaches this value, write it to permanent storage */
@Key(QUEUE_MESSAGE_COUNTER_MAX_IN_MEMORY)
@Default("100")
long getMessageCounterMaxInMemory();
+ @Key(QUEUE_MESSAGE_COUNTER_WRITE_TIMEOUT)
+ @Default("5000")
+ long getMessageCounterWriteTimeoutMillis();
+
/** How often to check whether new shard is needed for each queue */
@Key(QUEUE_SHARD_ALLOCATION_CHECK_FREQUENCY)
@Default("5000")
@@ -155,4 +168,8 @@ public interface QakkaFig extends GuicyFig, Serializable {
@Key(QUEUE_MAX_TTL)
@Default("1209600") // default is two weeks
int getMaxTtlSeconds();
+
+ @Key(QUEUE_IN_MEMORY)
+ @Default("false")
+ boolean getInMemoryCache();
}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/521047c1/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/serialization/queuemessages/impl/MessageCounterSerializationImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/serialization/queuemessages/impl/MessageCounterSerializationImpl.java b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/serialization/queuemessages/impl/MessageCounterSerializationImpl.java
index 5c66f43..769f584 100644
--- a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/serialization/queuemessages/impl/MessageCounterSerializationImpl.java
+++ b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/serialization/queuemessages/impl/MessageCounterSerializationImpl.java
@@ -35,6 +35,7 @@ import org.apache.usergrid.persistence.qakka.exceptions.QakkaRuntimeException;
import org.apache.usergrid.persistence.qakka.serialization.queuemessages.DatabaseQueueMessage;
import org.apache.usergrid.persistence.qakka.serialization.queuemessages.MessageCounterSerialization;
import org.apache.usergrid.persistence.qakka.serialization.sharding.Shard;
+import org.apache.usergrid.persistence.qakka.serialization.sharding.impl.ShardCounterSerializationImpl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -54,6 +55,7 @@ public class MessageCounterSerializationImpl implements MessageCounterSerializat
private final CassandraClient cassandraClient;
private final CassandraConfig cassandraConfig;
+ private final long writeTimeout;
final static String TABLE_MESSAGE_COUNTERS = "message_counters";
final static String COLUMN_QUEUE_NAME = "queue_name";
@@ -82,17 +84,18 @@ public class MessageCounterSerializationImpl implements MessageCounterSerializat
final AtomicLong totalInMemoryCount = new AtomicLong( 0L ); // for testing using only in-memory counter
final AtomicLong increment = new AtomicLong( 0L );
final AtomicLong decrement = new AtomicLong( 0L );
+ long lastWritten = 0L;
InMemoryCount( long baseCount ) {
this.baseCount = baseCount;
}
public void increment( long inc ) {
- increment.addAndGet( inc );
- totalInMemoryCount.addAndGet( inc );
+ this.increment.addAndGet( inc );
+ this.totalInMemoryCount.addAndGet( inc );
}
public void decrement( long dec ) {
- decrement.addAndGet( dec );
- totalInMemoryCount.addAndGet( -dec );
+ this.decrement.addAndGet( dec );
+ this.totalInMemoryCount.addAndGet( -dec );
}
public long getIncrement() {
return increment.get();
@@ -101,8 +104,11 @@ public class MessageCounterSerializationImpl implements MessageCounterSerializat
return decrement.get();
}
public void clearDeltas() {
- increment.set( 0L );
- decrement.set( 0L );
+ this.increment.set( 0L );
+ this.decrement.set( 0L );
+ }
+ boolean needsUpdate() {
+ return System.currentTimeMillis() - lastWritten > writeTimeout;
}
void reset() {
this.baseCount = 0;
@@ -110,13 +116,12 @@ public class MessageCounterSerializationImpl implements MessageCounterSerializat
this.decrement.set( 0L );
}
public long value() {
-
// return totalInMemoryCount.get(); // for testing using just in-memory counter:
-
return baseCount + increment.get() - decrement.get();
}
void setBaseCount( long baseCount ) {
this.baseCount = baseCount;
+ this.lastWritten = System.currentTimeMillis();
}
}
@@ -130,6 +135,7 @@ public class MessageCounterSerializationImpl implements MessageCounterSerializat
this.cassandraConfig = cassandraConfig;
this.maxChangesBeforeSave = qakkaFig.getMessageCounterMaxInMemory();
this.cassandraClient = cassandraClient;
+ this.writeTimeout = qakkaFig.getShardCounterWriteTimeoutMillis();
}
@@ -227,8 +233,19 @@ public class MessageCounterSerializationImpl implements MessageCounterSerializat
}
}
- long value = inMemoryCounters.get( key ).value();
- return value;
+ InMemoryCount inMemoryCount = inMemoryCounters.get( key );
+
+ synchronized ( inMemoryCount ) {
+
+ if ( inMemoryCount.needsUpdate() ) {
+ long totalIncrement = inMemoryCount.getIncrement();
+ incrementCounterInStorage( queueName, type, totalIncrement );
+ inMemoryCount.setBaseCount( retrieveCounterFromStorage( queueName, type ) );
+ inMemoryCount.clearDeltas();
+ }
+ }
+
+ return inMemoryCounters.get( key ).value();
}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/521047c1/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/serialization/sharding/impl/ShardCounterSerializationImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/serialization/sharding/impl/ShardCounterSerializationImpl.java b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/serialization/sharding/impl/ShardCounterSerializationImpl.java
index 8bb262b..16e31a1 100644
--- a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/serialization/sharding/impl/ShardCounterSerializationImpl.java
+++ b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/serialization/sharding/impl/ShardCounterSerializationImpl.java
@@ -52,6 +52,7 @@ public class ShardCounterSerializationImpl implements ShardCounterSerialization
private final CassandraClient cassandraClient;
private final CassandraConfig cassandraConfig;
+ private final long writeTimeout;
final static String TABLE_COUNTERS = "shard_counters";
final static String COLUMN_QUEUE_NAME = "queue_name";
@@ -72,7 +73,9 @@ public class ShardCounterSerializationImpl implements ShardCounterSerialization
class InMemoryCount {
long baseCount;
+ long lastWritten = 0L;
final AtomicLong increment = new AtomicLong( 0L );
+
InMemoryCount( long baseCount ) {
this.baseCount = baseCount;
}
@@ -84,10 +87,15 @@ public class ShardCounterSerializationImpl implements ShardCounterSerialization
}
void setBaseCount( long baseCount ) {
this.baseCount = baseCount;
+ this.lastWritten = System.currentTimeMillis();
+ }
+ boolean needsUpdate() {
+ return System.currentTimeMillis() - lastWritten > writeTimeout;
}
void reset() {
this.baseCount = 0;
this.increment.set( 0L );
+ this.lastWritten = System.currentTimeMillis();
}
}
@@ -97,9 +105,11 @@ public class ShardCounterSerializationImpl implements ShardCounterSerialization
@Inject
public ShardCounterSerializationImpl(
CassandraConfig cassandraConfig, QakkaFig qakkaFig, CassandraClient cassandraClient ) {
+
this.cassandraConfig = cassandraConfig;
this.maxInMemoryIncrement = qakkaFig.getShardCounterMaxInMemory();
this.cassandraClient = cassandraClient;
+ this.writeTimeout = qakkaFig.getShardCounterWriteTimeoutMillis();
}
@@ -160,6 +170,18 @@ public class ShardCounterSerializationImpl implements ShardCounterSerialization
}
}
+ InMemoryCount inMemoryCount = inMemoryCounters.get( key );
+
+ synchronized ( inMemoryCount ) {
+
+ if ( inMemoryCount.needsUpdate() ) {
+ long totalIncrement = inMemoryCount.getIncrement().get();
+ incrementCounterInStorage( queueName, type, shardId, totalIncrement );
+ inMemoryCount.setBaseCount( retrieveCounterFromStorage( queueName, type, shardId ) );
+ inMemoryCount.getIncrement().set( 0L );
+ }
+ }
+
return inMemoryCounters.get( key ).value();
}
[3/3] usergrid git commit: Remove explicit init message and option to
turn off in-memory queue cache.
Posted by sn...@apache.org.
Remove explicit init message and option to turn off in-memory queue cache.
Project: http://git-wip-us.apache.org/repos/asf/usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/usergrid/commit/522a5515
Tree: http://git-wip-us.apache.org/repos/asf/usergrid/tree/522a5515
Diff: http://git-wip-us.apache.org/repos/asf/usergrid/diff/522a5515
Branch: refs/heads/usergrid-1318-queue
Commit: 522a5515baf016bafaf6ee5e819dfa4437fc1412
Parents: 521047c
Author: Dave Johnson <sn...@apache.org>
Authored: Thu Nov 3 17:56:03 2016 -0400
Committer: Dave Johnson <sn...@apache.org>
Committed: Thu Nov 3 17:56:03 2016 -0400
----------------------------------------------------------------------
.../qakka/core/impl/QueueManagerImpl.java | 4 +-
.../distributed/DistributedQueueService.java | 4 -
.../qakka/distributed/actors/QueueActor.java | 93 ++------------
.../distributed/actors/QueueActorHelper.java | 126 ++++++++++++++++++-
.../distributed/actors/QueueActorRouter.java | 76 ++++++++++-
.../distributed/actors/QueueRefresher.java | 95 +-------------
.../impl/DistributedQueueServiceImpl.java | 55 +++-----
7 files changed, 227 insertions(+), 226 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/usergrid/blob/522a5515/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/core/impl/QueueManagerImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/core/impl/QueueManagerImpl.java b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/core/impl/QueueManagerImpl.java
index d51fe2d..88d307c 100644
--- a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/core/impl/QueueManagerImpl.java
+++ b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/core/impl/QueueManagerImpl.java
@@ -96,7 +96,7 @@ public class QueueManagerImpl implements QueueManager {
messageCounterSerialization.incrementCounter( queue.getName(), DatabaseQueueMessage.Type.DEFAULT, 0L );
messageCounterSerialization.incrementCounter( queue.getName(), DatabaseQueueMessage.Type.INFLIGHT, 0L );
- distributedQueueService.initQueue( queue.getName() );
+ //distributedQueueService.initQueue( queue.getName() );
distributedQueueService.refreshQueue( queue.getName() );
}
@@ -105,7 +105,7 @@ public class QueueManagerImpl implements QueueManager {
queueSerialization.writeQueue(queue.toDatabaseQueue());
- distributedQueueService.initQueue( queue.getName() );
+ //distributedQueueService.initQueue( queue.getName() );
distributedQueueService.refreshQueue( queue.getName() );
}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/522a5515/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/distributed/DistributedQueueService.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/distributed/DistributedQueueService.java b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/distributed/DistributedQueueService.java
index b11dcff..984cea2 100644
--- a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/distributed/DistributedQueueService.java
+++ b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/distributed/DistributedQueueService.java
@@ -34,8 +34,6 @@ public interface DistributedQueueService {
void init();
- void initQueue(String queueName);
-
void refresh();
void shutdown();
@@ -57,6 +55,4 @@ public interface DistributedQueueService {
Status ackMessage(String queueName, UUID messageId);
Status requeueMessage(String queueName, UUID messageId);
-
- Status clearMessages(String queueName);
}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/522a5515/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/distributed/actors/QueueActor.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/distributed/actors/QueueActor.java b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/distributed/actors/QueueActor.java
index 248f9cd..92e8607 100644
--- a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/distributed/actors/QueueActor.java
+++ b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/distributed/actors/QueueActor.java
@@ -20,121 +20,52 @@
package org.apache.usergrid.persistence.qakka.distributed.actors;
import akka.actor.ActorRef;
-import akka.actor.Cancellable;
import akka.actor.Props;
import akka.actor.UntypedActor;
import com.codahale.metrics.Timer;
import com.google.inject.Inject;
-import com.google.inject.Injector;
import org.apache.usergrid.persistence.actorsystem.GuiceActorProducer;
-import org.apache.usergrid.persistence.qakka.App;
import org.apache.usergrid.persistence.qakka.MetricsService;
-import org.apache.usergrid.persistence.qakka.QakkaFig;
-import org.apache.usergrid.persistence.qakka.core.impl.InMemoryQueue;
import org.apache.usergrid.persistence.qakka.distributed.DistributedQueueService;
import org.apache.usergrid.persistence.qakka.distributed.messages.*;
import org.apache.usergrid.persistence.qakka.serialization.queuemessages.DatabaseQueueMessage;
-import org.apache.usergrid.persistence.qakka.serialization.queuemessages.MessageCounterSerialization;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import scala.concurrent.duration.Duration;
-import java.text.DecimalFormat;
import java.util.*;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicLong;
public class QueueActor extends UntypedActor {
private static final Logger logger = LoggerFactory.getLogger( QueueActor.class );
- private final QakkaFig qakkaFig;
- private final InMemoryQueue inMemoryQueue;
private final QueueActorHelper queueActorHelper;
private final MetricsService metricsService;
- private final MessageCounterSerialization messageCounterSerialization;
- private final Map<String, Cancellable> refreshSchedulersByQueueName = new HashMap<>();
- private final Map<String, Cancellable> timeoutSchedulersByQueueName = new HashMap<>();
- private final Map<String, Cancellable> shardAllocationSchedulersByQueueName = new HashMap<>();
-
- private final Map<String, ActorRef> queueReadersByQueueName = new HashMap<>();
+ //private final Map<String, ActorRef> queueReadersByQueueName = new HashMap<>();
private final Map<String, ActorRef> queueTimeoutersByQueueName = new HashMap<>();
private final Map<String, ActorRef> shardAllocatorsByQueueName = new HashMap<>();
- private final Set<String> queuesSeen = new HashSet<>();
-
@Inject
public QueueActor(
- QakkaFig qakkaFig,
- InMemoryQueue inMemoryQueue,
QueueActorHelper queueActorHelper,
- MetricsService metricsService,
- MessageCounterSerialization messageCounterSerialization
+ MetricsService metricsService
) {
- this.qakkaFig = qakkaFig;
- this.inMemoryQueue = inMemoryQueue;
this.queueActorHelper = queueActorHelper;
this.metricsService = metricsService;
- this.messageCounterSerialization = messageCounterSerialization;
}
@Override
public void onReceive(Object message) {
- if ( message instanceof QueueInitRequest) {
- QueueInitRequest request = (QueueInitRequest)message;
-
- queuesSeen.add( request.getQueueName() );
-
- if ( refreshSchedulersByQueueName.get( request.getQueueName() ) == null ) {
- Cancellable scheduler = getContext().system().scheduler().schedule(
- Duration.create( 0, TimeUnit.MILLISECONDS),
- Duration.create( qakkaFig.getQueueRefreshMilliseconds(), TimeUnit.MILLISECONDS),
- self(),
- new QueueRefreshRequest( request.getQueueName(), false ),
- getContext().dispatcher(),
- getSelf());
- refreshSchedulersByQueueName.put( request.getQueueName(), scheduler );
- logger.debug("Created refresher for queue {}", request.getQueueName() );
- }
-
- if ( timeoutSchedulersByQueueName.get( request.getQueueName() ) == null ) {
- Cancellable scheduler = getContext().system().scheduler().schedule(
- Duration.create( 0, TimeUnit.MILLISECONDS),
- Duration.create( qakkaFig.getQueueTimeoutSeconds()/2, TimeUnit.SECONDS),
- self(),
- new QueueTimeoutRequest( request.getQueueName() ),
- getContext().dispatcher(),
- getSelf());
- timeoutSchedulersByQueueName.put( request.getQueueName(), scheduler );
- logger.debug("Created timeouter for queue {}", request.getQueueName() );
- }
-
- if ( shardAllocationSchedulersByQueueName.get( request.getQueueName() ) == null ) {
- Cancellable scheduler = getContext().system().scheduler().schedule(
- Duration.create( 0, TimeUnit.MILLISECONDS),
- Duration.create( qakkaFig.getShardAllocationCheckFrequencyMillis(), TimeUnit.MILLISECONDS),
- self(),
- new ShardCheckRequest( request.getQueueName() ),
- getContext().dispatcher(),
- getSelf());
- shardAllocationSchedulersByQueueName.put( request.getQueueName(), scheduler );
- logger.debug("Created shard allocater for queue {}", request.getQueueName() );
- }
-
-
- } else if ( message instanceof QueueRefreshRequest ) {
+ if ( message instanceof QueueRefreshRequest ) {
QueueRefreshRequest request = (QueueRefreshRequest)message;
- queuesSeen.add( request.getQueueName() );
-
-// // NOT asynchronous
-// queueActorHelper.queueRefresh( request.getQueueName() );
- if ( queueReadersByQueueName.get( request.getQueueName() ) == null ) {
+ // NOT asynchronous because we want this to happen locally in this JVM
+ queueActorHelper.queueRefresh( request.getQueueName() );
+ /* if ( queueReadersByQueueName.get( request.getQueueName() ) == null ) {
if ( !request.isOnlyIfEmpty() || inMemoryQueue.peek( request.getQueueName()) == null ) {
ActorRef readerRef = getContext().actorOf(
Props.create( GuiceActorProducer.class, QueueRefresher.class ),
@@ -142,16 +73,13 @@ public class QueueActor extends UntypedActor {
queueReadersByQueueName.put( request.getQueueName(), readerRef );
}
}
-
// hand-off to queue's reader
- queueReadersByQueueName.get( request.getQueueName() ).tell( request, self() );
+ queueReadersByQueueName.get( request.getQueueName() ).tell( request, self() ); */
} else if ( message instanceof QueueTimeoutRequest ) {
QueueTimeoutRequest request = (QueueTimeoutRequest)message;
- queuesSeen.add( request.getQueueName() );
-
if ( queueTimeoutersByQueueName.get( request.getQueueName() ) == null ) {
ActorRef readerRef = getContext().actorOf(
Props.create( GuiceActorProducer.class, QueueTimeouter.class),
@@ -166,8 +94,6 @@ public class QueueActor extends UntypedActor {
} else if ( message instanceof ShardCheckRequest ) {
ShardCheckRequest request = (ShardCheckRequest)message;
- queuesSeen.add( request.getQueueName() );
-
if ( shardAllocatorsByQueueName.get( request.getQueueName() ) == null ) {
ActorRef readerRef = getContext().actorOf(
Props.create( GuiceActorProducer.class, ShardAllocator.class),
@@ -186,12 +112,11 @@ public class QueueActor extends UntypedActor {
String queueName = queueGetRequest.getQueueName();
int numRequested = queueGetRequest.getNumRequested();
- queuesSeen.add( queueName );
-
Timer.Context timer = metricsService.getMetricRegistry().timer( MetricsService.GET_TIME_GET ).time();
try {
Collection<DatabaseQueueMessage> messages = queueActorHelper.getMessages( queueName, numRequested);
+ logger.trace("Returning queue {} messages {}", queueName, messages.size() );
getSender().tell( new QueueGetResponse(
DistributedQueueService.Status.SUCCESS, messages, queueName ), getSender() );
@@ -206,4 +131,6 @@ public class QueueActor extends UntypedActor {
}
+
+
}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/522a5515/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/distributed/actors/QueueActorHelper.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/distributed/actors/QueueActorHelper.java b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/distributed/actors/QueueActorHelper.java
index 6382661..4696a67 100644
--- a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/distributed/actors/QueueActorHelper.java
+++ b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/distributed/actors/QueueActorHelper.java
@@ -21,6 +21,7 @@ package org.apache.usergrid.persistence.qakka.distributed.actors;
import com.codahale.metrics.Timer;
import com.google.inject.Inject;
+import com.google.inject.Singleton;
import org.apache.usergrid.persistence.actorsystem.ActorSystemFig;
import org.apache.usergrid.persistence.qakka.MetricsService;
import org.apache.usergrid.persistence.qakka.QakkaFig;
@@ -37,12 +38,10 @@ import org.apache.usergrid.persistence.qakka.serialization.sharding.ShardIterato
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Optional;
-import java.util.UUID;
+import java.util.*;
+@Singleton
public class QueueActorHelper {
private static final Logger logger = LoggerFactory.getLogger( QueueActorHelper.class );
@@ -54,6 +53,8 @@ public class QueueActorHelper {
private final MetricsService metricsService;
private final CassandraClient cassandraClient;
+ Map<String, Long> startingShards = new HashMap<>();
+
@Inject
public QueueActorHelper(
@@ -97,6 +98,16 @@ public class QueueActorHelper {
Collection<DatabaseQueueMessage> getMessages(String queueName, int numRequested ) {
+ if ( qakkaFig.getInMemoryCache() ) {
+ return getMessagesFromMemory( queueName, numRequested );
+ } else {
+ return getMessagesFromStorage( queueName, numRequested );
+ }
+ }
+
+
+ Collection<DatabaseQueueMessage> getMessagesFromMemory(String queueName, int numRequested ) {
+
Collection<DatabaseQueueMessage> queueMessages = new ArrayList<>();
while (queueMessages.size() < numRequested) {
@@ -117,7 +128,51 @@ public class QueueActorHelper {
//logger.debug("{} returning {} for queue {}", this, queueMessages.size(), queueName);
return queueMessages;
+ }
+
+
+ Collection<DatabaseQueueMessage> getMessagesFromStorage(String queueName, int numRequested ) {
+
+ Collection<DatabaseQueueMessage> queueMessages = new ArrayList<>();
+
+ final Optional shardIdOptional;
+ final String shardKey =
+ createShardKey( queueName, Shard.Type.DEFAULT, actorSystemFig.getRegionLocal() );
+
+ Long shardId = startingShards.get( shardKey );
+ if ( shardId != null ) {
+ shardIdOptional = Optional.of( shardId );
+ } else {
+ shardIdOptional = Optional.empty();
+ }
+
+ String region = actorSystemFig.getRegionLocal();
+
+ ShardIterator shardIterator = new ShardIterator(
+ cassandraClient, queueName, region, Shard.Type.DEFAULT, shardIdOptional );
+
+ MultiShardMessageIterator multiShardIterator = new MultiShardMessageIterator(
+ cassandraClient, queueName, region, DatabaseQueueMessage.Type.DEFAULT, shardIterator, null);
+
+ int count = 0;
+
+ while ( multiShardIterator.hasNext() && count < numRequested ) {
+ DatabaseQueueMessage queueMessage = multiShardIterator.next();
+ if ( queueMessage != null && putInflight( queueMessage ) ) {
+ queueMessages.add( queueMessage );
+ count++;
+ }
+ }
+
+ Shard currentShard = multiShardIterator.getCurrentShard();
+ if ( currentShard != null ) {
+ shardId = currentShard.getShardId();
+ startingShards.put( shardKey, shardId );
+ }
+
+ //logger.debug("{} returning {} for queue {}", this, queueMessages.size(), queueName);
+ return queueMessages;
}
@@ -208,4 +263,67 @@ public class QueueActorHelper {
return DistributedQueueService.Status.ERROR;
}
}
+
+
+ void queueRefresh( String queueName ) {
+
+ Timer.Context timer = metricsService.getMetricRegistry().timer( MetricsService.REFRESH_TIME).time();
+
+ try {
+
+ if (inMemoryQueue.size( queueName ) < qakkaFig.getQueueInMemorySize()) {
+
+ final Optional shardIdOptional;
+ final String shardKey =
+ createShardKey( queueName, Shard.Type.DEFAULT, actorSystemFig.getRegionLocal() );
+ Long shardId = startingShards.get( shardKey );
+
+ if ( shardId != null ) {
+ shardIdOptional = Optional.of( shardId );
+ } else {
+ shardIdOptional = Optional.empty();
+ }
+
+ ShardIterator shardIterator = new ShardIterator(
+ cassandraClient, queueName, actorSystemFig.getRegionLocal(),
+ Shard.Type.DEFAULT, shardIdOptional );
+
+ UUID since = inMemoryQueue.getNewest( queueName );
+
+ String region = actorSystemFig.getRegionLocal();
+
+ MultiShardMessageIterator multiShardIterator = new MultiShardMessageIterator(
+ cassandraClient, queueName, region, DatabaseQueueMessage.Type.DEFAULT,
+ shardIterator, since);
+
+
+ int need = qakkaFig.getQueueInMemorySize() - inMemoryQueue.size( queueName );
+ int count = 0;
+
+ while ( multiShardIterator.hasNext() && count < need ) {
+ DatabaseQueueMessage queueMessage = multiShardIterator.next();
+ inMemoryQueue.add( queueName, queueMessage );
+ count++;
+ }
+
+ Shard currentShard = multiShardIterator.getCurrentShard();
+ if ( currentShard != null ) {
+ shardId = currentShard.getShardId();
+ startingShards.put( shardKey, shardId );
+ }
+
+ logger.trace("Refreshed queue {} region {} shard {} since {} found {}",
+ queueName, region, shardId, since, count );
+ }
+
+ } finally {
+ timer.close();
+ }
+
+ }
+
+ private String createShardKey(String queueName, Shard.Type type, String region ) {
+ return queueName + "_" + type + region;
+ }
+
}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/522a5515/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/distributed/actors/QueueActorRouter.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/distributed/actors/QueueActorRouter.java b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/distributed/actors/QueueActorRouter.java
index b5b9c30..71cf332 100644
--- a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/distributed/actors/QueueActorRouter.java
+++ b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/distributed/actors/QueueActorRouter.java
@@ -20,6 +20,7 @@
package org.apache.usergrid.persistence.qakka.distributed.actors;
import akka.actor.ActorRef;
+import akka.actor.Cancellable;
import akka.actor.Props;
import akka.actor.UntypedActor;
import akka.routing.ConsistentHashingRouter;
@@ -27,23 +28,43 @@ import akka.routing.FromConfig;
import com.google.inject.Inject;
import com.google.inject.Injector;
import org.apache.usergrid.persistence.actorsystem.GuiceActorProducer;
+import org.apache.usergrid.persistence.qakka.QakkaFig;
+import org.apache.usergrid.persistence.qakka.core.impl.InMemoryQueue;
import org.apache.usergrid.persistence.qakka.distributed.impl.QueueActorRouterProducer;
import org.apache.usergrid.persistence.qakka.distributed.messages.*;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import scala.concurrent.duration.Duration;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
/**
* Use consistent hashing to route messages to QueueActors
*/
public class QueueActorRouter extends UntypedActor {
+ private static final Logger logger = LoggerFactory.getLogger( QueueActorRouter.class );
private final ActorRef routerRef;
private final QueueActorRouterProducer queueActorRouterProducer;
+ private final QakkaFig qakkaFig;
+ private final Set<String> queuesSeen = new HashSet<>();
+
+ private final Map<String, Cancellable> refreshSchedulersByQueueName = new HashMap<>();
+ private final Map<String, Cancellable> timeoutSchedulersByQueueName = new HashMap<>();
+ private final Map<String, Cancellable> shardAllocationSchedulersByQueueName = new HashMap<>();
+
@Inject
- public QueueActorRouter( QueueActorRouterProducer queueActorRouterProducer ) {
+ public QueueActorRouter( QueueActorRouterProducer queueActorRouterProducer, QakkaFig qakkaFig ) {
this.queueActorRouterProducer = queueActorRouterProducer;
+ this.qakkaFig = qakkaFig;
this.routerRef = getContext().actorOf( FromConfig.getInstance().props(
Props.create( GuiceActorProducer.class, QueueActor.class)
@@ -56,6 +77,10 @@ public class QueueActorRouter extends UntypedActor {
if ( queueActorRouterProducer.getMessageTypes().contains( message.getClass() ) ) {
QakkaMessage qakkaMessage = (QakkaMessage) message;
+ if ( qakkaMessage.getQueueName() != null ) {
+ initIfNeeded( qakkaMessage.getQueueName() );
+ }
+
ConsistentHashingRouter.ConsistentHashableEnvelope envelope =
new ConsistentHashingRouter.ConsistentHashableEnvelope( message, qakkaMessage.getQueueName() );
routerRef.tell( envelope, getSender() );
@@ -64,4 +89,53 @@ public class QueueActorRouter extends UntypedActor {
unhandled(message);
}
}
+
+
+ /**
+ * Create scheduled refresh, timeout and shard-allocation tasks just in time.
+ */
+ private void initIfNeeded( String queueName ) {
+
+ if (!queuesSeen.contains( queueName )) {
+
+ queuesSeen.add( queueName );
+
+ if ( qakkaFig.getInMemoryCache() && refreshSchedulersByQueueName.get( queueName ) == null) {
+ Cancellable scheduler = getContext().system().scheduler().schedule(
+ Duration.create( 0, TimeUnit.MILLISECONDS ),
+ Duration.create( qakkaFig.getQueueRefreshMilliseconds(), TimeUnit.MILLISECONDS ),
+ self(),
+ new QueueRefreshRequest( queueName, false ),
+ getContext().dispatcher(),
+ getSelf() );
+ refreshSchedulersByQueueName.put( queueName, scheduler );
+ logger.debug( "Created refresher for queue {}", queueName );
+ }
+
+ if ( timeoutSchedulersByQueueName.get( queueName ) == null) {
+ Cancellable scheduler = getContext().system().scheduler().schedule(
+ Duration.create( 0, TimeUnit.MILLISECONDS ),
+ Duration.create( qakkaFig.getQueueTimeoutSeconds() / 2, TimeUnit.SECONDS ),
+ self(),
+ new QueueTimeoutRequest( queueName ),
+ getContext().dispatcher(),
+ getSelf() );
+ timeoutSchedulersByQueueName.put( queueName, scheduler );
+ logger.debug( "Created timeouter for queue {}", queueName );
+ }
+
+ if ( shardAllocationSchedulersByQueueName.get( queueName ) == null) {
+ Cancellable scheduler = getContext().system().scheduler().schedule(
+ Duration.create( 0, TimeUnit.MILLISECONDS ),
+ Duration.create( qakkaFig.getShardAllocationCheckFrequencyMillis(), TimeUnit.MILLISECONDS ),
+ self(),
+ new ShardCheckRequest( queueName ),
+ getContext().dispatcher(),
+ getSelf() );
+ shardAllocationSchedulersByQueueName.put( queueName, scheduler );
+ logger.debug( "Created shard allocater for queue {}", queueName );
+ }
+ }
+ }
+
}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/522a5515/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/distributed/actors/QueueRefresher.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/distributed/actors/QueueRefresher.java b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/distributed/actors/QueueRefresher.java
index 509ccd9..86f94f1 100644
--- a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/distributed/actors/QueueRefresher.java
+++ b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/distributed/actors/QueueRefresher.java
@@ -20,52 +20,21 @@
package org.apache.usergrid.persistence.qakka.distributed.actors;
import akka.actor.UntypedActor;
-import com.codahale.metrics.Timer;
import com.google.inject.Inject;
-import org.apache.usergrid.persistence.actorsystem.ActorSystemFig;
-import org.apache.usergrid.persistence.qakka.MetricsService;
-import org.apache.usergrid.persistence.qakka.QakkaFig;
-import org.apache.usergrid.persistence.qakka.core.CassandraClient;
-import org.apache.usergrid.persistence.qakka.core.impl.InMemoryQueue;
import org.apache.usergrid.persistence.qakka.distributed.messages.QueueRefreshRequest;
-import org.apache.usergrid.persistence.qakka.serialization.MultiShardMessageIterator;
-import org.apache.usergrid.persistence.qakka.serialization.auditlog.AuditLogSerialization;
-import org.apache.usergrid.persistence.qakka.serialization.queuemessages.DatabaseQueueMessage;
-import org.apache.usergrid.persistence.qakka.serialization.queuemessages.QueueMessageSerialization;
-import org.apache.usergrid.persistence.qakka.serialization.sharding.Shard;
-import org.apache.usergrid.persistence.qakka.serialization.sharding.ShardIterator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Optional;
-import java.util.UUID;
-
public class QueueRefresher extends UntypedActor {
private static final Logger logger = LoggerFactory.getLogger( QueueRefresher.class );
- private final ActorSystemFig actorSystemFig;
- private final InMemoryQueue inMemoryQueue;
- private final QakkaFig qakkaFig;
- private final MetricsService metricsService;
- private final CassandraClient cassandraClient;
+ private final QueueActorHelper queueActorHelper;
@Inject
- public QueueRefresher(
- ActorSystemFig actorSystemFig,
- InMemoryQueue inMemoryQueue,
- QakkaFig qakkaFig,
- MetricsService metricsService,
- CassandraClient cassandraClient
- ) {
- this.actorSystemFig = actorSystemFig;
- this.inMemoryQueue = inMemoryQueue;
- this.qakkaFig = qakkaFig;
- this.metricsService = metricsService;
- this.cassandraClient = cassandraClient;
+ public QueueRefresher( QueueActorHelper queueActorHelper ) {
+ this.queueActorHelper = queueActorHelper;
}
@@ -83,64 +52,8 @@ public class QueueRefresher extends UntypedActor {
}
}
- Map<String, Long> startingShards = new HashMap<>();
-
void queueRefresh( String queueName ) {
-
- Timer.Context timer = metricsService.getMetricRegistry().timer( MetricsService.REFRESH_TIME).time();
-
- try {
-
- if (inMemoryQueue.size( queueName ) < qakkaFig.getQueueInMemorySize()) {
-
- final Optional shardIdOptional;
- final String shardKey =
- createShardKey( queueName, Shard.Type.DEFAULT, actorSystemFig.getRegionLocal() );
- Long shardId = startingShards.get( shardKey );
-
- if ( shardId != null ) {
- shardIdOptional = Optional.of( shardId );
- } else {
- shardIdOptional = Optional.empty();
- }
-
- ShardIterator shardIterator = new ShardIterator(
- cassandraClient, queueName, actorSystemFig.getRegionLocal(),
- Shard.Type.DEFAULT, shardIdOptional );
-
- UUID since = inMemoryQueue.getNewest( queueName );
-
- String region = actorSystemFig.getRegionLocal();
-
- MultiShardMessageIterator multiShardIterator = new MultiShardMessageIterator(
- cassandraClient, queueName, region, DatabaseQueueMessage.Type.DEFAULT,
- shardIterator, since);
-
-
- int need = qakkaFig.getQueueInMemorySize() - inMemoryQueue.size( queueName );
- int count = 0;
-
- while ( multiShardIterator.hasNext() && count < need ) {
- DatabaseQueueMessage queueMessage = multiShardIterator.next();
- inMemoryQueue.add( queueName, queueMessage );
- count++;
- }
-
- startingShards.put( shardKey, shardId );
-
- logger.debug("Refreshed queue {} region {} shard {} since {} found {}",
- queueName, region, shardId, since, count );
- }
-
- } finally {
- timer.close();
- }
-
+ queueActorHelper.queueRefresh( queueName );
}
-
- private String createShardKey(String queueName, Shard.Type type, String region ) {
- return queueName + "_" + type + region;
- }
-
}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/522a5515/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/distributed/impl/DistributedQueueServiceImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/distributed/impl/DistributedQueueServiceImpl.java b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/distributed/impl/DistributedQueueServiceImpl.java
index 51f6fd3..96e8cab 100644
--- a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/distributed/impl/DistributedQueueServiceImpl.java
+++ b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/distributed/impl/DistributedQueueServiceImpl.java
@@ -80,22 +80,6 @@ public class DistributedQueueServiceImpl implements DistributedQueueService {
@Override
public void init() {
- try {
- List<String> queues = queueManager.getListOfQueues();
- for (String queueName : queues) {
- initQueue( queueName );
- }
- } catch (InvalidQueryException e) {
-
- if (e.getMessage().contains( "unconfigured columnfamily" )) {
- logger.info( "Unable to initialize queues since system is bootstrapping. " +
- "Queues will be initialized when created" );
- } else {
- throw e;
- }
-
- }
-
StringBuilder logMessage = new StringBuilder();
logMessage.append( "DistributedQueueServiceImpl initialized with config:\n" );
Method[] methods = qakkaFig.getClass().getMethods();
@@ -114,15 +98,6 @@ public class DistributedQueueServiceImpl implements DistributedQueueService {
@Override
- public void initQueue(String queueName) {
- logger.info("Initializing queue: {}", queueName);
- QueueInitRequest request = new QueueInitRequest( queueName );
- ActorRef clientActor = actorSystemManager.getClientActor();
- clientActor.tell( request, null );
- }
-
-
- @Override
public void refresh() {
for ( String queueName : queueManager.getListOfQueues() ) {
refreshQueue( queueName );
@@ -132,10 +107,13 @@ public class DistributedQueueServiceImpl implements DistributedQueueService {
@Override
public void refreshQueue(String queueName) {
- logger.info("{} Requesting refresh for queue: {}", this, queueName);
- QueueRefreshRequest request = new QueueRefreshRequest( queueName, false );
- ActorRef clientActor = actorSystemManager.getClientActor();
- clientActor.tell( request, null );
+ if ( qakkaFig.getInMemoryCache() ) {
+ logger.trace( "{} Requesting refresh for queue: {}", this, queueName );
+ QueueRefreshRequest request = new QueueRefreshRequest( queueName, false );
+ ActorRef clientActor = actorSystemManager.getClientActor();
+ clientActor.tell( request, null );
+ }
+
}
@@ -157,6 +135,8 @@ public class DistributedQueueServiceImpl implements DistributedQueueService {
String queueName, String sourceRegion, String destRegion, UUID messageId,
Long deliveryTime, Long expirationTime ) {
+ logger.trace("Sending message to queue {} region {}", queueName, destRegion);
+
Timer.Context timer = metricsService.getMetricRegistry().timer( MetricsService.SEND_TIME_TOTAL ).time();
try {
@@ -186,9 +166,11 @@ public class DistributedQueueServiceImpl implements DistributedQueueService {
logger.debug("SUCCESS after {} retries", retries );
}
- // send refresh-queue-if-empty message
- QueueRefreshRequest qrr = new QueueRefreshRequest( queueName, false );
- clientActor.tell( qrr, null );
+ if ( qakkaFig.getInMemoryCache() ) {
+ // send refresh-queue-if-empty message
+ QueueRefreshRequest qrr = new QueueRefreshRequest( queueName, false );
+ clientActor.tell( qrr, null );
+ }
return qarm.getSendStatus();
@@ -280,7 +262,6 @@ public class DistributedQueueServiceImpl implements DistributedQueueService {
logger.warn( "getNextMessage {} SUCCESS after {} tries", queueName, tries );
}
}
- logger.trace("Returning queue {} messages {}", queueName, qprm.getQueueMessages().size());
return qprm.getQueueMessages();
@@ -335,14 +316,6 @@ public class DistributedQueueServiceImpl implements DistributedQueueService {
}
- @Override
- public Status clearMessages(String queueName) {
-
- // TODO: implement clear queue
- throw new UnsupportedOperationException();
- }
-
-
private Status sendMessageToLocalRouters( QakkaMessage message ) {
int maxRetries = 5;