ZooKeeper – Primer (contd.)

[tweetmeme source=”anismiles” only_single=false http://www.URL.com%5D

>> continued from here.

Watches

The power of Zookeeper comes from Watches. Watches allow clients to get notified when a znode changes in some way. Watches are set by operations, and are triggered by ZooKeeper when anything gets changed. For example, a watch can be placed on a znode which will be triggered when the znode data changes or the znode itself gets deleted.

The catch here is that Watches are triggered only ONCE. This might look pretty restrictive at first, but this helps keep ZooKeeper simple and if our client is insistent for more notifications it can always re-register the watch.

There are 9 basic operations in ZooKeeper.

Operation	Type	Description
create	Write	Creates a znode. (parent must already exist)
delete	Write	Deletes a znode (must not have any children)
exists	Read	Tests whether a znode exists and retrieves its metadata
getACL, setACL	–	Gets/Sets the ACL for a znode
getChildren	Read	Gets a list of the children of a znode
getData, setData	Read/Write	Gets/Sets the data associated with a znode
sync	–	Synchronizes a client’s view of a znode.

The rule is: Watches are set by read operations, and triggered by write operations. Isn’t it very intuitive?

As stated in previous post, znodes maintain version numbers for data changes, ACL changes, and timestamps, to allow cache validations and coordinated updates. Every time you want to change znode’s data, or its ACL information, you need to provide the correct versio,n and after successful operation, version number further gets incremented. You can relate it to Hibernate’s optimistic locking methodology, where every row is assigned with a version to resolve concurrent modification conflicts. Anyways, we are talking about Watches here.

Read operations like exists, getChildren and getData set the Watches. And these Watches are triggered by write operations like, create, delete and setData. Important point to note here is that ACL operations do not trigger or register any Watches, though they indeed mess with version numbers. When a Watch is triggered, a watch event is generated and passed to the Watcher which can do whatever it wishes to do with it. Let us now find out when and how various watch events are triggered.

1. Watch set with exists operation gets triggered when the znode is created, deleted or someone updates its data.
2. Watch set on getData gets triggered when the znode is deleted or someone updates its data.
3. Watch set on getChildren gets triggered when a new child is added or removed or when the znode itself gets deleted.

Let’s summarize it in a table:

ZooKeeper Watch operate on dual layer. You can specify a Watch while instantiating ZooKeeper object which will be notified about ZooKeeper’s state. The same Watch also gets notified for znode changes, if you haven’t specified any explicit Watch during read operations.

Let’s now try to connect to ZooKeeper.

public class ZkConnector {

    // ZooKeeper Object
    ZooKeeper zooKeeper;

    // To block any operation until ZooKeeper is connected. It's initialized
    // with count 1, that is, ZooKeeper connect state.
    java.util.concurrent.CountDownLatch connectedSignal = new java.util.concurrent.CountDownLatch(1);

    /**
     * Connects to ZooKeeper servers specified by hosts.
     *
     * @param hosts
     * @throws IOException
     * @throws InterruptedException
     */
    public void connect(String hosts) throws IOException, InterruptedException {
	zooKeeper = new ZooKeeper(
                hosts, // ZooKeeper service hosts
                5000,  // Session timeout in milliseconds
		// Anonymous Watcher Object
		new Watcher() {
        	    @Override
        	    public void process(WatchedEvent event) {
        		// release lock if ZooKeeper is connected.
        		if (event.getState() == KeeperState.SyncConnected) {
        		    connectedSignal.countDown();
        		}
        	    }
        	}
	);
	connectedSignal.await();
    }

    /**
     * Closes connection with ZooKeeper
     *
     * @throws InterruptedException
     */
    public void close() throws InterruptedException {
	zooKeeper.close();
    }

    /**
     * @return the zooKeeper
     */
    public ZooKeeper getZooKeeper() {
        // Verify ZooKeeper's validity
        if (null == zooKeeper || !zooKeeper.getState().equals(States.CONNECTED)){
	    throw new IllegalStateException ("ZooKeeper is not connected.");
        }
        return zooKeeper;
    }

}

The above class connects to the ZooKeeper service. When a ZooKeeper instance is created, connect() method, it starts a thread to connect to the service, and returns immediately. However, the constructor accepts a Watcher to notify about ZooKeeper state changes, we must wait for connection to get established before running any operation on ZooKeeper object.

In this example, I have used CountDownLatch class, which blocks the thread after ZooKeeper constructor has returned. This will hold the thread until its count is reduced by 1. When the client has changed its status, our anonymous Watcher receives a call to its process() method with WatchedEvent object, which then verifies the client’s state and reduces CountDownLatch counter by 1. And out object is ready to use.

This diagram captures ZooKeeper’s state transitions:

Okay. Since we are connected to ZooKeeper service, let’s try to do something meaningful.

Let’s say, we have two processes, pA and pB. The process pA picks up a chuck of data and performs some sort of operations, while process pB waits for pA to finish and then issues an email notifying about the data changes.

Simple, huh? Sure, it can be solved by using Java’s concurrent package. But we will do it using ZooKeeper for obvious gains like scalability. Here are the steps:

1. Define a znode say, /game_is_over

   final String myPath = "/game_is_over”;
   

2. Get ZooKeeper object

   ZkConnector zkc = new ZkConnector();
   zkc.connect("localhost");
   ZooKeeper zk = zkc.getZooKeeper();
   

3. pB registers a Watch with ZooKeeper service with exists operation. This Watch will receive a call once the znode becomes available.

   zk.exists(myPath, new Watcher() {		// Anonymous Watcher
   	@Override
   	public void process(WatchedEvent event) {
   	   // check for event type NodeCreated
      	   boolean isNodeCreated = event.getType().equals(EventType.NodeCreated);
   	   // verify if this is the defined znode
   	   boolean isMyPath = event.getPath().equals(myPath);
   
   	   if (isNodeCreated && isMyPath) {
   		//<strong>TODO</strong>: send an email or whatever
   	   }
   	}
   });
   

4. pA, after finishing its job, creates the znode. Effectively alerting pB to start working.

   zk.create(
   	myPath, 		// Path of znode
   	null,			// Data not needed.
   	Ids.OPEN_ACL_UNSAFE, 	// ACL, set to Completely Open.
   	CreateMode.PERSISTENT	// Znode type, set to Persistent.
   );
   

That was easy. As soon as pA finishes its job, it creates a znode, ideally this should have been an ephemeral znode, on which pB already has registered a Watch which gets triggered off immediately, notifying pB to do its job.
With similar models, you can implement various distributed data-structures, locks, barriers etc on top of ZooKeeper. I will write few more posts on this, but for now you can refer to ZooKeeper’s recipes.
So, this is what ZooKeeper start-up primer is. This will get you kick-started immediately. However, there are still some fundamentals left to cover, like session, ACL, consistency models etc. Keep checking this space, I will write more on these in near future.

ZooKeeper – Primer

[tweetmeme source=”anismiles” only_single=false http://www.URL.com%5D
Distributed collaborative applications involve a set of processes or agents interacting with one another to accomplish a common goal. They execute on Wide Area environments with little or no knowledge of the infrastructure and almost no control over the resources available. Besides, they need to sequence and order events, and ensure atomicity of actions. Above all, the application needs to keep itself from nightmarish bugs like race conditions, deadlocks and partial failures.

ZooKeeper helps to build a distributed application by working as a coordination service provider.

It’s reliable and highly available. It exposes a simple set of primitives upon which distributed applications can build higher level services for

• Synchronization,
• Configuration Maintenance,
• Groups,
• Naming,
• Leader elections and other niche needs.

What lies beneath?

ZooKeeper maintains a shared hierarchical namespace modeled after standard file systems. The namespace consists of data registers, called znodes. They are similar to files and directories.

Note: Znodes store data in Memory primarily, with a logged backup on disk for reliability. It means that whatever data znodes can keep must fit into memory, hence it must be small, max to 1MB. On the other hand, it means high throughput and low latency.

Znodes are identified by unique absolute paths which are “/” delimited Unicode strings. To help achieve uniqueness, ZooKeeper provides sequential znodes where a globally maintained sequence number will be appended by ZooKeeper to paths, i.e. path “/zoo-1/tiger/white-” can be assigned with a sequence, say 5, and will become “/zoo-1/tiger/white-5”.

1. A client can create a znode, store up to 1MB of data and associate as many as children znodes as it wants.
2. Data access to and fro a znode is always atomic. Either the data is read and/or written in its entirety or it fails.
3. There are no renames and no append semantics available.
4. Each znode has an Access Control List (ACL) that restricts who can do what.
5. Znodes maintain version numbers for data changes, ACL changes, and timestamps, to allow cache validations and coordinated updates.

Znodes can be one of two types: ephemeral and persistent. Once set, the type can’t be changed.

1. Ephemeral znodes are deleted by ZooKeeper when the creating client’s session gets closed, while persistent znodes stay as long as not deleted explicitly.
2. Ephemeral znodes can’t have children.
3. Both types of znodes are visible to all clients eligible with ACL policy.

Up and Running

There are enough literature on installing ZooKeeper on Linux machine already. So, I am going to focus how to install ZooKeeper on Windows machines.

1. Download and install Cygwin. http://www.cygwin.com/
2. Download stable release of ZooKeeper. http://hadoop.apache.org/zookeeper/releases.html
3. Unzip ZooKeeper to some directory, say, D:/iLabs/zookeeper-3.3.1
4. Add a new environment variable ZOOKEEPER_INSTALL and point it to D:/iLabs/zookeeper-3.3.1
5. Edit PATH variable and append $ZOOKEEPER_INSTALL/bin to it.
6. Now start Cygwin.

Now, start ZooKeeper server.

$ zkServer.sh start

ouch! It threw an error:

ZooKeeper exited abnormally because it could not find the configuration file, zoo.cfg, which it expects in
$ZOOKEEPER_INSTALL/conf directory. This is a standard Java properties file.

Go ahead and create zoo.cfg file in the conf directory. Open it up, and add below properties:

# The number of milliseconds of each tick
tickTime=2000

# The directory where the snapshot is stored.
dataDir=D:/iLabs/zoo-data/

# The port at which the clients will connect
clientPort=2181

Go back to Cygwin, and issue the same command again. This time ZooKeeper should load properly.

Now, connect to ZooKeeper. You should probably open a new Cygwin window, and issue the following command.

$ zkCli.sh

This will connect to your ZooKeeper server running at localhost:2181 by default, and will open zk console.

Let’s create a znode, say /zoo-1

[zk: localhost:2181<CONNECTED> 1] create -s /zoo-1 “Hello World!” null

Flag –s creates a persistent znode. Hello World! is the data you assign to znode (/zoo-1) and null is its ACL.

To see all znodes,

[zk: localhost:2181<CONNECTED> 2] ls /
[zoo-1, zookeeper]

This means, there are 2 nodes at the root level, /zoo-1 and /zookeeper. ZooKeeper uses the /zookeeper sub-tree to store management information, such as information on quotas.

For more commands, type help. If you want to further explore on the command line tools, refer: http://hadoop.apache.org/zookeeper/docs/current/zookeeperStarted.html

continue reading the primer >>