[solr] branch jira/solr16175 created (now c1ae998e7ad)

[no...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

noble pushed a change to branch jira/solr16175
in repository https://gitbox.apache.org/repos/asf/solr.git


      at c1ae998e7ad race condition SizeLimitedDistributedMap

This branch includes the following new commits:

     new c1ae998e7ad race condition SizeLimitedDistributedMap

The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails.  The revisions
listed as "add" were already present in the repository and have only
been added to this reference.

[solr] 01/01: race condition SizeLimitedDistributedMap

[no...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

noble pushed a commit to branch jira/solr16175
in repository https://gitbox.apache.org/repos/asf/solr.git

commit c1ae998e7ad3650229449ff7b2a55ef222ec8b8c
Author: Noble Paul <no...@gmail.com>
AuthorDate: Wed May 4 13:55:24 2022 +1000

    race condition SizeLimitedDistributedMap
---
 .../solr/cloud/SizeLimitedDistributedMap.java      | 11 ++--
 .../solr/cloud/TestSizeLimitedDistributedMap.java  | 61 ++++++++++++++++++++--
 2 files changed, 65 insertions(+), 7 deletions(-)

diff --git a/solr/core/src/java/org/apache/solr/cloud/SizeLimitedDistributedMap.java b/solr/core/src/java/org/apache/solr/cloud/SizeLimitedDistributedMap.java
index db57ea4980b..2e021a54c53 100644
--- a/solr/core/src/java/org/apache/solr/cloud/SizeLimitedDistributedMap.java
+++ b/solr/core/src/java/org/apache/solr/cloud/SizeLimitedDistributedMap.java
@@ -91,9 +91,14 @@ public class SizeLimitedDistributedMap extends DistributedMap {
       for (String child : children) {
         Long id = childToModificationZxid.get(child);
         if (id != null && id <= topElementMzxId) {
-          zookeeper.delete(dir + "/" + child, -1, true);
-          if (onOverflowObserver != null)
-            onOverflowObserver.onChildDelete(child.substring(PREFIX.length()));
+          try {
+            zookeeper.delete(dir + "/" + child, -1, true);
+            if (onOverflowObserver != null) {
+              onOverflowObserver.onChildDelete(child.substring(PREFIX.length()));
+            }
+          } catch (KeeperException.NoNodeException e) {
+            //Ignore. It's already deleted by another thread
+          }
         }
       }
     }
diff --git a/solr/core/src/test/org/apache/solr/cloud/TestSizeLimitedDistributedMap.java b/solr/core/src/test/org/apache/solr/cloud/TestSizeLimitedDistributedMap.java
index a086656d13c..5b8f3c3d5cb 100644
--- a/solr/core/src/test/org/apache/solr/cloud/TestSizeLimitedDistributedMap.java
+++ b/solr/core/src/test/org/apache/solr/cloud/TestSizeLimitedDistributedMap.java
@@ -17,10 +17,13 @@
 
 package org.apache.solr.cloud;
 
-import java.util.HashSet;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Set;
+import java.util.*;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.solr.common.Callable;
 import org.apache.solr.common.cloud.SolrZkClient;
 
 public class TestSizeLimitedDistributedMap extends TestDistributedMap {
@@ -67,6 +70,56 @@ public class TestSizeLimitedDistributedMap extends TestDistributedMap {
     }
   }
 
+  public void testConcurrentCleanup() throws Exception {
+    final Set<String> expectedKeys = new HashSet<>();
+    final List<String> deletedItems = new LinkedList<>();
+    int numResponsesToStore = TEST_NIGHTLY ? Overseer.NUM_RESPONSES_TO_STORE : 100;
+
+    try (SolrZkClient zkClient = new SolrZkClient(zkServer.getZkHost(), 10000)) {
+      String path = getAndMakeInitialPath(zkClient);
+      final DistributedMap map = new SizeLimitedDistributedMap(zkClient, path, numResponsesToStore, deletedItems::add);
+      //fill the map to limit first
+      for (int i = 0; i < numResponsesToStore; i++) {
+        map.put("xyz_" + i, new byte[0]);
+      }
+
+      // add more elements concurrently to trigger cleanup
+      final int THREAD_COUNT = Math.min(100, numResponsesToStore);
+      List<Callable> callables = new ArrayList<>();
+      for (int i = 0; i < THREAD_COUNT; i++) {
+        final String key = "xyz_" + (numResponsesToStore + 1);
+        expectedKeys.add(key);
+        callables.add(new Callable() {
+          @Override
+          public void call(Object o) {
+            try {
+              map.put(key, new byte[0]);
+            } catch (Exception e) {
+              throw new RuntimeException(e);
+            }
+          }
+        });
+      }
+      ExecutorService executorService = Executors.newFixedThreadPool(THREAD_COUNT);
+      List<Future<Object>> futures = new ArrayList<>();
+      for (Callable callable : callables) {
+        futures.add(executorService.submit(callable));
+      }
+      try {
+        for (Future<Object> future : futures) {
+          future.get(); //none of them should throw exception
+        }
+        for (String expectedKey : expectedKeys) {
+          assertTrue(map.contains(expectedKey));
+        }
+        //there's no guarantees on exactly how many elements will be removed, but it should at least NOT throw exception
+        assertFalse(deletedItems.isEmpty());
+      } finally {
+        executorService.shutdown();
+        executorService.awaitTermination(10, TimeUnit.SECONDS);
+      }
+    }
+  }
   protected DistributedMap createMap(SolrZkClient zkClient, String path) {
     return new SizeLimitedDistributedMap(zkClient, path, Overseer.NUM_RESPONSES_TO_STORE, null);
   }