fix failing IT tests

engine/src/main/java/com/arcadedb/database/LocalDatabase.java

@@ -80,14 +80,33 @@
 import com.arcadedb.utility.MultiIterator;
 import com.arcadedb.utility.RWLockContext;
 
-import java.io.*;
-import java.nio.channels.*;
-import java.nio.file.*;
-import java.util.*;
-import java.util.concurrent.*;
-import java.util.concurrent.atomic.*;
-import java.util.concurrent.locks.*;
-import java.util.logging.*;
+  import java.io.File;
+import java.io.IOException;
+import java.io.RandomAccessFile;
+import java.nio.channels.ClosedChannelException;
+import java.nio.channels.FileChannel;
+import java.nio.channels.FileLock;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Random;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.logging.Level;
 
 /**
  * Local implementation of {@link Database}. It is based on files opened on the local file system.
@@ -99,9 +118,14 @@
 public class LocalDatabase extends RWLockContext implements DatabaseInternal {
   public static final  int                                       EDGE_LIST_INITIAL_CHUNK_SIZE         = 64;
   public static final  int                                       MAX_RECOMMENDED_EDGE_LIST_CHUNK_SIZE = 8192;
-  private static final Set<String>                               SUPPORTED_FILE_EXT                   = Set.of(Dictionary.DICT_EXT,
-      LocalBucket.BUCKET_EXT, LSMTreeIndexMutable.NOTUNIQUE_INDEX_EXT, LSMTreeIndexMutable.UNIQUE_INDEX_EXT,
-      LSMTreeIndexCompacted.NOTUNIQUE_INDEX_EXT, LSMTreeIndexCompacted.UNIQUE_INDEX_EXT, HnswVectorIndex.FILE_EXT);
+  private static final Set<String>                               SUPPORTED_FILE_EXT                   = Set.of(
+      Dictionary.DICT_EXT,
+      LocalBucket.BUCKET_EXT,
+      LSMTreeIndexMutable.NOTUNIQUE_INDEX_EXT,
+      LSMTreeIndexMutable.UNIQUE_INDEX_EXT,
+      LSMTreeIndexCompacted.NOTUNIQUE_INDEX_EXT,
+      LSMTreeIndexCompacted.UNIQUE_INDEX_EXT,
+      HnswVectorIndex.FILE_EXT);
   public final         AtomicLong                                indexCompactions                     = new AtomicLong();
   protected final      String                                    name;
   protected final      ComponentFile.MODE                        mode;
@@ -538,10 +562,10 @@ public void scanBucket(final String bucketName, final RecordCallback callback, f
   public Iterator<Record> iterateType(final String typeName, final boolean polymorphic) {
     stats.iterateType.incrementAndGet();
 
-    return (Iterator<Record>) executeInReadLock(() -> {
+    return executeInReadLock(() -> {
       checkDatabaseIsOpen();
-      final DocumentType type = schema.getType(typeName);
-      final MultiIterator iter = new MultiIterator();
+      var type = schema.getType(typeName);
+      var iter = new MultiIterator<Record>();
 
       // SET THE PROFILED LIMITS IF ANY
       iter.setLimit(getResultSetLimit());
@@ -1637,12 +1661,7 @@ record = events.onAfterRead(record);
     if (record instanceof Document document) {
       final DocumentType type = document.getType();
       if (type != null) {
-//        System.out.println("invokeAfterReadEvents for type = pre");
-
-        Record record1 = ((RecordEventsRegistry) type.getEvents()).onAfterRead(record);
-//        System.out.println("invokeAfterReadEvents for type = after");
-
-        return record1;
+        return ((RecordEventsRegistry) type.getEvents()).onAfterRead(record);
       }
     }
     return record;

engine/src/main/java/com/arcadedb/database/LocalTransactionExplicitLock.java

@@ -20,6 +20,7 @@
 package com.arcadedb.database;
 
 import com.arcadedb.engine.Bucket;
+import com.arcadedb.index.IndexInternal;
 import com.arcadedb.log.LogManager;
 import com.arcadedb.schema.DocumentType;
 
@@ -61,7 +62,7 @@ public LocalTransactionExplicitLock type(final String typeName) {
     // Lock all indexes for this type
     filesToLock.addAll(type.getAllIndexes(true).stream()
         .flatMap(i -> Arrays.stream(i.getIndexesOnBuckets()))
-        .map(b -> b.getFileId())
+        .map(IndexInternal::getFileId)
         .toList());
 
     // Lock all currently involved buckets for this type

engine/src/main/java/com/arcadedb/schema/LocalSchema.java

@@ -59,12 +59,26 @@
 import com.arcadedb.serializer.json.JSONObject;
 import com.arcadedb.utility.FileUtils;
 
-import java.io.*;
-import java.time.*;
-import java.util.*;
-import java.util.concurrent.*;
-import java.util.concurrent.atomic.*;
-import java.util.logging.*;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.time.ZoneId;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TimeZone;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.logging.Level;
 
 /**
  * Local implementation of the database schema.
@@ -1199,6 +1213,7 @@ public synchronized void saveConfiguration() {
     }
 
     try {
+      LogManager.instance().log(this, Level.INFO, "Saving schema configuration to file - versionSerial = %s ", versionSerial);
       versionSerial.incrementAndGet();
 
       update(toJSON());

engine/src/test/java/com/arcadedb/ACIDTransactionTest.java

@@ -36,6 +36,7 @@
 import com.arcadedb.schema.Schema;
 import com.arcadedb.schema.Type;
 import com.arcadedb.schema.VertexType;
+import org.awaitility.Awaitility;
 import org.junit.jupiter.api.Tag;
 import org.junit.jupiter.api.Test;
 
@@ -46,6 +47,12 @@
 import java.util.Iterator;
 import java.util.List;
 import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.logging.Level;
@@ -78,13 +85,6 @@ public void testAsyncTX() {
 
       db.async().waitCompletion();
 
-      try {
-        Thread.sleep(500);
-      } catch (final InterruptedException e) {
-        Thread.currentThread().interrupt();
-        // IGNORE IT
-      }
-
     } catch (final TransactionException e) {
       assertThat(e.getCause() instanceof IOException).isTrue();
     }
@@ -240,13 +240,6 @@ public Void call() throws IOException {
 
       db.async().waitCompletion();
 
-      try {
-        Thread.sleep(500);
-      } catch (final InterruptedException e) {
-        Thread.currentThread().interrupt();
-        // IGNORE IT
-      }
-
       assertThat(errors.get()).isEqualTo(1);
 
     } catch (final TransactionException e) {
@@ -486,10 +479,12 @@ public void testExceptionInsideTransaction() {
 
     final int CONCURRENT_THREADS = 4;
 
-    // SPAWN ALL THE THREADS
-    final Thread[] threads = new Thread[CONCURRENT_THREADS];
+    // SPAWN ALL THE THREADS USING EXECUTORSERVICE
+    final ExecutorService executorService = Executors.newFixedThreadPool(CONCURRENT_THREADS);
+    final List<Future<?>> futures = new ArrayList<>();
+
     for (int i = 0; i < CONCURRENT_THREADS; i++) {
-      threads[i] = new Thread(() -> {
+      Future<?> future = executorService.submit(() -> {
         for (int k = 0; k < TOT; ++k) {
           final int id = k;
 
@@ -512,16 +507,33 @@ public void testExceptionInsideTransaction() {
         }
 
       });
-      threads[i].start();
+      futures.add(future);
     }
 
     // WAIT FOR ALL THE THREADS
-    for (int i = 0; i < CONCURRENT_THREADS; i++)
+    for (Future<?> future : futures) {
       try {
-        threads[i].join();
+        future.get(120, TimeUnit.SECONDS);
       } catch (InterruptedException e) {
-        // IGNORE IT
+        Thread.currentThread().interrupt();
+        LogManager.instance().log(this, Level.WARNING, "Thread interrupted while waiting for future", e);
+      } catch (ExecutionException e) {
+        LogManager.instance().log(this, Level.WARNING, "Execution exception in future", e);
+      } catch (TimeoutException e) {
+        LogManager.instance().log(this, Level.SEVERE, "Future timed out after 120 seconds", e);
+        future.cancel(true);
       }
+    }
+
+    executorService.shutdown();
+    try {
+      if (!executorService.awaitTermination(60, TimeUnit.SECONDS)) {
+        executorService.shutdownNow();
+      }
+    } catch (InterruptedException e) {
+      executorService.shutdownNow();
+      Thread.currentThread().interrupt();
+    }
 
     assertThat(database.countType("Node", true)).isEqualTo(1);