You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@dolphinscheduler.apache.org by "Copilot (via GitHub)" <gi...@apache.org> on 2025/11/27 01:48:14 UTC
Re: [PR] [Improvement-17715][Registry] The registry center's lock mechanism supports reentrant acquisition. [dolphinscheduler]
Copilot commented on code in PR #17744:
URL: https://github.com/apache/dolphinscheduler/pull/17744#discussion_r2566933997
##########
dolphinscheduler-registry/dolphinscheduler-registry-plugins/dolphinscheduler-registry-jdbc/src/main/java/org/apache/dolphinscheduler/plugin/registry/jdbc/server/JdbcRegistryLockManager.java:
##########
@@ -92,6 +94,7 @@ public boolean acquireJdbcRegistryLock(Long clientId, String lockKey, long timeo
while (System.currentTimeMillis() - start <= timeout) {
LockEntry lockEntry = jdbcRegistryLockHolderMap.get(lockKey);
if (lockEntry != null && lockOwner.equals(lockEntry.getLockOwner())) {
+ lockEntry.lockCount.incrementAndGet();
Review Comment:
There is a potential race condition when checking for reentrancy. Between line 95 (checking if `lockEntry` exists) and line 97 (incrementing the counter), another thread could complete a release and remove the entry from the map. This could lead to incrementing a counter on a lock entry that is no longer in the map. Consider using `ConcurrentHashMap.compute()` or synchronization to ensure atomicity of the check-and-increment operation.
```suggestion
boolean reentered = jdbcRegistryLockHolderMap.compute(lockKey, (key, entry) -> {
if (entry != null && lockOwner.equals(entry.getLockOwner())) {
entry.lockCount.incrementAndGet();
return entry;
}
return entry;
}) != null && lockOwner.equals(jdbcRegistryLockHolderMap.get(lockKey).getLockOwner());
if (reentered) {
```
##########
dolphinscheduler-registry/dolphinscheduler-registry-plugins/dolphinscheduler-registry-it/src/test/java/org/apache/dolphinscheduler/plugin/registry/RegistryTestCase.java:
##########
@@ -258,6 +258,40 @@ public void testReleaseLock() {
assertThat(acquireResult.get()).isTrue();
}
+ @SneakyThrows
+ @Test
+ public void testReentrantLock() {
+ registry.start();
+ String lockKey = "/lock" + System.nanoTime();
+ // 1. Acquire the lock in the main thread
+ assertThat(registry.acquireLock(lockKey, 3000)).isTrue();
+ // Acquire the lock in the main thread
+ // It should acquire success
Review Comment:
The comment is misleading. It says "Acquire the lock in the main thread" but the actual behavior being tested is that the lock is acquired reentrantly (a second time) in the same thread that already holds it.
```suggestion
// 2. Acquire the lock reentrantly (a second time) in the same thread that already holds it
// It should acquire successfully due to reentrant locking
```
##########
dolphinscheduler-registry/dolphinscheduler-registry-plugins/dolphinscheduler-registry-it/src/test/java/org/apache/dolphinscheduler/plugin/registry/RegistryTestCase.java:
##########
@@ -258,6 +258,40 @@ public void testReleaseLock() {
assertThat(acquireResult.get()).isTrue();
}
+ @SneakyThrows
+ @Test
+ public void testReentrantLock() {
+ registry.start();
+ String lockKey = "/lock" + System.nanoTime();
+ // 1. Acquire the lock in the main thread
+ assertThat(registry.acquireLock(lockKey, 3000)).isTrue();
+ // Acquire the lock in the main thread
+ // It should acquire success
+ assertThat(registry.acquireLock(lockKey, 3000)).isTrue();
+
+ // 2. Acquire the lock at another thread
+ // It should acquire failed
+ CompletableFuture<Boolean> acquireResult =
+ CompletableFuture.supplyAsync(() -> registry.acquireLock(lockKey, 3000));
+ assertThat(acquireResult.get()).isFalse();
+
+ // 3. Release the lock in the main thread
+ assertThat(registry.releaseLock(lockKey)).isTrue();
+
+ // Acquire the lock at another thread
+ // It should acquire failed
+ acquireResult = CompletableFuture.supplyAsync(() -> registry.acquireLock(lockKey, 3000));
+ assertThat(acquireResult.get()).isFalse();
+
+ // 4. Release the lock in the main thread again
+ assertThat(registry.releaseLock(lockKey)).isTrue();
+
+ // Acquire the lock at another thread
+ // It should acquire success
+ acquireResult = CompletableFuture.supplyAsync(() -> registry.acquireLock(lockKey, 3000));
+ assertThat(acquireResult.get()).isTrue();
Review Comment:
The comment is misleading. It says "Acquire the lock at another thread" but the test is checking that the lock acquisition **succeeds** because the main thread has fully released it. Consider clarifying this with: "// Try to acquire the lock from another thread; it should now succeed because the main thread has fully released it"
##########
dolphinscheduler-registry/dolphinscheduler-registry-plugins/dolphinscheduler-registry-jdbc/src/main/java/org/apache/dolphinscheduler/plugin/registry/jdbc/server/JdbcRegistryLockManager.java:
##########
@@ -124,14 +127,22 @@ public boolean acquireJdbcRegistryLock(Long clientId, String lockKey, long timeo
@Override
public void releaseJdbcRegistryLock(Long clientId, String lockKey) {
+ String lockOwner = LockUtils.getLockOwner();
LockEntry lockEntry = jdbcRegistryLockHolderMap.get(lockKey);
- if (lockEntry == null) {
+ if (lockEntry == null || !lockOwner.equals(lockEntry.getLockOwner())) {
return;
}
if (!clientId.equals(lockEntry.getJdbcRegistryLock().getClientId())) {
throw new UnsupportedOperationException(
"The client " + clientId + " is not the lock owner of the lock: " + lockKey);
}
+ int newLockCount = lockEntry.lockCount.decrementAndGet();
+ if (newLockCount > 0) {
+ return;
+ }
+ if (newLockCount < 0) {
+ throw new IllegalMonitorStateException("Lock count has gone negative for lock: " + lockKey);
+ }
jdbcRegistryLockRepository.deleteById(lockEntry.getJdbcRegistryLock().getId());
jdbcRegistryLockHolderMap.remove(lockKey);
Review Comment:
There is a potential race condition in the release logic. Between checking the lock count at line 139-142 and removing the entry from the map at line 147, another thread could be attempting to acquire the lock. This could lead to a situation where the lock is removed from the map while another thread is checking for reentrancy, causing inconsistent state. Consider using synchronization or `ConcurrentHashMap.compute()` to ensure atomicity of the entire release operation.
##########
dolphinscheduler-registry/dolphinscheduler-registry-plugins/dolphinscheduler-registry-zookeeper/src/main/java/org/apache/dolphinscheduler/plugin/registry/zookeeper/ZookeeperRegistry.java:
##########
@@ -218,12 +218,6 @@ public boolean acquireLock(String key) {
try {
interProcessMutex =
Optional.ofNullable(processMutexMap.get(key)).orElse(new InterProcessMutex(client, key));
- if (interProcessMutex.isAcquiredInThisProcess()) {
- // Since etcd/jdbc cannot implement a reentrant lock, we need to check if the lock is already acquired
- // If it is already acquired, return true directly
- // This means you only need to release once when you acquire multiple times
- return true;
- }
interProcessMutex.acquire();
processMutexMap.put(key, interProcessMutex);
return true;
Review Comment:
The removal of the reentrant lock check before acquiring the lock is problematic. Curator's `InterProcessMutex` already supports reentrancy natively, and by removing the early return check, you're now calling `acquire()` multiple times on the same lock within the same thread. This changes the semantics: now you need to call `release()` the same number of times as `acquire()` was called. However, unlike JDBC and Etcd implementations which track the count explicitly, this relies on Curator's internal reentrancy counter. This could lead to inconsistent behavior across registry implementations if the release logic doesn't properly handle all reentrant acquisitions.
##########
dolphinscheduler-registry/dolphinscheduler-registry-plugins/dolphinscheduler-registry-it/src/test/java/org/apache/dolphinscheduler/plugin/registry/RegistryTestCase.java:
##########
@@ -258,6 +258,40 @@ public void testReleaseLock() {
assertThat(acquireResult.get()).isTrue();
}
+ @SneakyThrows
+ @Test
+ public void testReentrantLock() {
+ registry.start();
+ String lockKey = "/lock" + System.nanoTime();
+ // 1. Acquire the lock in the main thread
+ assertThat(registry.acquireLock(lockKey, 3000)).isTrue();
+ // Acquire the lock in the main thread
+ // It should acquire success
+ assertThat(registry.acquireLock(lockKey, 3000)).isTrue();
+
+ // 2. Acquire the lock at another thread
+ // It should acquire failed
+ CompletableFuture<Boolean> acquireResult =
+ CompletableFuture.supplyAsync(() -> registry.acquireLock(lockKey, 3000));
+ assertThat(acquireResult.get()).isFalse();
+
+ // 3. Release the lock in the main thread
+ assertThat(registry.releaseLock(lockKey)).isTrue();
+
+ // Acquire the lock at another thread
+ // It should acquire failed
Review Comment:
The comment is misleading. It says "Acquire the lock at another thread" but the test is checking that the lock acquisition **fails** because the main thread still holds it. Consider clarifying this with: "// Try to acquire the lock from another thread; it should fail because the main thread still holds it"
```suggestion
// Try to acquire the lock from another thread; it should fail because the main thread still holds it
```
##########
dolphinscheduler-registry/dolphinscheduler-registry-plugins/dolphinscheduler-registry-etcd/src/main/java/org/apache/dolphinscheduler/plugin/registry/etcd/EtcdRegistry.java:
##########
@@ -369,8 +374,18 @@ public boolean acquireLock(String key, long timeout) {
@Override
public boolean releaseLock(String key) {
try {
- Long leaseId = threadLocalLockMap.get().get(key);
- client.getLeaseClient().revoke(leaseId);
+ LockEntry lockEntry = threadLocalLockMap.get().get(key);
Review Comment:
There is a potential NPE risk when calling `threadLocalLockMap.get().get(key)`. If the ThreadLocal has been cleared by another operation or thread (e.g., at line 391), this will throw a NullPointerException. Consider checking if `threadLocalLockMap.get()` is null before calling `.get(key)` on it, similar to the pattern used in `acquireLock()` methods.
```suggestion
Map<String, LockEntry> lockMap = threadLocalLockMap.get();
if (lockMap == null) {
return true;
}
LockEntry lockEntry = lockMap.get(key);
```
##########
dolphinscheduler-registry/dolphinscheduler-registry-plugins/dolphinscheduler-registry-etcd/src/main/java/org/apache/dolphinscheduler/plugin/registry/etcd/EtcdRegistry.java:
##########
@@ -369,8 +374,18 @@ public boolean acquireLock(String key, long timeout) {
@Override
public boolean releaseLock(String key) {
try {
- Long leaseId = threadLocalLockMap.get().get(key);
- client.getLeaseClient().revoke(leaseId);
+ LockEntry lockEntry = threadLocalLockMap.get().get(key);
+ if (lockEntry == null) {
+ return true;
+ }
+ int newLockCount = lockEntry.lockCount.decrementAndGet();
+ if (newLockCount > 0) {
+ return true;
+ }
+ if (newLockCount < 0) {
+ throw new IllegalMonitorStateException("Lock count has gone negative for lock: " + key);
+ }
+ client.getLeaseClient().revoke(lockEntry.leaseId);
threadLocalLockMap.get().remove(key);
if (threadLocalLockMap.get().isEmpty()) {
threadLocalLockMap.remove();
Review Comment:
There is a potential NPE risk when calling `threadLocalLockMap.get().remove(key)` and `threadLocalLockMap.get().isEmpty()`. If the ThreadLocal has been cleared, this will throw a NullPointerException. Consider storing the result of `threadLocalLockMap.get()` in a local variable and checking for null before performing operations on it.
##########
dolphinscheduler-registry/dolphinscheduler-registry-plugins/dolphinscheduler-registry-jdbc/src/main/java/org/apache/dolphinscheduler/plugin/registry/jdbc/server/JdbcRegistryLockManager.java:
##########
@@ -56,6 +57,7 @@ public void acquireJdbcRegistryLock(Long clientId, String lockKey) {
while (true) {
LockEntry lockEntry = jdbcRegistryLockHolderMap.get(lockKey);
if (lockEntry != null && lockOwner.equals(lockEntry.getLockOwner())) {
+ lockEntry.lockCount.incrementAndGet();
Review Comment:
There is a potential race condition when checking for reentrancy. Between line 58 (checking if `lockEntry` exists) and line 60 (incrementing the counter), another thread could complete a release and remove the entry from the map. This could lead to incrementing a counter on a lock entry that is no longer in the map. Consider using `ConcurrentHashMap.compute()` or synchronization to ensure atomicity of the check-and-increment operation.
```suggestion
boolean reentered = jdbcRegistryLockHolderMap.compute(lockKey, (key, entry) -> {
if (entry != null && lockOwner.equals(entry.getLockOwner())) {
entry.lockCount.incrementAndGet();
return entry;
}
return entry;
}) != null && lockOwner.equals(jdbcRegistryLockHolderMap.get(lockKey).getLockOwner());
if (reentered) {
```
--
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: commits-unsubscribe@dolphinscheduler.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org