implemented scan and cleanup

zookeeper/src/main/java/site/ycsb/db/ZookeeperClient.java

@@ -5,18 +5,16 @@ import org.apache.zookeeper.KeeperException;
 import org.apache.zookeeper.ZooDefs;
 import org.apache.zookeeper.ZooKeeper;
 import org.apache.zookeeper.data.Stat;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import site.ycsb.*;
 
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Set;
+import java.util.*;
 
 import org.json.JSONObject;
 
-import java.util.Vector;
-
 /**
  * A client that can be used by YCSB to work with Zookeeper.
  */
@@ -24,6 +22,8 @@ public class ZookeeperClient extends DB {
 
   private static ZooKeeper zooKeeper;
   private static ZKConnection zkConnection;
+  private Logger logger = LoggerFactory.getLogger(ZookeeperClient.class);
+
 
   public void init() throws DBException {
     zkConnection = new ZKConnection();
@@ -65,7 +65,38 @@ public class ZookeeperClient extends DB {
   @Override
   public Status scan(String table, String startkey,
                      int recordcount, Set<String> fields, Vector<HashMap<String, ByteIterator>> result) {
-    return null;
+    try {
+      List<String> children = zooKeeper.getChildren("/", true);
+      Collections.sort(children);
+
+      // go through all children to find start key
+      for (int i = 0; i < children.size(); i++) {
+        if (children.get(i).equals(startkey)) {
+
+          // make sure the number of records left is greater or equal to recordcount
+          if (children.size() - i < recordcount) {
+            return Status.ERROR;
+          }
+
+          // read all records
+          for (int j = i; j < recordcount + i; j++) {
+            HashMap<String, ByteIterator> m = new HashMap<>();
+            Status status = read(table, children.get(j), fields, m);
+
+            // make sure the status is ok
+            if (status != Status.OK) {
+              return status;
+            }
+
+            result.add(m);
+          }
+        }
+      }
+    } catch (KeeperException | InterruptedException e) {
+      e.printStackTrace();
+    }
+
+    return Status.OK;
   }
 
   @Override
@@ -132,9 +163,29 @@ public class ZookeeperClient extends DB {
     } catch (InterruptedException | KeeperException e) {
       e.printStackTrace();
     }
+
     return Status.OK;
   }
 
+  @Override
+  public void cleanup() {
+    try {
+      List<String> children = zooKeeper.getChildren("/", true);
+
+      // delete all znodes except for zookeeper znode
+      for (String child : children) {
+        if (!child.equals("zookeeper")) {
+
+          // get pathStat of child znode
+          Stat pathStat = checkExists(child);
+          zooKeeper.delete(prefixKeyWithSlash(child), pathStat.getVersion());
+        }
+      }
+    } catch (KeeperException | InterruptedException e) {
+      e.printStackTrace();
+    }
+  }
+
   // all znodes start with backslash, so start key with backslash
   private String prefixKeyWithSlash(String key) {
     return  ("/" + key);
@@ -155,4 +206,5 @@ public class ZookeeperClient extends DB {
   }
 
 
+
 }
\ No newline at end of file