handle data corruption

zhuker · Oct 15, 2017 · b2b49d2 · b2b49d2
1 parent 942d2ad
commit b2b49d2
Showing 1 changed file with 11 additions and 1 deletion.
diff --git a/src/main/java/rtalk/RTalk.java b/src/main/java/rtalk/RTalk.java
@@ -389,12 +389,22 @@ public synchronized Response reserve(long blockTimeoutMsec) {
                 }
             }
 
-            Set<String> ids = r.zrange(kReadyQueue, 0, -1);
+            Set<String> ids = r.zrange(kReadyQueue, 0, 0);
             Optional<Job> firstJob_ = ids
                     .stream()
                     .map(id -> _getJob(r, id))
                     .filter(j -> j != null && !Job.BURIED.equals(j.state))
                     .findFirst();
+            if (!firstJob_.isPresent() && (readyQueueSize != 0 || toLong(r.zcard(kReadyQueue)) != 0)) {
+                System.err.println("data corruption detected on tube " + tube);
+                ids = r.zrange(kReadyQueue, 0, -1);
+                firstJob_ = ids
+                        .stream()
+                        .map(id -> _getJob(r, id))
+                        .filter(j -> j != null && !Job.BURIED.equals(j.state))
+                        .findFirst();
+
+            }
             return firstJob_;
         });