usergrid-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sfeld...@apache.org
Subject [1/2] usergrid git commit: add better exception handling
Date Fri, 09 Oct 2015 14:54:40 GMT
Repository: usergrid
Updated Branches:
  refs/heads/2.1-release b48551722 -> 80324de9e


add better exception handling


Project: http://git-wip-us.apache.org/repos/asf/usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/usergrid/commit/3cb0a0e0
Tree: http://git-wip-us.apache.org/repos/asf/usergrid/tree/3cb0a0e0
Diff: http://git-wip-us.apache.org/repos/asf/usergrid/diff/3cb0a0e0

Branch: refs/heads/2.1-release
Commit: 3cb0a0e0c5a8a1f1e06d3bd71b9a2bcef333ca80
Parents: b485517
Author: Shawn Feldman <sfeldman@apache.org>
Authored: Fri Oct 9 08:39:27 2015 -0600
Committer: Shawn Feldman <sfeldman@apache.org>
Committed: Fri Oct 9 08:39:27 2015 -0600

----------------------------------------------------------------------
 .../asyncevents/AmazonAsyncEventService.java     | 19 ++++++++++++++-----
 1 file changed, 14 insertions(+), 5 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/usergrid/blob/3cb0a0e0/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 0ac3860..b384b80 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
@@ -307,7 +307,9 @@ public class AmazonAsyncEventService implements AsyncEventService {
             }
         });
         //resolve the list and return it.
-        final List<IndexEventResult> indexEventResults = masterObservable.toList().toBlocking().lastOrDefault(null);
+        final List<IndexEventResult> indexEventResults = masterObservable
+            .collect(() -> new ArrayList<IndexEventResult>(), (list,indexEventResult)
-> list.add(indexEventResult) )
+            .toBlocking().lastOrDefault(null);
         //if nothing came back then return null
         if(indexEventResults==null){
             return null;
@@ -318,13 +320,16 @@ public class AmazonAsyncEventService implements AsyncEventService {
         //stream and filer the messages
         List<QueueMessage> messagesToAck = indexEventResults.stream()
             .map(indexEventResult -> {
+                //collect into the index submission
                 if (indexEventResult.getIndexOperationMessage().isPresent()) {
                     combined.ingest(indexEventResult.getIndexOperationMessage().get());
                 }
                 return indexEventResult;
             })
+            //filter out the ones that need to be ack'd
             .filter(indexEventResult -> indexEventResult.getQueueMessage().isPresent())
             .map(indexEventResult -> {
+                //record the cycle time
                 messageCycle.update(System.currentTimeMillis() - indexEventResult.getCreationTime());
                 return indexEventResult;
             })
@@ -333,8 +338,13 @@ public class AmazonAsyncEventService implements AsyncEventService {
             .collect(Collectors.toList());
 
         //send the batch
-        indexProducer.put(combined).toBlocking().lastOrDefault(null);
-
+        //TODO: should retry?
+        try {
+            indexProducer.put(combined).toBlocking().lastOrDefault(null);
+        }catch (Exception e){
+            logger.error("Failed to submit to index producer",messages,e);
+            throw e;
+        }
         return messagesToAck;
     }
 
@@ -584,10 +594,9 @@ public class AmazonAsyncEventService implements AsyncEventService {
                                     }
                                     //ack each message, but only if we didn't error.
                                     ack(messagesToAck);
-                                    //messagesToAck.stream().forEach(message -> ack(message));
                                     return messagesToAck;
                                 } catch (Exception e) {
-                                    logger.error("failed to ack messages to sqs", messages.get(0).getMessageId(),
e);
+                                    logger.error("failed to ack messages to sqs", messages,
e);
                                     return null;
                                     //do not rethrow so we can process all of them
                                 }


Mime
View raw message