You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ji...@apache.org on 2025/10/16 03:38:42 UTC
(iotdb) 01/01: Fix that ConcurrentLinkedDeque.removeIf is not actually thread-safe
This is an automated email from the ASF dual-hosted git repository.
jiangtian pushed a commit to branch fix_userlockinfo_concurrency2
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 3bb16c8e5d23738d78cfcb5ce0fbd0fd43279a7a
Author: Tian Jiang <jt...@163.com>
AuthorDate: Thu Oct 16 11:38:29 2025 +0800
Fix that ConcurrentLinkedDeque.removeIf is not actually thread-safe
---
.../org/apache/iotdb/db/auth/LoginLockManager.java | 22 ++++++++++++++--------
.../apache/iotdb/db/auth/LoginLockManagerTest.java | 2 +-
2 files changed, 15 insertions(+), 9 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/auth/LoginLockManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/auth/LoginLockManager.java
index 05e80cc23cc..0174123dbc7 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/auth/LoginLockManager.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/auth/LoginLockManager.java
@@ -100,14 +100,19 @@ public class LoginLockManager {
/** Inner class to store user lock information */
static class UserLockInfo {
+
// Deque to store timestamps of failed attempts (milliseconds)
- private final Deque<Long> failureTimestamps = new ConcurrentLinkedDeque<>();
+ private final Deque<Long> failureTimestamps;
+
+ UserLockInfo(int capacity) {
+ failureTimestamps = new ConcurrentLinkedDeque<>();
+ }
- void addFailureTime(long timestamp) {
+ synchronized void addFailureTime(long timestamp) {
failureTimestamps.addLast(timestamp);
}
- void removeOldFailures(long cutoffTime) {
+ synchronized void removeOldFailures(long cutoffTime) {
// Remove timestamps older than cutoffTime
failureTimestamps.removeIf(timestamp -> timestamp < cutoffTime);
}
@@ -191,7 +196,8 @@ public class LoginLockManager {
userIpKey,
(key, existing) -> {
if (existing == null) {
- existing = new UserLockInfo();
+ existing =
+ new UserLockInfo(Math.max(failedLoginAttempts, failedLoginAttemptsPerUser));
}
// Remove failures outside of sliding window
existing.removeOldFailures(cutoffTime);
@@ -199,7 +205,7 @@ public class LoginLockManager {
existing.addFailureTime(now);
// Check if threshold reached (log only when it just reaches)
int failCountIp = existing.getFailureCount();
- if (failCountIp >= failedLoginAttempts && failCountIp == failedLoginAttempts) {
+ if (failCountIp >= failedLoginAttempts) {
LOGGER.info("IP '{}' locked for user ID '{}'", ip, userId);
}
return existing;
@@ -212,7 +218,8 @@ public class LoginLockManager {
userId,
(key, existing) -> {
if (existing == null) {
- existing = new UserLockInfo();
+ existing =
+ new UserLockInfo(Math.max(failedLoginAttempts, failedLoginAttemptsPerUser));
}
// Remove failures outside of sliding window
existing.removeOldFailures(cutoffTime);
@@ -220,8 +227,7 @@ public class LoginLockManager {
existing.addFailureTime(now);
// Check if threshold reached (log only when it just reaches)
int failCountUser = existing.getFailureCount();
- if (failCountUser >= failedLoginAttemptsPerUser
- && failCountUser == failedLoginAttemptsPerUser) {
+ if (failCountUser >= failedLoginAttemptsPerUser) {
LOGGER.info(
"User ID '{}' locked due to {} failed attempts",
userId,
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/auth/LoginLockManagerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/auth/LoginLockManagerTest.java
index 535ec462496..86f78e30597 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/auth/LoginLockManagerTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/auth/LoginLockManagerTest.java
@@ -533,9 +533,9 @@ public class LoginLockManagerTest {
@Test
public void testConcurrentOperateLockInfo() throws InterruptedException, ExecutionException {
- UserLockInfo userLockInfo = new UserLockInfo();
int numThreads = 100;
final int numAttempts = 100000;
+ UserLockInfo userLockInfo = new UserLockInfo(numThreads * numAttempts);
ExecutorService executor = Executors.newFixedThreadPool(numThreads);
List<Future<Void>> threads = new ArrayList<>(numThreads);
for (int i = 0; i < numThreads; i++) {