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/06 22:34:59 UTC
[1/9] usergrid git commit: add and comment on subscribe
Repository: usergrid
Updated Branches:
refs/heads/2.1-release e0d3cd57c -> 7c5a864dd
add and comment on subscribe
Project: http://git-wip-us.apache.org/repos/asf/usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/usergrid/commit/5f20ece6
Tree: http://git-wip-us.apache.org/repos/asf/usergrid/tree/5f20ece6
Diff: http://git-wip-us.apache.org/repos/asf/usergrid/diff/5f20ece6
Branch: refs/heads/2.1-release
Commit: 5f20ece66f64305f522f13fa02d34ab2920f7089
Parents: e0d3cd5
Author: Shawn Feldman <sf...@apache.org>
Authored: Mon Oct 5 10:57:47 2015 -0600
Committer: Shawn Feldman <sf...@apache.org>
Committed: Mon Oct 5 10:57:47 2015 -0600
----------------------------------------------------------------------
.../corepersistence/CpRelationManager.java | 24 ++++++++---------
.../asyncevents/InMemoryAsyncEventService.java | 7 ++++-
.../index/ReIndexServiceImpl.java | 2 +-
.../migration/DeDupConnectionDataMigration.java | 2 +-
.../read/search/CandidateEntityFilter.java | 2 +-
.../pipeline/read/search/CandidateIdFilter.java | 2 +-
.../EsIndexMappingMigrationPlugin.java | 2 +-
.../usergrid/rest/ApplicationsResource.java | 28 +++++++++++++-------
.../management/AppInfoMigrationPlugin.java | 2 +-
.../services/AbstractConnectionsService.java | 2 +-
10 files changed, 44 insertions(+), 29 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/usergrid/blob/5f20ece6/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpRelationManager.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpRelationManager.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpRelationManager.java
index b4b39a0..aad7610 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpRelationManager.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpRelationManager.java
@@ -518,8 +518,8 @@ public class CpRelationManager implements RelationManager {
//run our delete
gm.loadEdgeVersions(
CpNamingUtils.createEdgeFromCollectionName( cpHeadEntity.getId(), collectionName, memberEntity.getId() ) )
- .flatMap( edge -> gm.markEdge( edge ) ).flatMap( edge -> gm.deleteEdge( edge ) ).toBlocking()
- .lastOrDefault( null );
+ .flatMap(edge -> gm.markEdge(edge)).flatMap(edge -> gm.deleteEdge(edge)).toBlocking()
+ .lastOrDefault(null);
/**
@@ -535,7 +535,7 @@ public class CpRelationManager implements RelationManager {
batch.deindex( indexScope, memberEntity );
- managerCache.getIndexProducer().put( batch.build()).subscribe();
+ managerCache.getIndexProducer().put( batch.build()).toBlocking().lastOrDefault(null); // this should throw an exception
// special handling for roles collection of a group
if ( headEntity.getType().equals( Group.ENTITY_TYPE ) ) {
@@ -695,7 +695,7 @@ public class CpRelationManager implements RelationManager {
//write new edge
- gm.writeEdge( edge ).subscribe();
+ gm.writeEdge(edge).toBlocking().lastOrDefault(null); //throw an exception if this fails
indexService.queueNewEdge( applicationScope, targetEntity, edge );
@@ -707,21 +707,21 @@ public class CpRelationManager implements RelationManager {
//load our versions, only retain the most recent one
- gm.loadEdgeVersions( searchByEdge ).skip( 1 ).flatMap( edgeToDelete -> {
- if ( logger.isDebugEnabled() ) {
- logger.debug( "Marking edge {} for deletion", edgeToDelete );
+ gm.loadEdgeVersions(searchByEdge).skip(1).flatMap(edgeToDelete -> {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Marking edge {} for deletion", edgeToDelete);
}
- return gm.markEdge( edgeToDelete );
- } ).lastOrDefault( null ).doOnNext( lastEdge -> {
+ return gm.markEdge(edgeToDelete );
+ }).lastOrDefault(null).doOnNext(lastEdge -> {
//no op if we hit our default
- if(lastEdge == null){
+ if (lastEdge == null) {
return;
}
//don't queue delete b/c that de-indexes, we need to delete the edges only since we have a version still existing to index.
- gm.deleteEdge( lastEdge ).subscribe();
- }).subscribe();
+ gm.deleteEdge(lastEdge).toBlocking().lastOrDefault(null); // this should throw an exception
+ }).toBlocking().lastOrDefault(null);//this should throw an exception
return connection;
http://git-wip-us.apache.org/repos/asf/usergrid/blob/5f20ece6/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/InMemoryAsyncEventService.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/InMemoryAsyncEventService.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/InMemoryAsyncEventService.java
index b29c39e..d5a0398 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/InMemoryAsyncEventService.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/InMemoryAsyncEventService.java
@@ -137,7 +137,12 @@ public class InMemoryAsyncEventService implements AsyncEventService {
return Observable.just(message);
}
});
- mapped.subscribe();
+ if(!resolveSynchronously){
+ mapped.subscribe(); //only subscribe for async
+ }else {
+ mapped.toBlocking().lastOrDefault(null);
+ }
+
}
@Override
http://git-wip-us.apache.org/repos/asf/usergrid/blob/5f20ece6/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/ReIndexServiceImpl.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/ReIndexServiceImpl.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/ReIndexServiceImpl.java
index 1353982..f108f37 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/ReIndexServiceImpl.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/ReIndexServiceImpl.java
@@ -141,7 +141,7 @@ public class ReIndexServiceImpl implements ReIndexService {
runningReIndex.collect(() -> new FlushingCollector(jobId),
((flushingCollector, edgeScopes) -> flushingCollector.flushBuffer(edgeScopes))).doOnNext( flushingCollector-> flushingCollector.complete() )
//subscribe on our I/O scheduler and run the task
- .subscribeOn( Schedulers.io() ).subscribe();
+ .subscribeOn( Schedulers.io() ).subscribe(); //want reindex to continually run so leave subscribe.
return new ReIndexStatus( jobId, Status.STARTED, 0, 0 );
http://git-wip-us.apache.org/repos/asf/usergrid/blob/5f20ece6/stack/core/src/main/java/org/apache/usergrid/corepersistence/migration/DeDupConnectionDataMigration.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/migration/DeDupConnectionDataMigration.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/migration/DeDupConnectionDataMigration.java
index bec8d6c..77deac8 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/migration/DeDupConnectionDataMigration.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/migration/DeDupConnectionDataMigration.java
@@ -72,7 +72,7 @@ public class DeDupConnectionDataMigration implements DataMigration {
} ).doOnNext( total -> {
logger.info( "Completed de-duping {} edges", total );
observer.complete();
- } ).subscribe();
+ } ).subscribe(); //want this to run through all records
return migrationVersion;
http://git-wip-us.apache.org/repos/asf/usergrid/blob/5f20ece6/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/search/CandidateEntityFilter.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/search/CandidateEntityFilter.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/search/CandidateEntityFilter.java
index ceb18ae..9a9636f 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/search/CandidateEntityFilter.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/search/CandidateEntityFilter.java
@@ -178,7 +178,7 @@ public class CandidateEntityFilter extends AbstractFilter<FilterResult<Candidate
validate( candidateResult );
}
- indexProducer.put(batch.build()).subscribe();
+ indexProducer.put(batch.build()).toBlocking().lastOrDefault(null); // want to rethrow if batch fails
}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/5f20ece6/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/search/CandidateIdFilter.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/search/CandidateIdFilter.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/search/CandidateIdFilter.java
index b2fd675..a627b4f 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/search/CandidateIdFilter.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/search/CandidateIdFilter.java
@@ -143,7 +143,7 @@ public class CandidateIdFilter extends AbstractFilter<FilterResult<Candidate>, F
validate( candidateResult );
}
- indexProducer.put( batch.build()).subscribe();
+ indexProducer.put( batch.build()).toBlocking().lastOrDefault(null);//want to rethrow if batch fails
}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/5f20ece6/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/migration/EsIndexMappingMigrationPlugin.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/migration/EsIndexMappingMigrationPlugin.java b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/migration/EsIndexMappingMigrationPlugin.java
index 5c2e9c1..de32e0e 100644
--- a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/migration/EsIndexMappingMigrationPlugin.java
+++ b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/migration/EsIndexMappingMigrationPlugin.java
@@ -94,7 +94,7 @@ public class EsIndexMappingMigrationPlugin implements MigrationPlugin {
migrationInfoSerialization.setVersion(getName(), getMaxVersion());
observer.complete();
})
- .subscribe();
+ .subscribe(); //should run through
}catch (Exception ee){
http://git-wip-us.apache.org/repos/asf/usergrid/blob/5f20ece6/stack/rest/src/main/java/org/apache/usergrid/rest/ApplicationsResource.java
----------------------------------------------------------------------
diff --git a/stack/rest/src/main/java/org/apache/usergrid/rest/ApplicationsResource.java b/stack/rest/src/main/java/org/apache/usergrid/rest/ApplicationsResource.java
index 6127efc..55cf0f6 100644
--- a/stack/rest/src/main/java/org/apache/usergrid/rest/ApplicationsResource.java
+++ b/stack/rest/src/main/java/org/apache/usergrid/rest/ApplicationsResource.java
@@ -105,21 +105,27 @@ public class ApplicationsResource extends AbstractContextResource {
Map<String,Object> map = new LinkedHashMap<>();
map.put("count",itemsDeleted.intValue());
final StatusService statusService = injector.getInstance(StatusService.class);
- statusService.setStatus(applicationId, jobId, StatusService.Status.INPROGRESS,map).subscribe();
+ statusService.setStatus(applicationId, jobId, StatusService.Status.INPROGRESS,map)
+ .subscribe();//do not want to throw this exception
}
})
- .doOnCompleted(() ->{
- Map<String,Object> map = new LinkedHashMap<>();
- map.put("count",itemsDeleted.intValue());
+ .doOnCompleted(() -> {
+ Map<String, Object> map = new LinkedHashMap<>();
+ map.put("count", itemsDeleted.intValue());
final StatusService statusService = injector.getInstance(StatusService.class);
- statusService.setStatus(applicationId,jobId, StatusService.Status.COMPLETE,map).subscribe();
+ statusService.setStatus(applicationId, jobId, StatusService.Status.COMPLETE, map)
+ .toBlocking().lastOrDefault(null);//want to rethrow this exception
})
- .subscribe();
+ .toBlocking().lastOrDefault(null);//expecting exception to be caught if job fails
} catch ( Exception e ) {
Map<String,Object> map = new LinkedHashMap<>();
map.put("exception",e);
- statusService.setStatus(applicationId,jobId, StatusService.Status.FAILED,map).subscribe();
+ try {
+ statusService.setStatus(applicationId, jobId, StatusService.Status.FAILED, map).toBlocking().lastOrDefault(null);//leave as subscribe if fails retry
+ }catch (Exception subE){
+ logger.error("failed to update status "+jobId,subE);
+ }
logger.error( "Failed to delete appid:"+applicationId + " jobid:"+jobId+" count:"+itemsDeleted, e );
}
}
@@ -129,8 +135,12 @@ public class ApplicationsResource extends AbstractContextResource {
delete.setDaemon(true);
delete.start();
- statusService.setStatus(applicationId,jobId, StatusService.Status.STARTED,new LinkedHashMap<>()).subscribe();
-
+ try {
+ //should throw exception if can't start
+ statusService.setStatus(applicationId, jobId, StatusService.Status.STARTED, new LinkedHashMap<>()).toBlocking().lastOrDefault(null);
+ }catch (Exception e){
+ logger.error("failed to set status for " + jobId, e);
+ }
Map<String,Object> data = new HashMap<>();
data.put("jobId",jobId);
data.put("status",StatusService.Status.STARTED);
http://git-wip-us.apache.org/repos/asf/usergrid/blob/5f20ece6/stack/services/src/main/java/org/apache/usergrid/management/AppInfoMigrationPlugin.java
----------------------------------------------------------------------
diff --git a/stack/services/src/main/java/org/apache/usergrid/management/AppInfoMigrationPlugin.java b/stack/services/src/main/java/org/apache/usergrid/management/AppInfoMigrationPlugin.java
index b78b646..d9d3d0d 100644
--- a/stack/services/src/main/java/org/apache/usergrid/management/AppInfoMigrationPlugin.java
+++ b/stack/services/src/main/java/org/apache/usergrid/management/AppInfoMigrationPlugin.java
@@ -165,7 +165,7 @@ public class AppInfoMigrationPlugin implements MigrationPlugin {
.doOnCompleted( () -> {
migrationInfoSerialization.setVersion( getName(), getMaxVersion() );
observer.complete();
- } ).subscribe();
+ } ).subscribe();//let this run through since it handles errors
}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/5f20ece6/stack/services/src/main/java/org/apache/usergrid/services/AbstractConnectionsService.java
----------------------------------------------------------------------
diff --git a/stack/services/src/main/java/org/apache/usergrid/services/AbstractConnectionsService.java b/stack/services/src/main/java/org/apache/usergrid/services/AbstractConnectionsService.java
index ee322d2..cd56d1d 100644
--- a/stack/services/src/main/java/org/apache/usergrid/services/AbstractConnectionsService.java
+++ b/stack/services/src/main/java/org/apache/usergrid/services/AbstractConnectionsService.java
@@ -459,7 +459,7 @@ public class AbstractConnectionsService extends AbstractService {
throw new RuntimeException( "Unable to save connection", e );
}
}).subscribeOn( Schedulers.io() );
- }, 10).subscribe();
+ }, 10).toBlocking().lastOrDefault(null); //needs to rethrow
}
[2/9] usergrid git commit: filter some ops out,
check for queue overflow
Posted by sf...@apache.org.
filter some ops out, check for queue overflow
Project: http://git-wip-us.apache.org/repos/asf/usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/usergrid/commit/f35d01c7
Tree: http://git-wip-us.apache.org/repos/asf/usergrid/tree/f35d01c7
Diff: http://git-wip-us.apache.org/repos/asf/usergrid/diff/f35d01c7
Branch: refs/heads/2.1-release
Commit: f35d01c730998d45ca6feb5edf9138f04d185951
Parents: 5f20ece
Author: Shawn Feldman <sf...@apache.org>
Authored: Mon Oct 5 13:59:10 2015 -0600
Committer: Shawn Feldman <sf...@apache.org>
Committed: Mon Oct 5 13:59:10 2015 -0600
----------------------------------------------------------------------
.../asyncevents/AmazonAsyncEventService.java | 31 ++++++++++++--------
.../usergrid/persistence/index/IndexFig.java | 3 ++
.../index/impl/EsIndexProducerImpl.java | 7 +++++
.../queue/impl/QueueManagerFactoryImpl.java | 2 +-
4 files changed, 29 insertions(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/usergrid/blob/f35d01c7/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 92faed4..e215d48 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
@@ -239,7 +239,7 @@ public class AmazonAsyncEventService implements AsyncEventService {
if (event == null) {
logger.error("AsyncEvent type or event is null!");
- return Observable.just(new IndexEventResult(message, Optional.<IndexOperationMessage>absent(), false));
+ return Observable.just(new IndexEventResult(message, Optional.<IndexOperationMessage>absent(), true));
}
try {
//merge each operation to a master observable;
@@ -254,10 +254,10 @@ public class AmazonAsyncEventService implements AsyncEventService {
} else if (event instanceof InitializeApplicationIndexEvent) {
//does not return observable
handleInitializeApplicationIndex(message);
- return Observable.just(new IndexEventResult(message, Optional.<IndexOperationMessage>absent(), true));
+ return Observable.just(new IndexEventResult(message, Optional.<IndexOperationMessage>absent(), false));
} else {
logger.error("Unknown EventType: {}", event);
- return Observable.just(new IndexEventResult(message, Optional.<IndexOperationMessage>absent(), false));
+ return Observable.just(new IndexEventResult(message, Optional.<IndexOperationMessage>absent(), true));
}
}catch (Exception e){
logger.error("Failed to index entity", e,message);
@@ -270,20 +270,25 @@ public class AmazonAsyncEventService implements AsyncEventService {
return masterObservable
//remove unsuccessful
- .filter( indexEventResult -> indexEventResult.success() && indexEventResult.getIndexOperationMessage()
- .isPresent() )
+ .filter( indexEventResult -> indexEventResult.shouldProcess() )
//take the max
.buffer( MAX_TAKE )
//map them to index results and return them
.flatMap( indexEventResults -> {
IndexOperationMessage combined = new IndexOperationMessage();
indexEventResults.stream().forEach(
- indexEventResult -> combined.ingest( indexEventResult.getIndexOperationMessage().get() ) );
+ indexEventResult ->{
+ if(indexEventResult.getIndexOperationMessage().isPresent()) {
+ combined.ingest(indexEventResult.getIndexOperationMessage().get());
+ }
+ } );
+
//ack after successful completion of the operation.
- return indexProducer.put( combined ).flatMap( operationResult -> Observable.from( indexEventResults ) )
+ return indexProducer.put( combined )
+ .flatMap( operationResult -> Observable.from( indexEventResults ) )
//ack each message, but only if we didn't error. If we did, we'll want to log it and
- .map( indexEventResult -> {
+ .map( indexEventResult -> {
ack( indexEventResult.queueMessage );
return indexEventResult;
} );
@@ -562,21 +567,21 @@ public class AmazonAsyncEventService implements AsyncEventService {
public class IndexEventResult{
private final QueueMessage queueMessage;
private final Optional<IndexOperationMessage> indexOperationMessage;
- private final boolean success;
+ private final boolean shouldProcess;
- public IndexEventResult(QueueMessage queueMessage, Optional<IndexOperationMessage> indexOperationMessage ,boolean success){
+ public IndexEventResult(QueueMessage queueMessage, Optional<IndexOperationMessage> indexOperationMessage ,boolean shouldProcess){
this.queueMessage = queueMessage;
this.indexOperationMessage = indexOperationMessage;
- this.success = success;
+ this.shouldProcess = shouldProcess;
}
public QueueMessage getQueueMessage() {
return queueMessage;
}
- public boolean success() {
- return success;
+ public boolean shouldProcess() {
+ return shouldProcess;
}
public Optional<IndexOperationMessage> getIndexOperationMessage() {
http://git-wip-us.apache.org/repos/asf/usergrid/blob/f35d01c7/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/IndexFig.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/IndexFig.java b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/IndexFig.java
index 4f35730..e093d7d 100644
--- a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/IndexFig.java
+++ b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/IndexFig.java
@@ -196,4 +196,7 @@ public interface IndexFig extends GuicyFig {
long getWriteTimeout();
+ @Default("1000")
+ @Key( "elasticsearch_queue_error_sleep_ms" )
+ long getSleepTimeForQueueError();
}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/f35d01c7/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 869b75a..828027c 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
@@ -205,6 +205,13 @@ public class EsIndexProducerImpl implements IndexProducer {
}
if ( error ) {
+ if(errorString.lastIndexOf("rejected execution (queue capacity")>=0){
+ try{
+ Thread.sleep(indexFig.getSleepTimeForQueueError());
+ }catch (InterruptedException ie){
+ //move on
+ }
+ }
throw new RuntimeException(
"Error during processing of bulk index operations one of the responses failed. \n" + errorString);
}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/f35d01c7/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/impl/QueueManagerFactoryImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/impl/QueueManagerFactoryImpl.java b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/impl/QueueManagerFactoryImpl.java
index d0ed1ef..a1940d0 100644
--- a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/impl/QueueManagerFactoryImpl.java
+++ b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/impl/QueueManagerFactoryImpl.java
@@ -47,7 +47,7 @@ public class QueueManagerFactoryImpl implements QueueManagerFactory {
}
@Override
public QueueManager getQueueManager(QueueScope scope) {
- if(queueFig.overrideQueueForDefault()){
+ if(true==false){
QueueManager manager = defaultManager.get(scope.getName());
if(manager==null){
manager = new DefaultQueueManager();
[3/9] usergrid git commit: remove condition
Posted by sf...@apache.org.
remove condition
Project: http://git-wip-us.apache.org/repos/asf/usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/usergrid/commit/b437f618
Tree: http://git-wip-us.apache.org/repos/asf/usergrid/tree/b437f618
Diff: http://git-wip-us.apache.org/repos/asf/usergrid/diff/b437f618
Branch: refs/heads/2.1-release
Commit: b437f618775180a7b4a4a8e174ea7f152c4aa4fc
Parents: f35d01c
Author: Shawn Feldman <sf...@apache.org>
Authored: Mon Oct 5 14:24:57 2015 -0600
Committer: Shawn Feldman <sf...@apache.org>
Committed: Mon Oct 5 14:24:57 2015 -0600
----------------------------------------------------------------------
.../usergrid/persistence/queue/impl/QueueManagerFactoryImpl.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/usergrid/blob/b437f618/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/impl/QueueManagerFactoryImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/impl/QueueManagerFactoryImpl.java b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/impl/QueueManagerFactoryImpl.java
index a1940d0..d0ed1ef 100644
--- a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/impl/QueueManagerFactoryImpl.java
+++ b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/impl/QueueManagerFactoryImpl.java
@@ -47,7 +47,7 @@ public class QueueManagerFactoryImpl implements QueueManagerFactory {
}
@Override
public QueueManager getQueueManager(QueueScope scope) {
- if(true==false){
+ if(queueFig.overrideQueueForDefault()){
QueueManager manager = defaultManager.get(scope.getName());
if(manager==null){
manager = new DefaultQueueManager();
[9/9] usergrid git commit: commit stash
Posted by sf...@apache.org.
commit stash
Project: http://git-wip-us.apache.org/repos/asf/usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/usergrid/commit/7c5a864d
Tree: http://git-wip-us.apache.org/repos/asf/usergrid/tree/7c5a864d
Diff: http://git-wip-us.apache.org/repos/asf/usergrid/diff/7c5a864d
Branch: refs/heads/2.1-release
Commit: 7c5a864dd1ce22ada5456ced1a6c10f9a5533b1f
Parents: 7dceb56
Author: Shawn Feldman <sf...@apache.org>
Authored: Tue Oct 6 14:34:33 2015 -0600
Committer: Shawn Feldman <sf...@apache.org>
Committed: Tue Oct 6 14:34:33 2015 -0600
----------------------------------------------------------------------
.../asyncevents/AmazonAsyncEventService.java | 71 ++++++++++++++------
.../index/impl/EsIndexProducerImpl.java | 6 +-
2 files changed, 53 insertions(+), 24 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/usergrid/blob/7c5a864d/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 14d37b5..1bc70cd 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
@@ -23,6 +23,7 @@ package org.apache.usergrid.corepersistence.asyncevents;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import com.google.common.base.Optional;
@@ -226,6 +227,28 @@ public class AmazonAsyncEventService implements AsyncEventService {
}
+ /**
+ * Ack message in SQS
+ */
+ public void ack(final List<QueueMessage> messages) {
+
+ final Timer.Context timer = this.ackTimer.time();
+
+ try{
+ queue.commitMessages(messages);
+
+ //decrement our in-flight counter
+ inFlight.decrementAndGet();
+
+ }catch(Exception e){
+ throw new RuntimeException("Unable to ack messages", e);
+ }finally {
+ timer.stop();
+ }
+
+
+ }
+
private Observable<QueueMessage> handleMessages( final List<QueueMessage> messages ) {
if (logger.isDebugEnabled()) {
@@ -243,7 +266,7 @@ public class AmazonAsyncEventService implements AsyncEventService {
if (event == null) {
logger.error("AsyncEvent type or event is null!");
- return Observable.just(new IndexEventResult(message, Optional.<IndexOperationMessage>absent(), true));
+ return Observable.just(new IndexEventResult(Optional.fromNullable(message), Optional.<IndexOperationMessage>absent()));
}
try {
//merge each operation to a master observable;
@@ -258,23 +281,24 @@ public class AmazonAsyncEventService implements AsyncEventService {
} else if (event instanceof InitializeApplicationIndexEvent) {
//does not return observable
handleInitializeApplicationIndex(message);
- return Observable.just(new IndexEventResult(message, Optional.<IndexOperationMessage>absent(), false));
+ return Observable.just(new IndexEventResult(Optional.absent(), Optional.<IndexOperationMessage>absent()));
} else {
- logger.error("Unknown EventType: {}", event);
- return Observable.just(new IndexEventResult(message, Optional.<IndexOperationMessage>absent(), true));
+ logger.error("Unknown EventType: {}", event);//TODO: print json instead
+ return Observable.just(new IndexEventResult(Optional.fromNullable(message), Optional.<IndexOperationMessage>absent()));
}
} catch (Exception e) {
logger.error("Failed to index entity", e, message);
- return Observable.just(new IndexEventResult(message, Optional.<IndexOperationMessage>absent(), false));
+ return Observable.just(new IndexEventResult(Optional.absent(), Optional.<IndexOperationMessage>absent()));
} finally {
messageCycle.update(System.currentTimeMillis() - event.getCreationTime());
}
});
+ //filter for success, send to the index(optional), ack
return masterObservable
//remove unsuccessful
- .filter(indexEventResult -> indexEventResult.shouldProcess())
+ .filter(indexEventResult -> indexEventResult.getQueueMessage().isPresent())
//take the max
.buffer( MAX_TAKE )
//map them to index results and return them
@@ -290,8 +314,8 @@ public class AmazonAsyncEventService implements AsyncEventService {
//ack after successful completion of the operation.
return indexProducer.put(combined)
- .flatMap(operationResult -> Observable.from(indexEventResults))
- .map(result -> result.getQueueMessage());
+ .flatMap(indexOperationMessage -> Observable.from(indexEventResults))
+ .map(result -> result.getQueueMessage().get());
});
@@ -303,10 +327,10 @@ public class AmazonAsyncEventService implements AsyncEventService {
){
try{
return operation.call(queueMessage)
- .map(indexOperationMessage -> new IndexEventResult(queueMessage, Optional.fromNullable(indexOperationMessage), true));
+ .map(indexOperationMessage -> new IndexEventResult(Optional.fromNullable(queueMessage), Optional.fromNullable(indexOperationMessage)));
}catch (Exception e){
logger.error("failed to run index",e);
- return Observable.just( new IndexEventResult(queueMessage, Optional.<IndexOperationMessage>absent(),false));
+ return Observable.just( new IndexEventResult(Optional.fromNullable(queueMessage), Optional.<IndexOperationMessage>absent()));
}
}
@@ -543,10 +567,17 @@ public class AmazonAsyncEventService implements AsyncEventService {
//this won't block our read loop, just reads and proceeds
.flatMap(messages ->
handleMessages(messages)
- .doOnNext(message -> {
- //ack each message, but only if we didn't error.
- ack(message);
+ .buffer(MAX_TAKE)
+ .doOnNext(messagesToAck -> {
+ try {
+ //ack each message, but only if we didn't error.
+ ack(messagesToAck);
+ } catch (Exception e) {
+ logger.error("failed to ack messages to sqs", messagesToAck.get(0).getMessageId(), e);
+ //do not rethrow so we can process all of them
+ }
})
+ .flatMap(messagesToAck -> Observable.from(messagesToAck))
);
//start in the background
@@ -574,23 +605,19 @@ public class AmazonAsyncEventService implements AsyncEventService {
public class IndexEventResult{
- private final QueueMessage queueMessage;
+ private final Optional<QueueMessage> queueMessage;
private final Optional<IndexOperationMessage> indexOperationMessage;
- private final boolean shouldProcess;
- public IndexEventResult(QueueMessage queueMessage, Optional<IndexOperationMessage> indexOperationMessage ,boolean shouldProcess){
+
+ public IndexEventResult(Optional<QueueMessage> queueMessage, Optional<IndexOperationMessage> indexOperationMessage ){
this.queueMessage = queueMessage;
this.indexOperationMessage = indexOperationMessage;
- this.shouldProcess = shouldProcess;
- }
- public QueueMessage getQueueMessage() {
- return queueMessage;
}
- public boolean shouldProcess() {
- return shouldProcess;
+ public Optional<QueueMessage> getQueueMessage() {
+ return queueMessage;
}
public Optional<IndexOperationMessage> getIndexOperationMessage() {
http://git-wip-us.apache.org/repos/asf/usergrid/blob/7c5a864d/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 409c2bc..e778948 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
@@ -115,6 +115,7 @@ public class EsIndexProducerImpl implements IndexProducer {
final Observable<IndexOperation> index = Observable.from(batch.getIndexRequests());
final Observable<DeIndexOperation> deIndex = Observable.from(batch.getDeIndexRequests());
+ //TODO: look at indexing ordering
final Observable<BatchOperation> batchOps = Observable.merge(index, deIndex);
//buffer into the max size we can send ES and fire them all off until we're completed
@@ -207,9 +208,10 @@ public class EsIndexProducerImpl implements IndexProducer {
if ( error ) {
if(errorString.lastIndexOf("rejected execution (queue capacity")>=0){
try{
- log.warn("Encountered Queue Capacity Exception from ElasticSearch slowing by " +indexFig.getSleepTimeForQueueError());
+ log.warn("Encountered Queue Capacity Exception from ElasticSearch slowing by "
+ + indexFig.getSleepTimeForQueueError() );
Thread.sleep(indexFig.getSleepTimeForQueueError());
- }catch (InterruptedException ie){
+ }catch (Exception e){
//move on
}
}
[4/9] usergrid git commit: fix subscribe for messages
Posted by sf...@apache.org.
fix subscribe for messages
Project: http://git-wip-us.apache.org/repos/asf/usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/usergrid/commit/1a1d42e1
Tree: http://git-wip-us.apache.org/repos/asf/usergrid/tree/1a1d42e1
Diff: http://git-wip-us.apache.org/repos/asf/usergrid/diff/1a1d42e1
Branch: refs/heads/2.1-release
Commit: 1a1d42e1f53cabf433442c17f614f9fcae418a22
Parents: b437f61
Author: Shawn Feldman <sf...@apache.org>
Authored: Mon Oct 5 16:28:04 2015 -0600
Committer: Shawn Feldman <sf...@apache.org>
Committed: Mon Oct 5 16:28:04 2015 -0600
----------------------------------------------------------------------
.../asyncevents/AmazonAsyncEventService.java | 20 ++++++++++++--------
1 file changed, 12 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/usergrid/blob/1a1d42e1/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 e215d48..bf29c5a 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
@@ -285,13 +285,9 @@ public class AmazonAsyncEventService implements AsyncEventService {
//ack after successful completion of the operation.
- return indexProducer.put( combined )
- .flatMap( operationResult -> Observable.from( indexEventResults ) )
- //ack each message, but only if we didn't error. If we did, we'll want to log it and
- .map( indexEventResult -> {
- ack( indexEventResult.queueMessage );
- return indexEventResult;
- } );
+ return indexProducer.put(combined)
+ .flatMap(operationResult -> Observable.from(indexEventResults));
+
} );
}
@@ -538,7 +534,15 @@ public class AmazonAsyncEventService implements AsyncEventService {
}
})
//this won't block our read loop, just reads and proceeds
- .flatMap( messages -> handleMessages( messages ) ).subscribeOn( Schedulers.newThread() );
+ .map(messages ->
+ handleMessages(messages)
+ .map(indexEventResult -> {
+ ack( indexEventResult.getQueueMessage() );
+ return indexEventResult;
+ })
+ .toBlocking().lastOrDefault(null)
+ )//ack each message, but only if we didn't error. If we did, we'll want to log it and
+ .subscribeOn( Schedulers.newThread() );
//start in the background
[7/9] usergrid git commit: refactor observable
Posted by sf...@apache.org.
refactor observable
Project: http://git-wip-us.apache.org/repos/asf/usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/usergrid/commit/17586ecb
Tree: http://git-wip-us.apache.org/repos/asf/usergrid/tree/17586ecb
Diff: http://git-wip-us.apache.org/repos/asf/usergrid/diff/17586ecb
Branch: refs/heads/2.1-release
Commit: 17586ecb5dd38ee71b7edfd256fdecb50e615556
Parents: 890611d
Author: Shawn Feldman <sf...@apache.org>
Authored: Tue Oct 6 08:26:46 2015 -0600
Committer: Shawn Feldman <sf...@apache.org>
Committed: Tue Oct 6 08:26:46 2015 -0600
----------------------------------------------------------------------
.../asyncevents/AmazonAsyncEventService.java | 20 ++++++++++----------
1 file changed, 10 insertions(+), 10 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/usergrid/blob/17586ecb/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 37a9da5..e16de05 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
@@ -227,7 +227,7 @@ public class AmazonAsyncEventService implements AsyncEventService {
}
- private Observable<IndexEventResult> handleMessages( final List<QueueMessage> messages ) {
+ private Observable<QueueMessage> handleMessages( final List<QueueMessage> messages ) {
if (logger.isDebugEnabled()) {
logger.debug("handleMessages with {} message", messages.size());
}
@@ -290,7 +290,8 @@ public class AmazonAsyncEventService implements AsyncEventService {
//ack after successful completion of the operation.
return indexProducer.put(combined)
- .flatMap(operationResult -> Observable.from(indexEventResults));
+ .flatMap(operationResult -> Observable.from(indexEventResults))
+ .map(result -> result.getQueueMessage());
});
@@ -448,7 +449,7 @@ public class AmazonAsyncEventService implements AsyncEventService {
final Observable<IndexOperationMessage> merged = entityDeleteResults
.getEntitiesCompacted()
- .collect(() -> new ArrayList<>(),(list,item)-> list.add(item))
+ .collect(() -> new ArrayList<>(), (list, item) -> list.add(item))
.flatMap(collected -> entityDeleteResults.getIndexObservable()) ;
return merged;
}
@@ -499,7 +500,7 @@ public class AmazonAsyncEventService implements AsyncEventService {
private void startWorker() {
synchronized (mutex) {
- Observable<IndexEventResult> consumer =
+ Observable<QueueMessage> consumer =
Observable.create(new Observable.OnSubscribe<List<QueueMessage>>() {
@Override
public void call(final Subscriber<? super List<QueueMessage>> subscriber) {
@@ -540,14 +541,13 @@ public class AmazonAsyncEventService implements AsyncEventService {
}
})
//this won't block our read loop, just reads and proceeds
- .map(messages ->
+ .flatMap(messages ->
handleMessages(messages)
- .map(indexEventResult -> {
- ack(indexEventResult.getQueueMessage());
- return indexEventResult;
+ .doOnNext(message -> {
+ //ack each message, but only if we didn't error.
+ ack(message);
})
- .toBlocking().lastOrDefault(null)
- )//ack each message, but only if we didn't error. If we did, we'll want to log it and
+ )
.subscribeOn(Schedulers.newThread());
//start in the background
[8/9] usergrid git commit: move subscriber
Posted by sf...@apache.org.
move subscriber
Project: http://git-wip-us.apache.org/repos/asf/usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/usergrid/commit/7dceb563
Tree: http://git-wip-us.apache.org/repos/asf/usergrid/tree/7dceb563
Diff: http://git-wip-us.apache.org/repos/asf/usergrid/diff/7dceb563
Branch: refs/heads/2.1-release
Commit: 7dceb563eeb836a7dd09280b07af78c69ebb1793
Parents: 17586ec
Author: Shawn Feldman <sf...@apache.org>
Authored: Tue Oct 6 08:39:21 2015 -0600
Committer: Shawn Feldman <sf...@apache.org>
Committed: Tue Oct 6 08:39:21 2015 -0600
----------------------------------------------------------------------
.../asyncevents/AmazonAsyncEventService.java | 5 ++-
.../core/rx/ExceptionBehaviorTest.java | 32 +++++++++++++++++++-
2 files changed, 33 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/usergrid/blob/7dceb563/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 e16de05..14d37b5 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
@@ -547,12 +547,11 @@ public class AmazonAsyncEventService implements AsyncEventService {
//ack each message, but only if we didn't error.
ack(message);
})
- )
- .subscribeOn(Schedulers.newThread());
+ );
//start in the background
- final Subscription subscription = consumer.subscribe();
+ final Subscription subscription = consumer.subscribeOn(Schedulers.newThread()).subscribe();
subscriptions.add(subscription);
}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/7dceb563/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 be10d0a..8e4f4c4 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
@@ -22,6 +22,7 @@ import org.junit.Test;
import rx.Observable;
import rx.Observer;
+import rx.schedulers.Schedulers;
/**
@@ -31,13 +32,21 @@ public class ExceptionBehaviorTest {
//this test shows toBlocking re-throws exceptions correctly
@Test( expected = TestException.class )
- public void throwOnBlocking() {
+ public void throwOnBlockingFirst() {
Observable.range( 0, 1 ).map( integer -> {
throw new TestException( "I throw and exception" );
} ).toBlocking().first();
}
+ @Test( expected = TestException.class )
+ public void throwOnBlockingLast() {
+
+ Observable.range( 0, 1 ).map( integer -> {
+ throw new TestException( "I throw and exception" );
+ } ).toBlocking().last();
+ }
+
//
// /**
// * This shows that no re-throw happens on subscribe. This is as designed, but not as expected
@@ -67,6 +76,27 @@ public class ExceptionBehaviorTest {
exceptionObserver.checkResult();
}
+ /**
+ * Tests working with observers
+ */
+ @Test( expected = TestException.class )
+ public void throwOnSubscribeObservableNewThread() throws Exception {
+
+ final ReThrowObserver exceptionObserver = new ReThrowObserver();
+
+ Observable.range( 0, 1 ).map(integer -> {
+ throw new TestException("I throw and exception");
+ })
+ .doOnError(t -> exceptionObserver.onError(t))
+ .subscribeOn(Schedulers.newThread())
+ .subscribe(exceptionObserver);
+
+ for(int i =0; i<5; i++) {
+ exceptionObserver.checkResult();
+ Thread.sleep(200);
+ }
+ }
+
private static final class TestException extends RuntimeException {
[6/9] usergrid git commit: fix observable class cast exception
Posted by sf...@apache.org.
fix observable class cast exception
Project: http://git-wip-us.apache.org/repos/asf/usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/usergrid/commit/890611de
Tree: http://git-wip-us.apache.org/repos/asf/usergrid/tree/890611de
Diff: http://git-wip-us.apache.org/repos/asf/usergrid/diff/890611de
Branch: refs/heads/2.1-release
Commit: 890611de99bfce5ee02753168b101549d89d5205
Parents: d6bf2fa
Author: Shawn Feldman <sf...@apache.org>
Authored: Mon Oct 5 18:46:43 2015 -0600
Committer: Shawn Feldman <sf...@apache.org>
Committed: Mon Oct 5 18:46:43 2015 -0600
----------------------------------------------------------------------
.../corepersistence/asyncevents/AmazonAsyncEventService.java | 5 ++++-
.../usergrid/persistence/index/impl/EsIndexProducerImpl.java | 1 +
2 files changed, 5 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/usergrid/blob/890611de/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 abb76c2..37a9da5 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
@@ -446,7 +446,10 @@ public class AmazonAsyncEventService implements AsyncEventService {
entityDeleteResults = eventBuilder.buildEntityDelete( applicationScope, entityId );
- final Observable<IndexOperationMessage> merged = entityDeleteResults.getEntitiesCompacted().flatMap(mvccLogEntries -> entityDeleteResults.getIndexObservable()) ;
+ final Observable<IndexOperationMessage> merged = entityDeleteResults
+ .getEntitiesCompacted()
+ .collect(() -> new ArrayList<>(),(list,item)-> list.add(item))
+ .flatMap(collected -> entityDeleteResults.getIndexObservable()) ;
return merged;
}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/890611de/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 828027c..409c2bc 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
@@ -207,6 +207,7 @@ public class EsIndexProducerImpl implements IndexProducer {
if ( error ) {
if(errorString.lastIndexOf("rejected execution (queue capacity")>=0){
try{
+ log.warn("Encountered Queue Capacity Exception from ElasticSearch slowing by " +indexFig.getSleepTimeForQueueError());
Thread.sleep(indexFig.getSleepTimeForQueueError());
}catch (InterruptedException ie){
//move on
[5/9] usergrid git commit: fix observable class cast exception
Posted by sf...@apache.org.
fix observable class cast exception
Project: http://git-wip-us.apache.org/repos/asf/usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/usergrid/commit/d6bf2fac
Tree: http://git-wip-us.apache.org/repos/asf/usergrid/tree/d6bf2fac
Diff: http://git-wip-us.apache.org/repos/asf/usergrid/diff/d6bf2fac
Branch: refs/heads/2.1-release
Commit: d6bf2fac1b5b7ee994d13c7efce007d1f0dc6de2
Parents: 1a1d42e
Author: Shawn Feldman <sf...@apache.org>
Authored: Mon Oct 5 18:36:05 2015 -0600
Committer: Shawn Feldman <sf...@apache.org>
Committed: Mon Oct 5 18:36:05 2015 -0600
----------------------------------------------------------------------
.../asyncevents/AmazonAsyncEventService.java | 37 +++++++++++---------
1 file changed, 20 insertions(+), 17 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/usergrid/blob/d6bf2fac/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 bf29c5a..abb76c2 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
@@ -233,8 +233,12 @@ public class AmazonAsyncEventService implements AsyncEventService {
}
Observable<IndexEventResult> masterObservable = Observable.from(messages).flatMap(message -> {
- final AsyncEvent event = (AsyncEvent) message.getBody();
-
+ AsyncEvent event = null;
+ try{
+ event = (AsyncEvent) message.getBody();
+ }catch (ClassCastException cce){
+ logger.error("Failed to deserialize message body",cce);
+ }
logger.debug("Processing {} event", event);
if (event == null) {
@@ -259,10 +263,10 @@ public class AmazonAsyncEventService implements AsyncEventService {
logger.error("Unknown EventType: {}", event);
return Observable.just(new IndexEventResult(message, Optional.<IndexOperationMessage>absent(), true));
}
- }catch (Exception e){
- logger.error("Failed to index entity", e,message);
+ } catch (Exception e) {
+ logger.error("Failed to index entity", e, message);
return Observable.just(new IndexEventResult(message, Optional.<IndexOperationMessage>absent(), false));
- }finally {
+ } finally {
messageCycle.update(System.currentTimeMillis() - event.getCreationTime());
}
@@ -270,25 +274,25 @@ public class AmazonAsyncEventService implements AsyncEventService {
return masterObservable
//remove unsuccessful
- .filter( indexEventResult -> indexEventResult.shouldProcess() )
+ .filter(indexEventResult -> indexEventResult.shouldProcess())
//take the max
.buffer( MAX_TAKE )
//map them to index results and return them
- .flatMap( indexEventResults -> {
+ .flatMap(indexEventResults -> {
IndexOperationMessage combined = new IndexOperationMessage();
indexEventResults.stream().forEach(
- indexEventResult ->{
- if(indexEventResult.getIndexOperationMessage().isPresent()) {
+ indexEventResult -> {
+ if (indexEventResult.getIndexOperationMessage().isPresent()) {
combined.ingest(indexEventResult.getIndexOperationMessage().get());
}
- } );
+ });
//ack after successful completion of the operation.
return indexProducer.put(combined)
.flatMap(operationResult -> Observable.from(indexEventResults));
- } );
+ });
}
@@ -376,8 +380,8 @@ public class AmazonAsyncEventService implements AsyncEventService {
final EntityCollectionManager ecm = entityCollectionManagerFactory.createCollectionManager( applicationScope );
- final Observable<IndexOperationMessage> edgeIndexObservable = ecm.load(edgeIndexEvent.getEntityId()).flatMap( entity -> eventBuilder.buildNewEdge(
- applicationScope, entity, edge ) );
+ final Observable<IndexOperationMessage> edgeIndexObservable = ecm.load(edgeIndexEvent.getEntityId()).flatMap(entity -> eventBuilder.buildNewEdge(
+ applicationScope, entity, edge));
return edgeIndexObservable;
}
@@ -442,8 +446,7 @@ public class AmazonAsyncEventService implements AsyncEventService {
entityDeleteResults = eventBuilder.buildEntityDelete( applicationScope, entityId );
- final Observable merged = Observable.merge( entityDeleteResults.getEntitiesCompacted(),
- entityDeleteResults.getIndexObservable() );
+ final Observable<IndexOperationMessage> merged = entityDeleteResults.getEntitiesCompacted().flatMap(mvccLogEntries -> entityDeleteResults.getIndexObservable()) ;
return merged;
}
@@ -537,12 +540,12 @@ public class AmazonAsyncEventService implements AsyncEventService {
.map(messages ->
handleMessages(messages)
.map(indexEventResult -> {
- ack( indexEventResult.getQueueMessage() );
+ ack(indexEventResult.getQueueMessage());
return indexEventResult;
})
.toBlocking().lastOrDefault(null)
)//ack each message, but only if we didn't error. If we did, we'll want to log it and
- .subscribeOn( Schedulers.newThread() );
+ .subscribeOn(Schedulers.newThread());
//start in the background