Changeset 534:099c597c0243 for kraken-logdb/src/main/java/org/krakenapps/logdb/query/command/Rpc.java

Timestamp:

11/14/11 11:47:19 (6 months ago)

Author:

xeraph

Branch:

default

Children:

535:c38ad740171a, 537:d8d85fe7264c

Message:

moved eof to separated rpc method (should be called by data session, otherwise, data loss will be occurred)

File:

• kraken-logdb/src/main/java/org/krakenapps/logdb/query/command/Rpc.java (modified) (5 diffs)

kraken-logdb/src/main/java/org/krakenapps/logdb/query/command/Rpc.java

@@ -19 +19 @@
 
         /**
-         * dist query guid
+         * mapreduce query guid
          */
         private String guid;
@@ -27 +27 @@
         private RpcConnection datastream;
         private RpcSession datasession;
+        private boolean end;
 
         public Rpc(String agentGuid, RpcConnection upstream, String guid, boolean sender) {
@@ -37 +38 @@
         @Override
         public void push(Map<String, Object> m) {
+                if (end) {
+                        logger.info("kraken logdb: loss (will be fixed) - {}", Primitive.stringify(m));
+                        return;
+                }
+
                 if (sender) {
                         if (datastream == null) {
@@ -77 +83 @@
         public void eof() {
                 if (datasession != null) {
+                        try {
+                                datasession.call("eof", guid);
+                        } catch (Exception e) {
+                                logger.error("kraken logdb: eof fail for mapreduce query " + guid, e);
+                        }
+
                         datastream.close();
                         datastream = null;
@@ -84 +96 @@
                 logger.info("kraken logdb: closed rpc data stream for query guid [{}]", guid);
                 super.eof();
+                end = true;
         }