Fix that ConcurrentLinkedDeque.removeIf is not actually thread-safe (apache#16598)

Author: jt2594838

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/auth/LoginLockManager.java

@@ -100,14 +100,19 @@ public 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,15 +196,16 @@ public void recordFailure(long userId, String ip) {
           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);
             // Record this failure
             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,16 +218,16 @@ public void recordFailure(long userId, String ip) {
           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);
             // Record this failure
             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,

iotdb-core/datanode/src/test/java/org/apache/iotdb/db/auth/LoginLockManagerTest.java

@@ -533,9 +533,9 @@ public void testConcurrentLockStateConsistency() throws InterruptedException {
 
   @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++) {