Guosmilesmile commented on code in PR #13795:
URL: https://github.com/apache/iceberg/pull/13795#discussion_r2271920136
##########
flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/ZkLockFactory.java:
##########
@@ -132,23 +165,38 @@ public void close() throws IOException {
/** Zookeeper lock implementation */
private static class ZkLock implements Lock {
private final SharedCount sharedCount;
+ private final String lockId;
+ private final String lockType;
+ private final String lockPath;
+
+ private static final int LOCKED = 1;
+ private static final int UNLOCKED = 0;
- private ZkLock(SharedCount sharedCount) {
+ private ZkLock(String lockId, String lockType, String lockPath,
SharedCount sharedCount) {
+ this.lockId = lockId;
+ this.lockType = lockType;
+ this.lockPath = lockPath;
this.sharedCount = sharedCount;
}
@Override
public boolean tryLock() {
VersionedValue<Integer> versionedValue = sharedCount.getVersionedValue();
if (isHeld(versionedValue)) {
- LOG.debug("Lock is already held for {}", this);
+ LOG.debug(
+ "Lock is already held for lockId: {}, type: {}, path: {}.",
lockId, lockType, lockPath);
return false;
}
try {
- return sharedCount.trySetCount(versionedValue, LOCKED);
+ boolean acquired = sharedCount.trySetCount(versionedValue, LOCKED);
+ if (!acquired) {
+ LOG.warn(
+ "Failed to acquire {} lock for lockId: {}, path: {}", lockType,
lockId, lockPath);
+ }
+ return acquired;
Review Comment:
nit: new line
##########
flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/ZkLockFactory.java:
##########
@@ -90,26 +104,45 @@ public void open() {
throw new IllegalStateException("Connection to Zookeeper timed out");
}
- this.taskSharedCount = new SharedCount(client, LOCK_BASE_PATH + lockId +
"/task", 0);
- this.recoverySharedCount = new SharedCount(client, LOCK_BASE_PATH +
lockId + "/recovery", 0);
+ this.taskSharedCount = new SharedCount(client, getTaskSharePath(), 0);
+ this.recoverySharedCount = new SharedCount(client,
getRecoverySharedPath(), 0);
taskSharedCount.start();
recoverySharedCount.start();
+ isOpen = true;
+ LOG.info("ZkLockFactory initialized for lockId: {}.", lockId);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new RuntimeException("Interrupted while connecting to Zookeeper",
e);
} catch (Exception e) {
+ closeQuietly();
throw new RuntimeException("Failed to initialize SharedCount", e);
}
}
+ private String getTaskSharePath() {
+ return LOCK_BASE_PATH + lockId + "/task";
+ }
+
+ private String getRecoverySharedPath() {
+ return LOCK_BASE_PATH + lockId + "/recovery";
+ }
+
+ private void closeQuietly() {
+ try {
+ close();
+ } catch (Exception e) {
+ LOG.warn("Failed to close ZkLockFactory for lockId: {}.", lockId, e);
+ }
+ }
+
@Override
public Lock createLock() {
- return new ZkLock(taskSharedCount);
+ return new ZkLock(lockId, "taskShare", getTaskSharePath(),
taskSharedCount);
Review Comment:
Maybe `Task` is better.
##########
flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestZkLockFactory.java:
##########
@@ -39,15 +48,32 @@ void before() {
super.before();
}
- @Override
- TriggerLockFactory lockFactory() {
- return new ZkLockFactory(zkTestServer.getConnectString(), testLockId,
5000, 3000, 1000, 3);
+ @Test
+ void testMultiTableLock() throws IOException {
+ // Create two lock factories for different tables
+ ZkLockFactory other = lockFactory("tableName2");
+ other.open();
+ TriggerLockFactory.Lock lock1 = lockFactory.createLock();
+ TriggerLockFactory.Lock lock2 = other.createLock();
+
+ // Verify that locks for different tables can be acquired independently
+ assertThat(lock1.tryLock()).isTrue();
+ assertThat(lock2.tryLock()).isTrue();
+
+ // Clean up
+ lock1.unlock();
+ lock2.unlock();
+ other.close();
}
- @After
+ ZkLockFactory lockFactory(String tableName) {
+ String lockId = "test_job_" + tableName + "_" +
UUID.randomUUID().toString().replace("-", "");
Review Comment:
Can we directly pass the `LockId` instead of passing the `tableName` and
then constructing the `LockId`? It seems that the current usage essentially
distinguishes different `LockId`s using the table name directly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
