Changeset 534:099c597c0243

Timestamp:

11/14/11 11:47:19 (3 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)

Location:

kraken-logdb/src/main/java/org/krakenapps/logdb

Files:

• impl/MapReduceRpcService.java (modified) (2 diffs)
• query/command/Rpc.java (modified) (5 diffs)

kraken-logdb/src/main/java/org/krakenapps/logdb/impl/MapReduceRpcService.java

@@ -222 +222 @@
         }
 
+        @RpcMethod(name = "eof")
+        public void eof(String queryGuid) {
+                // TODO: check if all mapper queries are ended
+                // for now, send eof if one mapper query is ended
+                RpcSession session = RpcContext.getSession();
+                String nodeGuid = session.getConnection().getPeerGuid();
+
+                Rpc rpc = rpcFromMap.get(queryGuid);
+                if (rpc != null)
+                        rpc.eof();
+                else
+                        logger.warn("kraken logdb: rpcfrom not found for mapreduce query [{}]", queryGuid);
+        }
+
         @RpcMethod(name = "createMapQuery")
         public int createMapQuery(String queryGuid, String query) {
@@ -306 +320 @@
                         if (q != null)
                                 q.setEnd(true);
-
-                        // TODO: check if all mapper queries are ended
-                        // for now, send eof if one mapper query is ended
-                        String queryGuid = remoteQueryMappings.get(key);
-                        Rpc rpc = rpcFromMap.get(queryGuid);
-                        if (rpc != null)
-                                rpc.eof();
-                        else
-                                logger.warn("kraken logdb: rpcfrom not found for mapreduce query [{}]", queryGuid);
                 }
                         break;

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;
         }