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/09 15:47:36 UTC
[2/4] usergrid git commit: Rename send message methods to
sendMessageToAllRegions() and sendMessageToLocalRegion(),
and fix implementation of both in Hakka imply.
Rename send message methods to sendMessageToAllRegions() and sendMessageToLocalRegion(), and fix implementation of both in Hakka imply.
Project: http://git-wip-us.apache.org/repos/asf/usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/usergrid/commit/5a81eb29
Tree: http://git-wip-us.apache.org/repos/asf/usergrid/tree/5a81eb29
Diff: http://git-wip-us.apache.org/repos/asf/usergrid/diff/5a81eb29
Branch: refs/heads/usergrid-1318-queue
Commit: 5a81eb2980b05cf34684c2c47489fed21daf6585
Parents: ae16458
Author: Dave Johnson <sn...@apache.org>
Authored: Wed Nov 9 10:45:58 2016 -0500
Committer: Dave Johnson <sn...@apache.org>
Committed: Wed Nov 9 10:45:58 2016 -0500
----------------------------------------------------------------------
.../asyncevents/AsyncEventServiceImpl.java | 37 +++++++++++++++++---
.../usergrid/persistence/qakka/QakkaFig.java | 16 ++++++---
.../persistence/queue/LegacyQueueManager.java | 4 +--
.../persistence/queue/LocalQueueManager.java | 6 ++--
.../queue/impl/QakkaQueueManager.java | 26 ++++++++------
.../queue/impl/SNSQueueManagerImpl.java | 6 ++--
.../queue/LegacyQueueManagerTest.java | 4 +--
.../impl/ApplicationQueueManagerImpl.java | 2 +-
.../services/queues/ImportQueueManager.java | 4 +--
9 files changed, 71 insertions(+), 34 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/usergrid/blob/5a81eb29/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AsyncEventServiceImpl.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AsyncEventServiceImpl.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AsyncEventServiceImpl.java
index 9f931d3..66b527b 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AsyncEventServiceImpl.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AsyncEventServiceImpl.java
@@ -200,7 +200,7 @@ public class AsyncEventServiceImpl implements AsyncEventService {
try {
//signal to SQS
- this.queue.sendMessage( operation );
+ this.queue.sendMessageToLocalRegion( operation );
} catch (IOException e) {
throw new RuntimeException("Unable to queue message", e);
} finally {
@@ -215,9 +215,9 @@ public class AsyncEventServiceImpl implements AsyncEventService {
try {
//signal to SQS
if (forUtilityQueue) {
- this.utilityQueue.sendMessageToTopic(operation);
+ this.utilityQueue.sendMessageToAllRegions(operation);
} else {
- this.queue.sendMessageToTopic(operation);
+ this.queue.sendMessageToAllRegions(operation);
}
}
catch ( IOException e ) {
@@ -461,6 +461,11 @@ public class AsyncEventServiceImpl implements AsyncEventService {
public void queueInitializeApplicationIndex( final ApplicationScope applicationScope) {
IndexLocationStrategy indexLocationStrategy = indexLocationStrategyFactory.getIndexLocationStrategy(
applicationScope);
+
+
+ logger.trace("Offering InitializeApplicationIndexEvent for {}:{}",
+ applicationScope.getApplication().getUuid(), applicationScope.getApplication().getType());
+
offerTopic( new InitializeApplicationIndexEvent( queueFig.getPrimaryRegion(),
new ReplicatedIndexLocationStrategy( indexLocationStrategy ) ), false);
}
@@ -471,7 +476,11 @@ public class AsyncEventServiceImpl implements AsyncEventService {
final Entity entity, long updatedAfter) {
- offer(new EntityIndexEvent(queueFig.getPrimaryRegion(),new EntityIdScope(applicationScope, entity.getId()), 0));
+ logger.trace("Offering EntityIndexEvent for {}:{}",
+ entity.getId().getUuid(), entity.getId().getType());
+
+ offer(new EntityIndexEvent(queueFig.getPrimaryRegion(),
+ new EntityIdScope(applicationScope, entity.getId()), 0));
final EntityIndexOperation entityIndexOperation =
new EntityIndexOperation( applicationScope, entity.getId(), updatedAfter);
@@ -517,6 +526,9 @@ public class AsyncEventServiceImpl implements AsyncEventService {
final Entity entity,
final Edge newEdge) {
+ logger.trace("Offering EdgeIndexEvent for edge type {} entity {}:{}",
+ newEdge.getType(), entity.getId().getUuid(), entity.getId().getType());
+
offer( new EdgeIndexEvent( queueFig.getPrimaryRegion(), applicationScope, entity.getId(), newEdge ));
}
@@ -549,6 +561,9 @@ public class AsyncEventServiceImpl implements AsyncEventService {
public void queueDeleteEdge(final ApplicationScope applicationScope,
final Edge edge) {
+ logger.trace("Offering EdgeDeleteEvent for type {} to target {}:{}",
+ edge.getType(), edge.getTargetNode().getUuid(), edge.getTargetNode().getType());
+
// sent in region (not offerTopic) as the delete IO happens in-region, then queues a multi-region de-index op
offer( new EdgeDeleteEvent( queueFig.getPrimaryRegion(), applicationScope, edge ) );
}
@@ -610,6 +625,8 @@ public class AsyncEventServiceImpl implements AsyncEventService {
//send to the topic so all regions index the batch
+ logger.trace("Offering ElasticsearchIndexEvent for message {}", newMessageId );
+
offerTopic( elasticsearchIndexEvent, forUtilityQueue );
}
@@ -678,6 +695,10 @@ public class AsyncEventServiceImpl implements AsyncEventService {
public void queueDeIndexOldVersion(final ApplicationScope applicationScope, final Id entityId, UUID markedVersion) {
// queue the de-index of old versions to the topic so cleanup happens in all regions
+
+ logger.trace("Offering DeIndexOldVersionsEvent for app {} {}:{}",
+ applicationScope.getApplication().getUuid(), entityId.getUuid(), entityId.getType());
+
offerTopic( new DeIndexOldVersionsEvent( queueFig.getPrimaryRegion(),
new EntityIdScope( applicationScope, entityId), markedVersion), false);
@@ -736,6 +757,8 @@ public class AsyncEventServiceImpl implements AsyncEventService {
@Override
public void queueEntityDelete(final ApplicationScope applicationScope, final Id entityId) {
+ logger.trace("Offering EntityDeleteEvent for {}:{}", entityId.getUuid(), entityId.getType());
+
// sent in region (not offerTopic) as the delete IO happens in-region, then queues a multi-region de-index op
offer( new EntityDeleteEvent(queueFig.getPrimaryRegion(), new EntityIdScope( applicationScope, entityId ) ) );
}
@@ -894,7 +917,8 @@ public class AsyncEventServiceImpl implements AsyncEventService {
submitToIndex( indexEventResults, isUtilityQueue );
if ( messagesToAck.size() < messages.size() ) {
- logger.warn( "Missing {} message(s) from index processing",
+ logger.warn(
+ "Missing {} message(s) from index processing",
messages.size() - messagesToAck.size() );
}
@@ -916,6 +940,7 @@ public class AsyncEventServiceImpl implements AsyncEventService {
//do not rethrow so we can process all of them
}
} ).subscribeOn( rxTaskScheduler.getAsyncIOScheduler() );
+
//end flatMap
}, indexProcessorFig.getEventConcurrencyFactor() );
@@ -983,6 +1008,8 @@ public class AsyncEventServiceImpl implements AsyncEventService {
});
+ logger.trace("Offering batch of EntityIndexEvent of size {}", batch.size());
+
offerBatch( batch, forUtilityQueue );
}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/5a81eb29/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 105c193..c034b92 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
@@ -40,9 +40,13 @@ public interface QakkaFig extends GuicyFig, Serializable {
String QUEUE_TIMEOUT_SECONDS = "queue.timeout.seconds";
- String QUEUE_REFRESH_MILLISECONDS = "queue.refresh.millis";
+ String QUEUE_REFRESH_MILLISECONDS = "queue.inmemory.refresh.millis";
- String QUEUE_INMEMORY_SIZE = "queue.inmemory.size";
+ String QUEUE_IN_MEMORY = "queue.inmemory.cache";
+
+ String QUEUE_INMEMORY_SIZE = "queue.inmemory.cache.size";
+
+ String QUEUE_IN_MEMORY_REFRESH_ASYNC = "queue.inmemory.cache.async";
String QUEUE_SEND_MAX_RETRIES = "queue.send.max.retries";
@@ -53,9 +57,11 @@ 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";
@@ -68,8 +74,6 @@ 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 */
@@ -172,4 +176,8 @@ public interface QakkaFig extends GuicyFig, Serializable {
@Key(QUEUE_IN_MEMORY)
@Default("false")
boolean getInMemoryCache();
+
+ @Key(QUEUE_IN_MEMORY_REFRESH_ASYNC)
+ @Default("true")
+ boolean getInMemoryRefreshAsync();
}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/5a81eb29/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/LegacyQueueManager.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/LegacyQueueManager.java b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/LegacyQueueManager.java
index e38d3bc..6627148 100644
--- a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/LegacyQueueManager.java
+++ b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/LegacyQueueManager.java
@@ -63,13 +63,13 @@ public interface LegacyQueueManager {
* @param body
* @throws IOException
*/
- <T extends Serializable> void sendMessage(T body)throws IOException;
+ <T extends Serializable> void sendMessageToLocalRegion(T body)throws IOException;
/**
* Send a messae to the topic to be sent to other queues
* @param body
*/
- <T extends Serializable> void sendMessageToTopic(T body) throws IOException;
+ <T extends Serializable> void sendMessageToAllRegions(T body) throws IOException;
/**
* purge messages
http://git-wip-us.apache.org/repos/asf/usergrid/blob/5a81eb29/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/LocalQueueManager.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/LocalQueueManager.java b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/LocalQueueManager.java
index 630c2e7..90621c0 100644
--- a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/LocalQueueManager.java
+++ b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/LocalQueueManager.java
@@ -85,7 +85,7 @@ public class LocalQueueManager implements LegacyQueueManager {
@Override
- public <T extends Serializable> void sendMessage( final T body ) throws IOException {
+ public <T extends Serializable> void sendMessageToLocalRegion(final T body ) throws IOException {
String uuid = UUID.randomUUID().toString();
try {
queue.offer(new LegacyQueueMessage(uuid, "handle_" + uuid, body, "put type here"),5000,TimeUnit.MILLISECONDS);
@@ -97,8 +97,8 @@ public class LocalQueueManager implements LegacyQueueManager {
@Override
- public <T extends Serializable> void sendMessageToTopic( final T body ) throws IOException {
- sendMessage( body );
+ public <T extends Serializable> void sendMessageToAllRegions(final T body ) throws IOException {
+ sendMessageToLocalRegion( body );
}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/5a81eb29/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 9d7a341..6900a46 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
@@ -21,12 +21,9 @@ import com.datastax.driver.core.DataType;
import com.datastax.driver.core.ProtocolVersion;
import com.google.inject.Inject;
import com.google.inject.assistedinject.Assisted;
-import org.apache.usergrid.persistence.qakka.QakkaFig;
import org.apache.usergrid.persistence.qakka.core.*;
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.queue.LegacyQueueFig;
import org.apache.usergrid.persistence.queue.LegacyQueueManager;
import org.apache.usergrid.persistence.queue.LegacyQueueMessage;
import org.apache.usergrid.persistence.queue.LegacyQueueScope;
@@ -77,10 +74,7 @@ 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() );
+ private <T extends Serializable> void doSendMessage( T body, List<String> regions ) throws IOException {
createQueueIfNecessary();
@@ -93,7 +87,7 @@ public class QakkaQueueManager implements LegacyQueueManager {
queueMessageManager.sendMessages(
scope.getName(),
- regions.getRegions( scope.getRegionImplementation().name() ),
+ regions,
null, // delay millis
null, // expiration seconds
"application/octet-stream",
@@ -102,8 +96,18 @@ public class QakkaQueueManager implements LegacyQueueManager {
@Override
- public <T extends Serializable> void sendMessageToTopic(T body) throws IOException {
- sendMessage( body );
+ public <T extends Serializable> void sendMessageToLocalRegion(T body) throws IOException {
+ List<String> regionsList = regions.getRegions( Regions.LOCAL );
+ logger.trace( "Sending message to queue {} local region {}", scope.getName(), regionsList );
+ doSendMessage( body, regionsList );
+ }
+
+
+ @Override
+ public <T extends Serializable> void sendMessageToAllRegions(T body) throws IOException {
+ List<String> regionsList = regions.getRegions( Regions.ALL );
+ logger.trace( "Sending message to queue {} all regions {}", scope.getName(), regionsList );
+ doSendMessage( body, regionsList );
}
@@ -175,7 +179,7 @@ public class QakkaQueueManager implements LegacyQueueManager {
public void sendMessages( List bodies ) throws IOException {
for ( Object body : bodies ) {
- sendMessage( (Serializable)body );
+ sendMessageToLocalRegion( (Serializable)body );
}
}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/5a81eb29/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/impl/SNSQueueManagerImpl.java
----------------------------------------------------------------------
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 637f157..853fcdd 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
@@ -539,7 +539,7 @@ public class SNSQueueManagerImpl implements LegacyQueueManager {
@Override
- public <T extends Serializable> void sendMessageToTopic( final T body ) throws IOException {
+ public <T extends Serializable> void sendMessageToAllRegions(final T body ) throws IOException {
if ( snsAsync == null ) {
logger.error( "SNS client is null, perhaps it failed to initialize successfully" );
return;
@@ -582,13 +582,13 @@ public class SNSQueueManagerImpl implements LegacyQueueManager {
}
for ( Object body : bodies ) {
- sendMessage( ( Serializable ) body );
+ sendMessageToLocalRegion( ( Serializable ) body );
}
}
@Override
- public <T extends Serializable> void sendMessage( final T body ) throws IOException {
+ public <T extends Serializable> void sendMessageToLocalRegion(final T body ) throws IOException {
if ( sqsAsync == null ) {
logger.error( "SQS client is null, perhaps it failed to initialize successfully" );
http://git-wip-us.apache.org/repos/asf/usergrid/blob/5a81eb29/stack/corepersistence/queue/src/test/java/org/apache/usergrid/persistence/queue/LegacyQueueManagerTest.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queue/src/test/java/org/apache/usergrid/persistence/queue/LegacyQueueManagerTest.java b/stack/corepersistence/queue/src/test/java/org/apache/usergrid/persistence/queue/LegacyQueueManagerTest.java
index 13fb195..535c30c 100644
--- a/stack/corepersistence/queue/src/test/java/org/apache/usergrid/persistence/queue/LegacyQueueManagerTest.java
+++ b/stack/corepersistence/queue/src/test/java/org/apache/usergrid/persistence/queue/LegacyQueueManagerTest.java
@@ -25,8 +25,6 @@ import com.google.inject.Injector;
import org.apache.usergrid.persistence.actorsystem.ActorSystemFig;
import org.apache.usergrid.persistence.qakka.AbstractTest;
import org.apache.usergrid.persistence.qakka.App;
-import org.apache.usergrid.persistence.qakka.core.CassandraClient;
-import org.apache.usergrid.persistence.qakka.core.CassandraClientImpl;
import org.apache.usergrid.persistence.qakka.distributed.DistributedQueueService;
import org.apache.usergrid.persistence.queue.impl.LegacyQueueScopeImpl;
import org.junit.Test;
@@ -68,7 +66,7 @@ public class LegacyQueueManagerTest extends AbstractTest {
LegacyQueueManager qm = qmf.getQueueManager(scope);
String value = "bodytest";
- qm.sendMessage(value);
+ qm.sendMessageToLocalRegion(value);
Thread.sleep(5000);
http://git-wip-us.apache.org/repos/asf/usergrid/blob/5a81eb29/stack/services/src/main/java/org/apache/usergrid/services/notifications/impl/ApplicationQueueManagerImpl.java
----------------------------------------------------------------------
diff --git a/stack/services/src/main/java/org/apache/usergrid/services/notifications/impl/ApplicationQueueManagerImpl.java b/stack/services/src/main/java/org/apache/usergrid/services/notifications/impl/ApplicationQueueManagerImpl.java
index 96e2dbd..d26cd5f 100644
--- a/stack/services/src/main/java/org/apache/usergrid/services/notifications/impl/ApplicationQueueManagerImpl.java
+++ b/stack/services/src/main/java/org/apache/usergrid/services/notifications/impl/ApplicationQueueManagerImpl.java
@@ -356,7 +356,7 @@ public class ApplicationQueueManagerImpl implements ApplicationQueueManager {
if(logger.isTraceEnabled()) {
logger.trace("Queueing notification message for device: {}", message.get().getDeviceId());
}
- qm.sendMessage( message.get() );
+ qm.sendMessageToLocalRegion( message.get() );
queueMeter.mark();
}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/5a81eb29/stack/services/src/main/java/org/apache/usergrid/services/queues/ImportQueueManager.java
----------------------------------------------------------------------
diff --git a/stack/services/src/main/java/org/apache/usergrid/services/queues/ImportQueueManager.java b/stack/services/src/main/java/org/apache/usergrid/services/queues/ImportQueueManager.java
index d9db84a..47758a3 100644
--- a/stack/services/src/main/java/org/apache/usergrid/services/queues/ImportQueueManager.java
+++ b/stack/services/src/main/java/org/apache/usergrid/services/queues/ImportQueueManager.java
@@ -65,13 +65,13 @@ public class ImportQueueManager implements LegacyQueueManager {
@Override
- public <T extends Serializable> void sendMessage( final T body ) throws IOException {
+ public <T extends Serializable> void sendMessageToLocalRegion(final T body ) throws IOException {
}
@Override
- public <T extends Serializable> void sendMessageToTopic( final T body ) throws IOException {
+ public <T extends Serializable> void sendMessageToAllRegions(final T body ) throws IOException {
}