Formatting, debug log statements, etc.
Project: http://git-wip-us.apache.org/repos/asf/usergrid/repo Commit: http://git-wip-us.apache.org/repos/asf/usergrid/commit/5d37ed58 Tree: http://git-wip-us.apache.org/repos/asf/usergrid/tree/5d37ed58 Diff: http://git-wip-us.apache.org/repos/asf/usergrid/diff/5d37ed58 Branch: refs/heads/master Commit: 5d37ed58d8649e9498e8a0fe7fe858f52762978c Parents: 7be8c27 Author: Dave Johnson <snoopd...@apache.org> Authored: Mon Oct 10 13:51:53 2016 -0400 Committer: Dave Johnson <snoopd...@apache.org> Committed: Mon Oct 10 13:51:53 2016 -0400 ---------------------------------------------------------------------- .../qakka/distributed/actors/QueueActorHelper.java | 2 +- .../persistence/qakka/distributed/actors/ShardAllocator.java | 7 ------- .../qakka/serialization/MultiShardMessageIterator.java | 4 ++++ .../queuemessages/impl/QueueMessageSerializationImpl.java | 3 +++ .../usergrid/persistence/queue/LegacyQueueManagerTest.java | 2 -- 5 files changed, 8 insertions(+), 10 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/usergrid/blob/5d37ed58/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 fcb3fba..6382661 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 @@ -115,7 +115,7 @@ public class QueueActorHelper { } } - logger.debug("{} returning {} for queue {}", this, queueMessages.size(), queueName); + //logger.debug("{} returning {} for queue {}", this, queueMessages.size(), queueName); return queueMessages; } http://git-wip-us.apache.org/repos/asf/usergrid/blob/5d37ed58/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/distributed/actors/ShardAllocator.java ---------------------------------------------------------------------- diff --git a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/distributed/actors/ShardAllocator.java b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/distributed/actors/ShardAllocator.java index 46dc0ed..1863472 100644 --- a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/distributed/actors/ShardAllocator.java +++ b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/distributed/actors/ShardAllocator.java @@ -74,13 +74,6 @@ public class ShardAllocator extends UntypedActor { this.shardCounterSerialization = shardCounterSerialization; this.metricsService = metricsService; this.cassandraClient = cassandraClient; - -// this.qakkaFig = injector.getInstance( QakkaFig.class ); -// this.shardCounterSerialization = injector.getInstance( ShardCounterSerializationImpl.class ); -// this.shardSerialization = injector.getInstance( ShardSerializationImpl.class ); -// this.actorSystemFig = injector.getInstance( ActorSystemFig.class ); -// this.metricsService = injector.getInstance( MetricsService.class ); -// this.cassandraClient = injector.getInstance( CassandraClientImpl.class ); } http://git-wip-us.apache.org/repos/asf/usergrid/blob/5d37ed58/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/serialization/MultiShardMessageIterator.java ---------------------------------------------------------------------- diff --git a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/serialization/MultiShardMessageIterator.java b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/serialization/MultiShardMessageIterator.java index 29327e2..6a066e9 100644 --- a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/serialization/MultiShardMessageIterator.java +++ b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/qakka/serialization/MultiShardMessageIterator.java @@ -133,6 +133,7 @@ public class MultiShardMessageIterator implements Iterator<DatabaseQueueMessage> .and(regionClause) .and(shardIdClause) .limit(PAGE_SIZE); + } else { Clause messageIdClause = QueryBuilder.gt( COLUMN_QUEUE_MESSAGE_ID, nextStart); @@ -144,8 +145,11 @@ public class MultiShardMessageIterator implements Iterator<DatabaseQueueMessage> .limit(PAGE_SIZE); } + List<Row> rows = cassandraClient.getQueueMessageSession().execute(query).all(); + //logger.debug("Query got {}: {}", rows.size(), query); + if ( (rows == null || rows.size() == 0) && shardIterator.hasNext()) { currentShard = shardIterator.next(); http://git-wip-us.apache.org/repos/asf/usergrid/blob/5d37ed58/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 33de7bc..fba2bed 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 @@ -150,6 +150,9 @@ public class QueueMessageSerializationImpl implements QueueMessageSerialization Statement insert = createWriteMessageStatement( message ); cassandraClient.getQueueMessageSession().execute(insert); +// logger.debug("Wrote queue {} queue message {} shardId {}", +// message.getQueueName(), message.getQueueMessageId(), message.getShardId() ); + shardCounterSerialization.incrementCounter( message.getQueueName(), shardType, message.getShardId(), 1 ); messageCounterSerialization.incrementCounter( message.getQueueName(), message.getType(), 1L ); http://git-wip-us.apache.org/repos/asf/usergrid/blob/5d37ed58/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 92075b6..13fb195 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 @@ -92,8 +92,6 @@ public class LegacyQueueManagerTest extends AbstractTest { Injector myInjector = getInjector(); - CassandraClient cassandraClient = myInjector.getInstance( CassandraClientImpl.class ); - ActorSystemFig actorSystemFig = myInjector.getInstance( ActorSystemFig.class ); String region = actorSystemFig.getRegionLocal();