Wacther question

classic Classic list List threaded Threaded
3 messages Options
Reply | Threaded
Open this post in threaded view
|

Wacther question

Neo Anderson
My question is - what is the right way to received notification once a znode is created?

My scenario is that I have two threads, one (thread A) is going to save data to hdfs, another thread (thread B) is going to delete data stored in hdfs once thread A finishes its procedure. Thread A will create a znode and Thread B uses zk.exists() to register a watch.

Thread A will only do

zk.create(path, null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

at the end (in the final block) of a clean function.


The watcher in therad B

      Watcher watcher = new Watcher(){
        public void process(WatchedEvent e){
          boolean exist = e.getType().equals(EventType.NodeCreated);
          boolean path = e.getPath().equals("/"+job_id+"/"+task_id);
          if(exist && path){
              // start cleaning data in hdfs
          }
        }

and

pass watcher to zk.exists(path, watcher);

but I gets `KeeperErrorCode = NoNode for' exception

org.apache.zookeeper.KeeperException$NoNodeException: KeeperErrorCode = NoNode for /job_id/task_id
        at org.apache.zookeeper.KeeperException.create(KeeperException.java:102)
        at org.apache.zookeeper.KeeperException.create(KeeperException.java:42)
        at org.apache.zookeeper.ZooKeeper.create(ZooKeeper.java:637)

It looks like znode doesn't exists because thread A still not yet create znode with path '/job_id/task_id'; however, if I switch to make thread A create znode first then therad B executes zk.exists(path, watcher). Problems becomes thread B (or its watcher) would never get triggered deleting data in hdfs.

How to fix this?

I appreciate any suggestion.

Thanks.







Reply | Threaded
Open this post in threaded view
|

Re: Wacther question

Jared Cantwell
Neo,

To me it looks like this exception is while you are trying to call:

zk.create(path, null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

where:

path = "/job_id/task_id/"

Does the node /job_id already exist when you are making this request?
Zookeeper does not recursively create the path you specify, only the
leaf node.  If a parent in the path does not exist,  you can get this
exception I believe.

As for your concern about the ordering of calling create and exist,
have you tried something like this:

if(zk.exists(path, watcher))
    // start cleaning data in hdfs

If the watcher isn't going to get called (because the node already
exists), you can take action immediately rather than in the watcher
callback...

~Jared


On Sat, Nov 6, 2010 at 7:50 AM, Neo Anderson
<[hidden email]> wrote:

> My question is - what is the right way to received notification once a znode is created?
>
> My scenario is that I have two threads, one (thread A) is going to save data to hdfs, another thread (thread B) is going to delete data stored in hdfs once thread A finishes its procedure. Thread A will create a znode and Thread B uses zk.exists() to register a watch.
>
> Thread A will only do
>
> zk.create(path, null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
>
> at the end (in the final block) of a clean function.
>
>
> The watcher in therad B
>
>      Watcher watcher = new Watcher(){
>        public void process(WatchedEvent e){
>          boolean exist = e.getType().equals(EventType.NodeCreated);
>          boolean path = e.getPath().equals("/"+job_id+"/"+task_id);
>          if(exist && path){
>              // start cleaning data in hdfs
>          }
>        }
>
> and
>
> pass watcher to zk.exists(path, watcher);
>
> but I gets `KeeperErrorCode = NoNode for' exception
>
> org.apache.zookeeper.KeeperException$NoNodeException: KeeperErrorCode = NoNode for /job_id/task_id
>        at org.apache.zookeeper.KeeperException.create(KeeperException.java:102)
>        at org.apache.zookeeper.KeeperException.create(KeeperException.java:42)
>        at org.apache.zookeeper.ZooKeeper.create(ZooKeeper.java:637)
>
> It looks like znode doesn't exists because thread A still not yet create znode with path '/job_id/task_id'; however, if I switch to make thread A create znode first then therad B executes zk.exists(path, watcher). Problems becomes thread B (or its watcher) would never get triggered deleting data in hdfs.
>
> How to fix this?
>
> I appreciate any suggestion.
>
> Thanks.
>
>
>
>
>
>
>
>
Reply | Threaded
Open this post in threaded view
|

Re: Wacther question

Neo Anderson

Hi Jared,

Thank you. The problem is exactly as you explain below. After checking if the parent path e.g. job_id exists first, then the problem goes away.

Thanks for your help.

--- On Tue, 9/11/10, Jared Cantwell <[hidden email]> wrote:

> From: Jared Cantwell <[hidden email]>
> Subject: Re: Wacther question
> To: [hidden email]
> Date: Tuesday, 9 November, 2010, 20:18
> Neo,
>
> To me it looks like this exception is while you are trying
> to call:
>
> zk.create(path, null, Ids.OPEN_ACL_UNSAFE,
> CreateMode.PERSISTENT);
>
> where:
>
> path = "/job_id/task_id/"
>
> Does the node /job_id already exist when you are making
> this request?
> Zookeeper does not recursively create the path you specify,
> only the
> leaf node.  If a parent in the path does not
> exist,  you can get this
> exception I believe.
>
> As for your concern about the ordering of calling create
> and exist,
> have you tried something like this:
>
> if(zk.exists(path, watcher))
>     // start cleaning data in hdfs
>
> If the watcher isn't going to get called (because the node
> already
> exists), you can take action immediately rather than in the
> watcher
> callback...
>
> ~Jared
>
>
> On Sat, Nov 6, 2010 at 7:50 AM, Neo Anderson
> <[hidden email]>
> wrote:
> > My question is - what is the right way to received
> notification once a znode is created?
> >
> > My scenario is that I have two threads, one (thread A)
> is going to save data to hdfs, another thread (thread B) is
> going to delete data stored in hdfs once thread A finishes
> its procedure. Thread A will create a znode and Thread B
> uses zk.exists() to register a watch.
> >
> > Thread A will only do
> >
> > zk.create(path, null, Ids.OPEN_ACL_UNSAFE,
> CreateMode.PERSISTENT);
> >
> > at the end (in the final block) of a clean function.
> >
> >
> > The watcher in therad B
> >
> >      Watcher watcher = new Watcher(){
> >        public void process(WatchedEvent e){
> >          boolean exist =
> e.getType().equals(EventType.NodeCreated);
> >          boolean path =
> e.getPath().equals("/"+job_id+"/"+task_id);
> >          if(exist && path){
> >              // start cleaning data in hdfs
> >          }
> >        }
> >
> > and
> >
> > pass watcher to zk.exists(path, watcher);
> >
> > but I gets `KeeperErrorCode = NoNode for' exception
> >
> > org.apache.zookeeper.KeeperException$NoNodeException:
> KeeperErrorCode = NoNode for /job_id/task_id
> >        at
> org.apache.zookeeper.KeeperException.create(KeeperException.java:102)
> >        at
> org.apache.zookeeper.KeeperException.create(KeeperException.java:42)
> >        at
> org.apache.zookeeper.ZooKeeper.create(ZooKeeper.java:637)
> >
> > It looks like znode doesn't exists because thread A
> still not yet create znode with path '/job_id/task_id';
> however, if I switch to make thread A create znode first
> then therad B executes zk.exists(path, watcher). Problems
> becomes thread B (or its watcher) would never get triggered
> deleting data in hdfs.
> >
> > How to fix this?
> >
> > I appreciate any suggestion.
> >
> > Thanks.
> >
> >
> >
> >
> >
> >
> >
> >
>