Sync with main

Signed-off-by: Artem Ananev <[email protected]>

Author: artemananiev

.github/dependabot.yml

@@ -1,5 +1,11 @@
+# SPDX-License-Identifier: Apache-2.0
 version: 2
 updates:
+  - package-ecosystem: "gradle"
+    directory: "/" # for 'plugins { }' in 'settings.gradle.kts'
+    schedule:
+      interval: "daily"
+    open-pull-requests-limit: 10
   - package-ecosystem: "gradle"
     directory: "/hiero-dependency-versions"
     schedule:

hiero-dependency-versions/build.gradle.kts

@@ -28,7 +28,7 @@ val junit5 = "5.10.3!!" // no updates beyond 5.10.3 until #17125 is resolved
 val log4j = "2.24.3"
 val mockito = "5.17.0"
 val pbj = "0.11.6" // ATTENTION: keep in sync with plugin version in 'hapi/hapi/build.gradle.kts'
-val protobuf = "4.30.2"
+val protobuf = "4.31.0"
 val testContainers = "1.21.0"
 val tuweni = "2.4.2"
 val webcompare = "2.1.8"
@@ -50,7 +50,7 @@ dependencies.constraints {
         because("com.fasterxml.jackson.dataformat.yaml")
     }
     api("com.github.ben-manes.caffeine:caffeine:3.2.0") { because("com.github.benmanes.caffeine") }
-    api("com.github.docker-java:docker-java-api:3.5.0") { because("com.github.dockerjava.api") }
+    api("com.github.docker-java:docker-java-api:3.5.1") { because("com.github.dockerjava.api") }
     api("com.github.spotbugs:spotbugs-annotations:4.9.3") {
         because("com.github.spotbugs.annotations")
     }
@@ -111,7 +111,7 @@ dependencies.constraints {
         because("org.hyperledger.besu.nativelib.secp256k1")
     }
     api("org.jetbrains:annotations:26.0.2") { because("org.jetbrains.annotations") }
-    api("org.json:json:20250107") { because("org.json") }
+    api("org.json:json:20250517") { because("org.json") }
     api("org.junit.jupiter:junit-jupiter-api:$junit5") { because("org.junit.jupiter.api") }
     api("org.junit.jupiter:junit-jupiter-engine:$junit5") { because("org.junit.jupiter.engine") }
     api("org.junit-pioneer:junit-pioneer:2.3.0") { because("org.junitpioneer") }

platform-sdk/swirlds-merkledb/src/main/java/com/swirlds/merkledb/MerkleDbCompactionCoordinator.java

@@ -13,16 +13,13 @@
 import edu.umd.cs.findbugs.annotations.NonNull;
 import java.io.IOException;
 import java.nio.channels.ClosedByInterruptException;
+import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.Callable;
-import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Future;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicInteger;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
@@ -75,21 +72,15 @@ static synchronized ExecutorService getCompactionExecutor(final @NonNull MerkleD
         return compactionExecutor;
     }
 
-    private final AtomicBoolean compactionEnabled = new AtomicBoolean();
+    // Synchronized on this
+    private boolean compactionEnabled = false;
 
-    // A map of compactor futures by task names
-    final Map<String, Future<Boolean>> futuresByName = new ConcurrentHashMap<>(16);
-
-    // A map of compactors by task names
-    final Map<String, DataFileCompactor> compactorsByName = new ConcurrentHashMap<>(16);
+    // A map of compactors by task names. Synchronized on this
+    final Map<String, DataFileCompactor> compactorsByName = new HashMap<>(16);
 
     @NonNull
     private final MerkleDbConfig merkleDbConfig;
 
-    // Number of compaction tasks currently running. Checked during shutdown to make sure all
-    // tasks are stopped
-    private final AtomicInteger tasksRunning = new AtomicInteger(0);
-
     /**
      * Creates a new instance of {@link MerkleDbCompactionCoordinator}.
      * @param tableName the name of the table
@@ -104,8 +95,8 @@ public MerkleDbCompactionCoordinator(@NonNull String tableName, @NonNull MerkleD
     /**
      * Enables background compaction.
      */
