You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@usergrid.apache.org by sf...@apache.org on 2015/10/13 18:14:38 UTC
[1/4] usergrid git commit: add a better status code check for index
overload
Repository: usergrid
Updated Branches:
refs/heads/remove-inmemory-event-service e17dab4c0 -> f6409ce35
add a better status code check for index overload
Project: http://git-wip-us.apache.org/repos/asf/usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/usergrid/commit/5b650860
Tree: http://git-wip-us.apache.org/repos/asf/usergrid/tree/5b650860
Diff: http://git-wip-us.apache.org/repos/asf/usergrid/diff/5b650860
Branch: refs/heads/remove-inmemory-event-service
Commit: 5b6508609c3ce1687993b4015f88126c8be6676d
Parents: 80324de
Author: Shawn Feldman <sf...@apache.org>
Authored: Fri Oct 9 11:34:16 2015 -0600
Committer: Shawn Feldman <sf...@apache.org>
Committed: Fri Oct 9 11:34:16 2015 -0600
----------------------------------------------------------------------
.../persistence/index/impl/EsIndexProducerImpl.java | 11 ++++++++---
1 file changed, 8 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/usergrid/blob/5b650860/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsIndexProducerImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsIndexProducerImpl.java b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsIndexProducerImpl.java
index e778948..031556a 100644
--- a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsIndexProducerImpl.java
+++ b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsIndexProducerImpl.java
@@ -27,6 +27,7 @@ import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.client.Client;
+import org.elasticsearch.rest.RestStatus;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -192,13 +193,17 @@ public class EsIndexProducerImpl implements IndexProducer {
final StringBuilder errorString = new StringBuilder( );
+ boolean hasTooManyRequests= false;
for ( BulkItemResponse response : responses ) {
if ( response.isFailed() ) {
// log error and continue processing
log.error( "Unable to index id={}, type={}, index={}, failureMessage={} ", response.getId(),
- response.getType(), response.getIndex(), response.getFailureMessage() );
-
+ response.getType(), response.getIndex(), response.getFailureMessage() );
+ //if index is overloaded on the queue fail.
+ if(response.getFailure()!=null && response.getFailure().getStatus() == RestStatus.TOO_MANY_REQUESTS){
+ hasTooManyRequests =true;
+ }
error = true;
errorString.append( response.getFailureMessage() ).append( "\n" );
@@ -206,7 +211,7 @@ public class EsIndexProducerImpl implements IndexProducer {
}
if ( error ) {
- if(errorString.lastIndexOf("rejected execution (queue capacity")>=0){
+ if(hasTooManyRequests){
try{
log.warn("Encountered Queue Capacity Exception from ElasticSearch slowing by "
+ indexFig.getSleepTimeForQueueError() );
[2/4] usergrid git commit: add new exception for empty sequences
Posted by sf...@apache.org.
add new exception for empty sequences
Project: http://git-wip-us.apache.org/repos/asf/usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/usergrid/commit/a2a07aac
Tree: http://git-wip-us.apache.org/repos/asf/usergrid/tree/a2a07aac
Diff: http://git-wip-us.apache.org/repos/asf/usergrid/diff/a2a07aac
Branch: refs/heads/remove-inmemory-event-service
Commit: a2a07aac1c7f53b5bbaffe0a9bc433d05777cbdf
Parents: 5b65086
Author: Shawn Feldman <sf...@apache.org>
Authored: Mon Oct 12 11:05:49 2015 -0600
Committer: Shawn Feldman <sf...@apache.org>
Committed: Mon Oct 12 11:05:49 2015 -0600
----------------------------------------------------------------------
.../corepersistence/asyncevents/AmazonAsyncEventService.java | 4 ++++
1 file changed, 4 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/usergrid/blob/a2a07aac/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AmazonAsyncEventService.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AmazonAsyncEventService.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AmazonAsyncEventService.java
index 4ee2094..45d83cd 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AmazonAsyncEventService.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AmazonAsyncEventService.java
@@ -302,6 +302,10 @@ public class AmazonAsyncEventService implements AsyncEventService {
.collect(() -> new IndexOperationMessage(), (collector, single ) -> collector.ingest(single))
.toBlocking().lastOrDefault(null);
+ if(indexOperationMessage == null){
+ throw new IllegalArgumentException("Received null index operation.");
+ }
+
//return type that can be indexed and ack'd later
return new IndexEventResult(Optional.fromNullable(message), Optional.fromNullable(indexOperationMessage), thisEvent.getCreationTime());
} catch (Exception e) {
[3/4] usergrid git commit: add better logging for empty sequences
Posted by sf...@apache.org.
add better logging for empty sequences
Project: http://git-wip-us.apache.org/repos/asf/usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/usergrid/commit/9cbe283d
Tree: http://git-wip-us.apache.org/repos/asf/usergrid/tree/9cbe283d
Diff: http://git-wip-us.apache.org/repos/asf/usergrid/diff/9cbe283d
Branch: refs/heads/remove-inmemory-event-service
Commit: 9cbe283d0af45b877fb93dfae81cdb343cec26c5
Parents: a2a07aa
Author: Shawn Feldman <sf...@apache.org>
Authored: Mon Oct 12 16:27:13 2015 -0600
Committer: Shawn Feldman <sf...@apache.org>
Committed: Mon Oct 12 16:27:13 2015 -0600
----------------------------------------------------------------------
.../asyncevents/AmazonAsyncEventService.java | 13 ++++++-----
.../asyncevents/EventBuilderImpl.java | 3 ++-
.../core/rx/ExceptionBehaviorTest.java | 24 ++++++++++++++++++++
.../persistence/queue/QueueMessage.java | 10 ++++++++
.../queue/impl/SNSQueueManagerImpl.java | 2 ++
.../queue/impl/SQSQueueManagerImpl.java | 1 +
6 files changed, 46 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/usergrid/blob/9cbe283d/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AmazonAsyncEventService.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AmazonAsyncEventService.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AmazonAsyncEventService.java
index 45d83cd..f1a02ce 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AmazonAsyncEventService.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AmazonAsyncEventService.java
@@ -299,17 +299,18 @@ public class AmazonAsyncEventService implements AsyncEventService {
//collect all of the
IndexOperationMessage indexOperationMessage =
indexoperationObservable
- .collect(() -> new IndexOperationMessage(), (collector, single ) -> collector.ingest(single))
+ .collect(() -> new IndexOperationMessage(), (collector, single) -> collector.ingest(single))
.toBlocking().lastOrDefault(null);
- if(indexOperationMessage == null){
- throw new IllegalArgumentException("Received null index operation.");
+ if (indexOperationMessage == null || indexOperationMessage.isEmpty()) {
+ logger.info("Received empty index sequence message:({}), body:({}) ",
+ message.getMessageId(),message.getStringBody());
}
//return type that can be indexed and ack'd later
return new IndexEventResult(Optional.fromNullable(message), Optional.fromNullable(indexOperationMessage), thisEvent.getCreationTime());
} catch (Exception e) {
- logger.error("Failed to index message: " + message.getMessageId(), e, message);
+ logger.error("Failed to index message: " + message.getMessageId(), message.getStringBody() ,e);
return new IndexEventResult(Optional.absent(), Optional.<IndexOperationMessage>absent(), event.getCreationTime());
}
});
@@ -559,7 +560,7 @@ public class AmazonAsyncEventService implements AsyncEventService {
List<IndexEventResult> indexEventResults = callEventHandlers(messages);
List<QueueMessage> messagesToAck = submitToIndex(indexEventResults);
if (messagesToAck == null || messagesToAck.size() == 0) {
- logger.error("No messages came back from the queue operation",messages);
+ logger.error("No messages came back from the queue operation should have seen "+messages.size(),messages);
return messagesToAck;
}
if(messagesToAck.size()<messages.size()){
@@ -569,7 +570,7 @@ public class AmazonAsyncEventService implements AsyncEventService {
ack(messagesToAck);
return messagesToAck;
} catch (Exception e) {
- logger.error("failed to ack messages to sqs", messages, e);
+ logger.error("failed to ack messages to sqs", e);
return null;
//do not rethrow so we can process all of them
}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/9cbe283d/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/EventBuilderImpl.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/EventBuilderImpl.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/EventBuilderImpl.java
index 4bf5695..cc0356b 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/EventBuilderImpl.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/EventBuilderImpl.java
@@ -181,8 +181,9 @@ public class EventBuilderImpl implements EventBuilder {
return true;
}
+ //entityIndexOperation.getUpdatedSince will always be 0 except for reindexing the application
//only re-index if it has been updated and been updated after our timestamp
- return modified.getValue() >= entityIndexOperation.getUpdatedSince();
+ return modified.getValue() >= entityIndexOperation.getUpdatedSince();
} )
//perform indexing on the task scheduler and start it
.flatMap( entity -> indexService.indexEntity( applicationScope, entity ) );
http://git-wip-us.apache.org/repos/asf/usergrid/blob/9cbe283d/stack/corepersistence/common/src/test/java/org/apache/usergrid/persistence/core/rx/ExceptionBehaviorTest.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/common/src/test/java/org/apache/usergrid/persistence/core/rx/ExceptionBehaviorTest.java b/stack/corepersistence/common/src/test/java/org/apache/usergrid/persistence/core/rx/ExceptionBehaviorTest.java
index 8e4f4c4..cb39ca1 100644
--- a/stack/corepersistence/common/src/test/java/org/apache/usergrid/persistence/core/rx/ExceptionBehaviorTest.java
+++ b/stack/corepersistence/common/src/test/java/org/apache/usergrid/persistence/core/rx/ExceptionBehaviorTest.java
@@ -18,12 +18,16 @@
package org.apache.usergrid.persistence.core.rx;
+import org.junit.Assert;
import org.junit.Test;
import rx.Observable;
import rx.Observer;
import rx.schedulers.Schedulers;
+import java.util.ArrayList;
+import java.util.List;
+
/**
* Tests RX exception behavior
@@ -47,6 +51,26 @@ public class ExceptionBehaviorTest {
} ).toBlocking().last();
}
+ @Test()
+ public void testSequence(){
+ ArrayList listReturn = Observable.range(0, 1).flatMap(i -> Observable.empty())
+ .collect(()->new ArrayList(),(list,i) ->{
+ list.add(i);
+ }).toBlocking().lastOrDefault(null);
+
+ Assert.assertEquals(listReturn,new ArrayList<Integer>());
+ }
+
+ @Test()
+ public void testSequence2(){
+ ArrayList listReturn = Observable.range(0, 2).buffer(2).flatMap(i -> Observable.empty())
+ .collect(()->new ArrayList(),(list,i) ->{
+ list.add(i);
+ }).toBlocking().lastOrDefault(null);
+
+ Assert.assertEquals(listReturn,new ArrayList<Integer>());
+ }
+
//
// /**
// * This shows that no re-throw happens on subscribe. This is as designed, but not as expected
http://git-wip-us.apache.org/repos/asf/usergrid/blob/9cbe283d/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/QueueMessage.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/QueueMessage.java b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/QueueMessage.java
index 0874e9c..55f79f4 100644
--- a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/QueueMessage.java
+++ b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/QueueMessage.java
@@ -22,6 +22,7 @@ public class QueueMessage {
private final String messageId;
private final String handle;
private final String type;
+ private String stringBody;
public QueueMessage(String messageId, String handle, Object body,String type) {
@@ -29,6 +30,7 @@ public class QueueMessage {
this.messageId = messageId;
this.handle = handle;
this.type = type;
+ this.stringBody = "";
}
public String getHandle() {
@@ -47,4 +49,12 @@ public class QueueMessage {
public String getType() {
return type;
}
+
+ public void setStringBody(String stringBody) {
+ this.stringBody = stringBody;
+ }
+
+ public String getStringBody() {
+ return stringBody;
+ }
}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/9cbe283d/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 bc63f53..a2b5d72 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
@@ -364,6 +364,7 @@ public class SNSQueueManagerImpl implements QueueManager {
for (Message message : messages) {
Object body;
+ final String originalBody = message.getBody();
try {
final JsonNode bodyNode = mapper.readTree(message.getBody());
@@ -375,6 +376,7 @@ public class SNSQueueManagerImpl implements QueueManager {
}
QueueMessage queueMessage = new QueueMessage(message.getMessageId(), message.getReceiptHandle(), body, message.getAttributes().get("type"));
+ queueMessage.setStringBody(originalBody);
queueMessages.add(queueMessage);
}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/9cbe283d/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/impl/SQSQueueManagerImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/impl/SQSQueueManagerImpl.java b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/impl/SQSQueueManagerImpl.java
index daa1cb5..fa9a7ac 100644
--- a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/impl/SQSQueueManagerImpl.java
+++ b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/impl/SQSQueueManagerImpl.java
@@ -196,6 +196,7 @@ public class SQSQueueManagerImpl implements QueueManager {
}
QueueMessage queueMessage = new QueueMessage(message.getMessageId(), message.getReceiptHandle(), body, message.getAttributes().get("type"));
+ queueMessage.setStringBody(message.getBody());
queueMessages.add(queueMessage);
}
[4/4] usergrid git commit: Merge branch '2.1-release' of
https://git-wip-us.apache.org/repos/asf/usergrid into
remove-inmemory-event-service
Posted by sf...@apache.org.
Merge branch '2.1-release' of https://git-wip-us.apache.org/repos/asf/usergrid into remove-inmemory-event-service
Project: http://git-wip-us.apache.org/repos/asf/usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/usergrid/commit/f6409ce3
Tree: http://git-wip-us.apache.org/repos/asf/usergrid/tree/f6409ce3
Diff: http://git-wip-us.apache.org/repos/asf/usergrid/diff/f6409ce3
Branch: refs/heads/remove-inmemory-event-service
Commit: f6409ce353e9b9be2adad8c47e3a6800f94402ae
Parents: e17dab4 9cbe283
Author: Shawn Feldman <sf...@apache.org>
Authored: Tue Oct 13 10:14:11 2015 -0600
Committer: Shawn Feldman <sf...@apache.org>
Committed: Tue Oct 13 10:14:11 2015 -0600
----------------------------------------------------------------------
.../asyncevents/AmazonAsyncEventService.java | 13 +++++++----
.../asyncevents/EventBuilderImpl.java | 3 ++-
.../core/rx/ExceptionBehaviorTest.java | 24 ++++++++++++++++++++
.../index/impl/EsIndexProducerImpl.java | 11 ++++++---
.../persistence/queue/QueueMessage.java | 10 ++++++++
.../queue/impl/SNSQueueManagerImpl.java | 2 ++
.../queue/impl/SQSQueueManagerImpl.java | 1 +
7 files changed, 56 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/usergrid/blob/f6409ce3/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/AmazonAsyncEventService.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/usergrid/blob/f6409ce3/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/EventBuilderImpl.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/usergrid/blob/f6409ce3/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsIndexProducerImpl.java
----------------------------------------------------------------------