LockingQueue bug

[vipjml]

[zk: localhost:2181(CONNECTED) 39] ls /videoseg/taskqueue/entries
[entry-201-0000059505]
[zk: localhost:2181(CONNECTED) 40] ls /videoseg/taskqueue/taken
[entry-201-0000059505, entry-401-0000059506]
this is my LockingQueue after runnig for a while,LockingQueue not clean entry in taken;

def _take(self, id_):
        try:
            self.client.create(
                "{path}/{id}".format(
                    path=self._lock_path,
                    id=id_),
                self.id,
                ephemeral=True)
            value, stat = self.client.retry(
                self.client.get,
                "{path}/{id}".format(path=self._entries_path, id=id_))
        except (NoNodeError, NodeExistsError):
            # Item is already consumed or locked
            return None
        return (id_, value)

this code may problem produce this bug. the ephemeral node should be removed if

  value, stat = self.client.retry(
                self.client.get,
                "{path}/{id}".format(path=self._entries_path, id=id_)) 

failed.

for example two client A,B.
step1:A check and find a item
step2: A try to lock the item by create a ephemeral Node,A check ensure get the lock
if B lock and consume the same item between step1 and step2,then the ephemeral node created by A will not be removed

[bsanders]

Is this the same issue as #159 (which has a pull request #347 with unittest and still applies cleanly)?

[vipjml]

no,not the same, #347 is a improvement,but the problem here is a bug.
I write a patch #373, i close is due to test problem,later i will reopen it. the follow code should fix it
queue.py:321

def _take(self, id_):
        try:
            self.client.create(
                "{path}/{id}".format(
                    path=self._lock_path,
                    id=id_),
                self.id,
                ephemeral=True)
        except NodeExistsError:
            return None

        try:
            value, stat = self.client.retry(
                self.client.get,
                "{path}/{id}".format(path=self._entries_path, id=id_))
        except NoNodeError:
            # Item is already consumed or locked
            self.client.retry(
            self.client.delete,
            "{path}/{id}".format(path=self._lock_path,id=id_))
            return None
        return (id_, value)