-    void enableBackgroundCompaction() {
-        compactionEnabled.set(true);
+    synchronized void enableBackgroundCompaction() {
+        compactionEnabled = true;
     }
 
     /**
@@ -114,7 +105,7 @@ void enableBackgroundCompaction() {
      * critical for snapshots (e.g. update an index), it will be stopped until {@link
      * #resumeCompaction()}} is called.
      */
-    public void pauseCompaction() throws IOException {
+    synchronized void pauseCompaction() throws IOException {
         for (final DataFileCompactor compactor : compactorsByName.values()) {
             compactor.pauseCompaction();
         }
@@ -123,7 +114,7 @@ public void pauseCompaction() throws IOException {
     /**
      * Resumes previously stopped data file collection compaction.
      */
-    public void resumeCompaction() throws IOException {
+    synchronized void resumeCompaction() throws IOException {
         for (final DataFileCompactor compactor : compactorsByName.values()) {
             compactor.resumeCompaction();
         }
@@ -133,26 +124,24 @@ public void resumeCompaction() throws IOException {
      * Stops all compactions in progress and disables background compaction. All subsequent calls to
      * compacting methods will be ignored until {@link #enableBackgroundCompaction()} is called.
      */
-    void stopAndDisableBackgroundCompaction() {
-        compactionEnabled.set(false);
+    synchronized void stopAndDisableBackgroundCompaction() {
+        // Make sure no new compaction tasks are scheduled
+        compactionEnabled = false;
         // Interrupt all running compaction tasks, if any
-        for (final Future<Boolean> futureEntry : futuresByName.values()) {
-            futureEntry.cancel(true);
+        for (final DataFileCompactor compactor : compactorsByName.values()) {
+            compactor.interruptCompaction();
         }
-        futuresByName.clear();
-        compactorsByName.clear();
         // Wait till all the tasks are stopped
-        final long now = System.currentTimeMillis();
         try {
-            while ((tasksRunning.get() != 0) && (System.currentTimeMillis() - now < SHUTDOWN_TIMEOUT_MILLIS)) {
-                Thread.sleep(1);
+            while (!compactorsByName.isEmpty()) {
+                wait(SHUTDOWN_TIMEOUT_MILLIS);
             }
         } catch (final InterruptedException e) {
             logger.warn(MERKLE_DB.getMarker(), "Interrupted while waiting for compaction tasks to complete", e);
         }
         // If some tasks are still running, there is nothing else to than to log it
-        if (tasksRunning.get() != 0) {
-            logger.error(MERKLE_DB.getMarker(), "Failed to stop all compactions tasks");
+        if (!compactorsByName.isEmpty()) {
+            logger.warn(MERKLE_DB.getMarker(), "Timed out waiting to stop all compactions tasks");
         }
     }
 
@@ -163,19 +152,18 @@ void stopAndDisableBackgroundCompaction() {
      * @param key Compaction task name
      * @param compactor Compactor to run
      */
-    public void compactIfNotRunningYet(final String key, final DataFileCompactor compactor) {
-        if (!compactionEnabled.get()) {
+    public synchronized void compactIfNotRunningYet(final String key, final DataFileCompactor compactor) {
+        if (!compactionEnabled) {
             return;
         }
         if (isCompactionRunning(key)) {
             logger.info(MERKLE_DB.getMarker(), "Compaction for {} is already in progress", key);
             return;
         }
-        assert !compactorsByName.containsKey(key);
         compactorsByName.put(key, compactor);
         final ExecutorService executor = getCompactionExecutor(merkleDbConfig);
         final CompactionTask task = new CompactionTask(key, compactor);
-        futuresByName.put(key, executor.submit(task));
+        executor.submit(task);
     }
 
     /**
@@ -184,13 +172,12 @@ public void compactIfNotRunningYet(final String key, final DataFileCompactor com
      * @param key Compactor name
      * @return {@code true} if compaction with this name is currently running, {@code false} otherwise
      */
-    public boolean isCompactionRunning(final String key) {
-        final Future<?> future = futuresByName.get(key);
-        return (future != null) && !future.isDone();
+    synchronized boolean isCompactionRunning(final String key) {
+        return compactorsByName.containsKey(key);
     }
 
-    boolean isCompactionEnabled() {
-        return compactionEnabled.get();
+    synchronized boolean isCompactionEnabled() {
+        return compactionEnabled;
     }
 
     /**
@@ -211,7 +198,6 @@ public CompactionTask(@NonNull String id, @NonNull DataFileCompactor compactor)
 
         @Override
         public Boolean call() {
-            tasksRunning.incrementAndGet();
             try {
                 return compactor.compact();
             } catch (final InterruptedException | ClosedByInterruptException e) {
@@ -221,9 +207,10 @@ public Boolean call() {
                 // will stop all future merges from happening
                 logger.error(EXCEPTION.getMarker(), "[{}] Compaction failed", id, e);
             } finally {
-                compactorsByName.remove(id);
-                futuresByName.remove(id);
-                tasksRunning.decrementAndGet();
+                synchronized (MerkleDbCompactionCoordinator.this) {
+                    compactorsByName.remove(id);
+                    MerkleDbCompactionCoordinator.this.notifyAll();
+                }
             }
             return false;
         }

platform-sdk/swirlds-merkledb/src/main/java/com/swirlds/merkledb/collections/AbstractLongList.java

@@ -12,6 +12,7 @@
 import com.swirlds.merkledb.config.MerkleDbConfig;
 import com.swirlds.merkledb.utilities.MerkleDbFileUtils;
 import edu.umd.cs.findbugs.annotations.NonNull;
+import edu.umd.cs.findbugs.annotations.Nullable;
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.channels.FileChannel;
@@ -20,8 +21,10 @@
 import java.nio.file.StandardOpenOption;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Objects;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReferenceArray;
+import java.util.function.BooleanSupplier;
 import java.util.stream.LongStream;
 import java.util.stream.StreamSupport;
 
@@ -781,20 +784,30 @@ private void checkValue(final long value) {
         }
     }
 
-    /** {@inheritDoc} */
+    /**
+     * {@inheritDoc}
+     *
+     * <p>This long list implementation checks the condition (if not null) before every list
+     * item is processed.
+     */
     @Override
-    public <T extends Throwable> void forEach(final LongAction<T> action) throws InterruptedException, T {
+    public <T extends Throwable> boolean forEach(
+            @NonNull final LongAction<T> action, @Nullable final BooleanSupplier cond) throws InterruptedException, T {
         final long max = maxValidIndex.get();
         if (max < 0) {
             // Empty list, nothing to do
-            return;
+            return true;
         }
-        for (long i = minValidIndex.get(); i <= max; i++) {
+        Objects.requireNonNull(action);
+        final BooleanSupplier condition = cond != null ? cond : () -> true;
+        long i = minValidIndex.get();
+        for (; (i <= max) && condition.getAsBoolean(); i++) {
             final long value = get(i);
             if (value != IMPERMISSIBLE_VALUE) {
                 action.handle(i, value);
             }
         }
+        return i == max + 1;
     }
 
     /**

platform-sdk/swirlds-merkledb/src/main/java/com/swirlds/merkledb/collections/CASableLongIndex.java

@@ -1,6 +1,10 @@
 // SPDX-License-Identifier: Apache-2.0
 package com.swirlds.merkledb.collections;
 
+import edu.umd.cs.findbugs.annotations.NonNull;
+import edu.umd.cs.findbugs.annotations.Nullable;
+import java.util.function.BooleanSupplier;
+
 /**
  * An interface for classes that provide long index functionality. Such long indices must
  * support Compare-And-Swap operations and iterations over all valid index entries.
@@ -33,17 +37,25 @@ public interface CASableLongIndex {
     /**
      * Iterates over all valid index entries and calls the specified action for each of them.
      *
-     * @param action Action to call.
+     * <p>If a condition to check is provided, implementations should do their best checking it,
+     * but there is no guarantee it is checked before each entry in the list, or checked at all.
+     * If the condition is {@code false}, no more index entries are iterated over, and this
+     * method returns.
+     *
      * @param <T> Type of throwables allowed to throw by this method
+     * @param action Action to call
+     * @param whileCondition Optional condition to check if this method should be stopped
+     * @return {@code true} if all index items have been processed by this method
      * @throws InterruptedException If the thread running the method is interrupted
      * @throws T If an error occurs
      */
-    <T extends Throwable> void forEach(LongAction<T> action) throws InterruptedException, T;
+    <T extends Throwable> boolean forEach(@NonNull LongAction<T> action, @Nullable BooleanSupplier whileCondition)
+            throws InterruptedException, T;
 
     /**
-     * Action interface to use in {@link #forEach(LongAction)}. It could be a standard Java API
-     * interface like BiFunction, but all these APIs work with boxed Long type instead of
-     * primitive longs, which adds unnecessary load on GC. So let's use something more simple
+     * Action interface to use in {@link #forEach(LongAction, BooleanSupplier)}. It could be a standard Java
+     * API interface like BiFunction, but all these APIs work with boxed Long type instead of primitive
+     * longs, which adds unnecessary load on GC. So let's use something more simple
      *
      * @param <T> Type of throwable allowed to throw by the action
      */