private InterProcessLock createTableLock() { String tableLockZNode = ZKUtil.joinZNode(zkWatcher.tableLockZNode, tableName.getNameAsString()); ZooKeeperProtos.TableLock data = ZooKeeperProtos.TableLock.newBuilder() .setTableName(ProtobufUtil.toProtoTableName(tableName)) .setLockOwner(ProtobufUtil.toServerName(serverName)) .setThreadId(Thread.currentThread().getId()) .setPurpose(purpose) .setIsShared(isShared) .setCreateTime(EnvironmentEdgeManager.currentTime()).build(); byte[] lockMetadata = toBytes(data); InterProcessReadWriteLock lock = new ZKInterProcessReadWriteLock(zkWatcher, tableLockZNode, METADATA_HANDLER); return isShared ? lock.readLock(lockMetadata) : lock.writeLock(lockMetadata); }
private InterProcessLock createTableLock() { String tableLockZNode = ZKUtil.joinZNode(zkWatcher.tableLockZNode, tableName.getNameAsString()); ZooKeeperProtos.TableLock data = ZooKeeperProtos.TableLock.newBuilder() .setTableName(ProtobufUtil.toProtoTableName(tableName)) .setLockOwner(ProtobufUtil.toServerName(serverName)) .setThreadId(Thread.currentThread().getId()) .setPurpose(purpose) .setIsShared(isShared) .setCreateTime(EnvironmentEdgeManager.currentTimeMillis()).build(); byte[] lockMetadata = toBytes(data); InterProcessReadWriteLock lock = new ZKInterProcessReadWriteLock(zkWatcher, tableLockZNode, METADATA_HANDLER); return isShared ? lock.readLock(lockMetadata) : lock.writeLock(lockMetadata); }
@Test(timeout = 60000) public void testMultipleClients() throws Exception { //tests lock usage from multiple zookeeper clients with different sessions. //acquire one read lock, then one write lock final String testName = "testMultipleClients"; //different zookeeper sessions with separate identifiers ZooKeeperWatcher zkWatcher1 = new ZooKeeperWatcher(conf, "testMultipleClients-1", null); ZooKeeperWatcher zkWatcher2 = new ZooKeeperWatcher(conf, "testMultipleClients-2", null); String znode = ZKUtil.joinZNode(zkWatcher1.tableLockZNode, testName); ZKInterProcessReadWriteLock clientLock1 = new ZKInterProcessReadWriteLock(zkWatcher1, znode, null); ZKInterProcessReadWriteLock clientLock2 = new ZKInterProcessReadWriteLock(zkWatcher2, znode, null); InterProcessLock lock1 = clientLock1.readLock(Bytes.toBytes("client1")); lock1.acquire(); //try to acquire, but it will timeout. We are testing whether this will cause any problems //due to the read lock being from another client InterProcessLock lock2 = clientLock2.writeLock(Bytes.toBytes("client2")); assertFalse(lock2.tryAcquire(1000)); lock1.release(); //this time it will acquire assertTrue(lock2.tryAcquire(5000)); lock2.release(); zkWatcher1.close(); zkWatcher2.close(); }