Changeset 534:099c597c0243 for kraken-logdb/src/main/java/org/krakenapps/logdb/impl/MapReduceRpcService.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/impl/MapReduceRpcService.java (modified) (2 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;