Improve data store handling with thread safety and resource management enhancements (#2954)

* Improve data store handling with thread safety and resource management enhancements

This commit addresses several improvements in the data store processing implementation:

**Thread Safety Improvements:**
- DataStoreFactory: Changed dataStoreMap from LinkedHashMap to ConcurrentHashMap for thread-safe concurrent access
- DataStoreFactory: Made dataStoreNames and lastLoadedTime volatile to ensure visibility across threads
- DataStoreFactory: Made getDataStoreNames() synchronized to prevent race conditions during cache refresh
- AbstractDataStore: Made alive field volatile to ensure proper visibility in multi-threaded scenarios
- FileListIndexUpdateCallbackImpl: Changed deleteUrlList from ArrayList to CopyOnWriteArrayList for thread-safe access from executor threads

**Resource Handling:**
- DataStoreFactory: Added null check for jarFiles array to prevent NullPointerException
- DataStoreFactory: Added existence check for XML configuration files before attempting to read them, reducing unnecessary exception handling

**Code Quality:**
- AbstractDataStore: Moved static logger declaration before instance fields following Java best practices
- FileListIndexUpdateCallbackImpl: Fixed incorrect StringUtil import (was using Apache POI's StringUtil instead of Fess's)

These improvements enhance the reliability and thread safety of the data store processing layer,
particularly important for concurrent crawling operations.

* Optimize deleteUrlList by replacing CopyOnWriteArrayList with ArrayList

Changed deleteUrlList from CopyOnWriteArrayList to ArrayList since all access
is already protected by synchronized(indexUpdateCallback) blocks. CopyOnWriteArrayList
creates a copy on every write operation which is unnecessary overhead when synchronization
is already in place.

Changes:
- Replaced CopyOnWriteArrayList with ArrayList for deleteUrlList
- Added synchronized block in commit() method for consistency
- Updated javadoc to clarify synchronization strategy

This improves performance for batch delete operations by eliminating unnecessary
array copying on each URL addition.

* Fix ProcessHelperTest timing issues by using long-running processes

Fixed two failing tests that were using 'echo' commands which complete
almost instantly, making it impossible to verify process running state:

1. test_startProcess_replaceExistingProcess:
   - Changed from 'echo' to 'sleep 10' commands
   - Added proper wait time before checking process state
   - Added verification that only one process exists for the session

2. test_destroyProcess_withRunningProcess:
   - Changed from 'echo' to 'sleep 10' command
   - Simplified polling logic - just wait 100ms for process to start
   - Updated exit code assertion (forcibly destroyed processes may have non-zero exit codes)

These changes ensure the tests can reliably verify that processes are running
before attempting to check their state or destroy them.

* Add comprehensive thread safety tests for data store improvements

Added extensive test coverage for the thread safety improvements made to
the data store handling layer.

DataStoreFactoryTest additions:
- test_add_concurrentAccess: Verifies ConcurrentHashMap handles concurrent additions
- test_getDataStore_concurrentWithAdd: Tests concurrent read/write operations
- test_getDataStoreNames_concurrentAccess: Verifies synchronized method correctness
- test_volatileFields_visibility: Ensures volatile fields are visible across threads
- test_cacheRefresh_withConcurrentReads: Validates cache refresh with concurrent access
- test_nullSafety_concurrentAccess: Tests null handling under concurrent load

AbstractDataStoreTest additions:
- test_aliveField_volatileVisibility: Verifies volatile alive field visibility
- test_stop_volatileVisibility: Tests stop() method sets alive and is visible
- test_stop_concurrentAccess: Validates concurrent calls to stop()
- test_aliveField_concurrentReadWrite: Tests concurrent read/write of alive field
- test_getName_concurrentAccess: Verifies getName() is thread-safe

FileListIndexUpdateCallbackImplTest additions:
- test_deleteUrlList_synchronizedAccess: Validates ArrayList with synchronized blocks
- test_deleteUrlList_concurrentReads: Tests concurrent reads from deleteUrlList
- test_deleteUrlList_clearOperation: Verifies clear() operation thread safety
- test_deleteUrlList_iteration: Tests iteration with proper synchronization
- test_deleteUrlList_isEmptyCheck: Validates isEmpty() check under concurrent access

These tests verify:
1. ConcurrentHashMap correctly handles concurrent operations in DataStoreFactory
2. Volatile fields ensure proper visibility across threads
3. Synchronized methods prevent race conditions during cache refresh
4. ArrayList works correctly when all access is properly synchronized
5. No ConcurrentModificationException occurs in any concurrent scenario

All tests use minimal or no sleep to maintain fast test execution time.

* Revert dataStoreMap from ConcurrentHashMap to LinkedHashMap

Based on feedback that dataStoreMap is not accessed by multiple threads,
reverted the ConcurrentHashMap change back to LinkedHashMap.

Changes:
- Reverted dataStoreMap from ConcurrentHashMap to LinkedHashMap
- Removed ConcurrentHashMap import
- Removed dataStoreMap-related concurrency tests:
  * test_add_concurrentAccess
  * test_getDataStore_concurrentWithAdd
  * test_nullSafety_concurrentAccess

Kept volatile fields and synchronized getDataStoreNames():
- lastLoadedTime and dataStoreNames remain volatile (for cache visibility)
- getDataStoreNames() remains synchronized (for cache refresh safety)
- Related tests for getDataStoreNames() concurrency are maintained

The dataStoreMap is accessed in single-threaded context during application
initialization and data store registration.

---------

Co-authored-by: Claude <[email protected]>

Author: marevol

src/main/java/org/codelibs/fess/ds/AbstractDataStore.java

@@ -43,13 +43,6 @@
  */
 public abstract class AbstractDataStore implements DataStore {
 
-    /**
-     * Default constructor.
-     */
-    public AbstractDataStore() {
-        // nothing
-    }
-
     private static final Logger logger = LogManager.getLogger(AbstractDataStore.class);
 
     /**
@@ -64,8 +57,16 @@ public AbstractDataStore() {
 
     /**
      * The flag to check if the data store is alive.
+     * Volatile to ensure visibility across threads.
      */
-    protected boolean alive = true;
+    protected volatile boolean alive = true;
+
+    /**
+     * Default constructor.
+     */
+    public AbstractDataStore() {
+        // nothing
+    }
 
     /**
      * Register this data store.

src/main/java/org/codelibs/fess/ds/DataStoreFactory.java

@@ -71,13 +71,14 @@ public class DataStoreFactory {
      * Cached array of available data store names discovered from plugin JAR files.
      * This cache is refreshed periodically based on the lastLoadedTime.
      */
-    protected String[] dataStoreNames = StringUtil.EMPTY_STRINGS;
+    protected volatile String[] dataStoreNames = StringUtil.EMPTY_STRINGS;
 
     /**
      * Timestamp of the last time data store names were loaded from plugin files.
      * Used to implement a time-based cache refresh mechanism.
+     * Volatile to ensure visibility across threads.
      */
-    protected long lastLoadedTime = 0;
+    protected volatile long lastLoadedTime = 0;
 
     /**
      * Creates a new instance of DataStoreFactory.
@@ -130,7 +131,7 @@ public DataStore getDataStore(final String name) {
      *
      * @return array of data store names sorted alphabetically, never null
      */
-    public String[] getDataStoreNames() {
+    public synchronized String[] getDataStoreNames() {
         final long now = ComponentUtil.getSystemHelper().getCurrentTimeAsLong();
         if (now - lastLoadedTime > 60000L) {
             final List<String> nameList = loadDataStoreNameList();
@@ -154,9 +155,18 @@ public String[] getDataStoreNames() {
     protected List<String> loadDataStoreNameList() {
         final Set<String> nameSet = new HashSet<>();
         final File[] jarFiles = ResourceUtil.getPluginJarFiles(PluginHelper.ArtifactType.DATA_STORE.getId());
+        if (jarFiles == null) {
+            return nameSet.stream().sorted().collect(Collectors.toList());
+        }
         for (final File jarFile : jarFiles) {
             try (FileSystem fs = FileSystems.newFileSystem(jarFile.toPath(), ClassLoader.getSystemClassLoader())) {
                 final Path xmlPath = fs.getPath("fess_ds++.xml");
+                if (!Files.exists(xmlPath)) {
+                    if (logger.isDebugEnabled()) {
+                        logger.debug("Configuration file not found in {}", jarFile.getAbsolutePath());
+                    }
+                    continue;
+                }
                 try (InputStream is = Files.newInputStream(xmlPath)) {
                     final DocumentBuilderFactory factory = DocumentBuilderFactory.newInstance();
                     factory.setFeature(org.codelibs.fess.crawler.Constants.FEATURE_SECURE_PROCESSING, true);

src/main/java/org/codelibs/fess/ds/callback/FileListIndexUpdateCallbackImpl.java

@@ -34,7 +34,7 @@
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
-import org.apache.poi.util.StringUtil;
+import org.codelibs.core.lang.StringUtil;
 import org.codelibs.fess.Constants;
 import org.codelibs.fess.crawler.builder.RequestDataBuilder;
 import org.codelibs.fess.crawler.client.CrawlerClient;
@@ -89,8 +89,11 @@ public class FileListIndexUpdateCallbackImpl implements IndexUpdateCallback {
     /** Factory for creating crawler clients to handle different URL schemes. */
     protected CrawlerClientFactory crawlerClientFactory;
 
-    /** List of URLs to be deleted, cached for batch processing. */
-    protected List<String> deleteUrlList = new ArrayList<>(100);
+    /**
+     * List of URLs to be deleted, cached for batch processing.
+     * All access is synchronized via indexUpdateCallback lock.
+     */
+    protected List<String> deleteUrlList = new ArrayList<>();
 
     /** Maximum size of the delete URL cache before batch deletion is triggered. */
     protected int maxDeleteDocumentCacheSize;
@@ -585,8 +588,10 @@ public void commit() {
             executor.shutdownNow();
         }
 
-        if (!deleteUrlList.isEmpty()) {
-            deleteDocuments();
+        synchronized (indexUpdateCallback) {
+            if (!deleteUrlList.isEmpty()) {
+                deleteDocuments();
+            }
         }
         indexUpdateCallback.commit();
     }

src/test/java/org/codelibs/fess/ds/AbstractDataStoreTest.java

@@ -108,4 +108,235 @@ public void test_convertValue() {
         value = " ";
         assertNull(dataStore.convertValue(Constants.DEFAULT_SCRIPT, value, paramMap));
     }
+
+    // ========== Thread Safety Tests ==========
+
+    /**
+     * Test that the volatile alive field is visible across threads.
+     * One thread sets alive to false, other threads should see the change immediately.
+     */
+    public void test_aliveField_volatileVisibility() throws Exception {
+        // Ensure alive starts as true
+        assertTrue(dataStore.alive);
+
+        final int readerThreadCount = 10;
+        final Thread[] readerThreads = new Thread[readerThreadCount];
+        final boolean[][] observations = new boolean[readerThreadCount][100];
+
+        // Start reader threads that continuously check alive field
+        for (int i = 0; i < readerThreadCount; i++) {
+            final int threadIndex = i;
+            readerThreads[i] = new Thread(() -> {
+                for (int j = 0; j < 100; j++) {
+                    observations[threadIndex][j] = dataStore.alive;
+                    // Small yield to allow context switching
+                    Thread.yield();
+                }
+            });
+            readerThreads[i].start();
+        }
+
+        // Writer thread sets alive to false
+        Thread writerThread = new Thread(() -> {
+            dataStore.alive = false;
+        });
+        writerThread.start();
+        writerThread.join();
+
+        // Wait for all reader threads to complete
+        for (Thread thread : readerThreads) {
+            thread.join();
+        }
+
+        // Verify that alive was changed to false
+        assertFalse("alive should be false after writer thread", dataStore.alive);
+
+        // At least some observations from reader threads should have seen false
+        // due to volatile ensuring visibility
+        int falseCount = 0;
+        for (int i = 0; i < readerThreadCount; i++) {
+            for (int j = 0; j < 100; j++) {
+                if (!observations[i][j]) {
+                    falseCount++;
+                }
+            }
+        }
+        assertTrue("Some threads should have observed alive=false", falseCount > 0);
+    }
+
+    /**
+     * Test stop() method sets alive to false and is visible to other threads.
+     */
+    public void test_stop_volatileVisibility() throws Exception {
+        assertTrue(dataStore.alive);
+
+        final boolean[] observedValues = new boolean[10];
+        final Thread[] threads = new Thread[10];
+
+        // Call stop() in main thread
+        dataStore.stop();
+
+        // Multiple threads read the alive field
+        for (int i = 0; i < 10; i++) {
+            final int index = i;
+            threads[i] = new Thread(() -> {
+                observedValues[index] = dataStore.alive;
+            });
+        }
+
+        for (Thread thread : threads) {
+            thread.start();
+        }
+
+        for (Thread thread : threads) {
+            thread.join();
+        }
+
+        // All threads should observe alive=false due to volatile
+        for (int i = 0; i < 10; i++) {
+            assertFalse("Thread " + i + " should see alive=false", observedValues[i]);
+        }
+    }
+
+    /**
+     * Test concurrent access to stop() method.
+     * Multiple threads call stop() simultaneously - should be safe.
+     */
+    public void test_stop_concurrentAccess() throws Exception {
+        final int threadCount = 10;
+        final Thread[] threads = new Thread[threadCount];
+        final Exception[] exceptions = new Exception[threadCount];
+
+        for (int i = 0; i < threadCount; i++) {
+            final int index = i;
+            threads[i] = new Thread(() -> {
+                try {
+                    dataStore.stop();
+                } catch (Exception e) {
+                    exceptions[index] = e;
+                }
+            });
+        }
+
+        // Start all threads
+        for (Thread thread : threads) {
+            thread.start();
+        }
+
+        // Wait for all threads to complete
+        for (Thread thread : threads) {
+            thread.join();
+        }
+
+        // Verify no exceptions occurred
+        for (int i = 0; i < threadCount; i++) {
+            assertNull("Thread " + i + " threw exception", exceptions[i]);
+        }
+
+        // Verify alive is false
+        assertFalse("alive should be false after all stop() calls", dataStore.alive);
+    }
+
+    /**
+     * Test that multiple threads can safely read alive field while one writes.
+     */
+    public void test_aliveField_concurrentReadWrite() throws Exception {
+        dataStore.alive = true;
+
+        final int readerCount = 5;
+        final int iterations = 1000;
+        final Thread[] readers = new Thread[readerCount];
+        final Exception[] exceptions = new Exception[readerCount + 1];
+        final int[] trueCount = new int[readerCount];
+        final int[] falseCount = new int[readerCount];
+
+        // Start reader threads
+        for (int i = 0; i < readerCount; i++) {
+            final int index = i;
+            readers[i] = new Thread(() -> {
+                try {
+                    for (int j = 0; j < iterations; j++) {
+                        if (dataStore.alive) {
+                            trueCount[index]++;
+                        } else {
+                            falseCount[index]++;
+                        }
+                    }
+                } catch (Exception e) {
+                    exceptions[index] = e;
+                }
+            });
+        }
+
+        // Start writer thread that toggles alive
+        Thread writer = new Thread(() -> {
+            try {
+                for (int i = 0; i < 100; i++) {
+                    dataStore.alive = !dataStore.alive;
+                    Thread.yield();
+                }
+            } catch (Exception e) {
+                exceptions[readerCount] = e;
+            }
+        });
+
+        // Start all threads
+        for (Thread reader : readers) {
+            reader.start();
+        }
+        writer.start();
+
+        // Wait for completion
+        for (Thread reader : readers) {
+            reader.join();
+        }
+        writer.join();
+
+        // Verify no exceptions
+        for (int i = 0; i <= readerCount; i++) {
+            assertNull("Thread " + i + " threw exception", exceptions[i]);
+        }
+
+        // Verify all readers completed all iterations
+        for (int i = 0; i < readerCount; i++) {
+            assertEquals("Reader " + i + " should complete all iterations", iterations, trueCount[i] + falseCount[i]);
+        }
+    }
+
+    /**
+     * Test getName() method can be called concurrently without issues.
+     */
+    public void test_getName_concurrentAccess() throws Exception {
+        final int threadCount = 10;
+        final Thread[] threads = new Thread[threadCount];
+        final String[] results = new String[threadCount];
+        final Exception[] exceptions = new Exception[threadCount];
+
+        for (int i = 0; i < threadCount; i++) {
+            final int index = i;
+            threads[i] = new Thread(() -> {
+                try {
+                    for (int j = 0; j < 100; j++) {
+                        results[index] = dataStore.getName();
+                    }
+                } catch (Exception e) {
+                    exceptions[index] = e;
+                }
+            });
+        }
+
+        for (Thread thread : threads) {
+            thread.start();
+        }
+
+        for (Thread thread : threads) {
+            thread.join();
+        }
+
+        // Verify no exceptions
+        for (int i = 0; i < threadCount; i++) {
+            assertNull("Thread " + i + " threw exception", exceptions[i]);
+            assertEquals("Thread " + i + " got wrong name", "Test", results[i]);
+        }
+    }
 }