ZooKeeper for Beginners (2): Programming with the ZooKeeper API

Tutorial index: Big Data for Beginners: tutorial series

Last post got ZooKeeper installed. The command line works, but most of the time you operate ZooKeeper through API calls, so this post covers the basic beginner-level operations.

All code here is public at https://github.com/renfei/demo/tree/master/zookeeper/zookeeper-zpi

Prerequisites

Since we’re demoing many operations and each needs a fresh client, let’s be lazy and use JUnit’s @Before to build a new client for us every time:

public class ZookeeperApiDemo {
    private static final String CONNECT_STRING = "localhost:2181";
    private static final int SESSION_TIMEOUT = 2000;
    private ZooKeeper zkClient = null;
    @Before
    public void init() throws Exception {
        zkClient = new ZooKeeper(CONNECT_STRING, SESSION_TIMEOUT, event -> {
            // Callback fired after an event notification (your business logic)
            System.out.println(event.getType() + "--" + event.getPath());
            // Re-register the watcher
            try {
                zkClient.getChildren("/", true);
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
    }
}

Full source: https://github.com/renfei/demo/blob/master/zookeeper/zookeeper-zpi/src/main/java/net/renfei/zookeeper/ZookeeperApiDemo.java

Basic CRUD

/**
 * Create a child node
 *
 * @throws Exception
 */
@Test
public void create() throws Exception {
    // Args: node path to create; node data; node ACL; node type
    String nodeCreated = zkClient.create("/renfei", "demo".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    System.out.println(nodeCreated);
}
/**
 * Set data
 *
 * @throws KeeperException
 * @throws InterruptedException
 */
@Test
public void set() throws KeeperException, InterruptedException {
    Stat stat = zkClient.setData("/renfei", "how big".getBytes(), 0);
    System.out.println(stat.toString());
}
/**
 * Get data
 *
 * @throws KeeperException
 * @throws InterruptedException
 */
@Test
public void get() throws KeeperException, InterruptedException {
    Stat stat = new Stat();
    byte[] dataBytes = zkClient.getData("/renfei", true, stat);
    // Note the version number: writes fail if the version doesn't match
    System.out.println(stat.getVersion());
    System.out.println(new String(dataBytes));
}
/**
 * Existence check
 *
 * @throws KeeperException
 * @throws InterruptedException
 */
@Test
public void exists() throws KeeperException, InterruptedException {
    Stat stat = zkClient.exists("/renfei", false);
    if (stat == null) {
        System.out.println("node does not exist");
    } else {
        System.out.println(stat.getDataLength());
    }
}
/**
 * Delete data
 *
 * @throws KeeperException
 * @throws InterruptedException
 */
@Test
public void delete() throws KeeperException, InterruptedException {
    Stat stat = zkClient.exists("/renfei", false);
    if (stat != null) {
        zkClient.delete("/renfei", stat.getVersion());
    }
}

Registering a Watcher

ZooKeeper gives us a change-notification mechanism for nodes, so we can watch a node: when its data changes, ZooKeeper hands us the latest data and state.

Watching is asynchronous, which means it needs two threads — so the main thread has to sleep and wait, or it exits and takes the child thread with it.

public void register() throws KeeperException, InterruptedException {
    byte[] data = zkClient.getData("/renfei", watchedEvent -> {
        try {
            // Recursive call here, so it keeps watching for changes
            register();
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }, new Stat());
    System.out.println(new String(data));
}
@Test
public void registerTest() throws InterruptedException {
    try {
        register();
    } catch (KeeperException e) {
        e.printStackTrace();
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    // Block the main thread so we can watch register() recurse repeatedly
    Thread.sleep(Long.MAX_VALUE);
}