enabling actual parallel backfilling of multiple flow types and fixing misleading log

removing unneeded modules

Signed-off-by: Alfredo Gutierrez Grajeda <alfredo@hashgraph.com>

Author: AlfredoG87

block-node/backfill/src/main/java/module-info.java

@@ -20,14 +20,11 @@
     requires transitive com.swirlds.metrics.api;
     requires transitive org.hiero.block.node.spi;
     requires transitive org.hiero.block.protobuf.pbj;
-    requires transitive io.grpc.stub;
-    requires transitive io.grpc;
-    requires transitive io.helidon.grpc.core;
-    requires transitive io.helidon.webclient.grpc;
     requires com.hedera.pbj.grpc.client.helidon;
     requires org.hiero.block.node.base;
     requires io.helidon.common.tls;
     requires io.helidon.webclient.api;
+    requires io.helidon.webclient.grpc;
     requires org.antlr.antlr4.runtime;
     requires static transitive com.github.spotbugs.annotations;
 

block-node/backfill/src/main/java/org/hiero/block/node/backfill/BackfillPlugin.java

@@ -208,7 +208,8 @@ public void start() {
                 TRACE,
                 "Scheduling backfill process to start in {0} milliseconds",
                 backfillConfiguration.initialDelay());
-        scheduler = Executors.newSingleThreadScheduledExecutor();
+        scheduler = Executors.newScheduledThreadPool(
+                2); // Two threads: one for autonomous backfill, one for on-demand backfill
         scheduler.scheduleAtFixedRate(
                 this::detectGaps,
                 backfillConfiguration.initialDelay(),
@@ -333,6 +334,11 @@ private void backfillGap(LongRange gap, BackfillType backfillType) throws Interr
             // to avoid deadlocks, since blocks that fail verification are not persisted
             getLatch(backfillType).set(new CountDownLatch(batchOfBlocks.size()));
 
+            if (batchOfBlocks.isEmpty()) {
+                LOGGER.log(TRACE, "No blocks fetched for gap {0}, skipping", chunk);
+                continue; // Skip empty batches
+            }
+
             // Process each fetched block
             for (BlockUnparsed blockUnparsed : batchOfBlocks) {
                 long blockNumber = extractBlockNumber(blockUnparsed);
@@ -349,21 +355,25 @@ private void backfillGap(LongRange gap, BackfillType backfillType) throws Interr
             boolean backfillFinished = getLatch(backfillType).get().await(timeout, TimeUnit.MILLISECONDS);
 
             // Check if the backfill finished successfully
-            if (!backfillFinished) {
+            if (backfillFinished) {
+                // just log a victory message for each chunk
+                LOGGER.log(TRACE, "Successfully backfilled gap {0}", chunk);
+            } else {
                 LOGGER.log(TRACE, "Backfill for gap {0} did not finish in time", chunk);
                 backfillFetchErrors.increment();
                 // If it didn't finish, we will retry it later but move on to next chunk
-            } else {
-                // just log a victory message for each chunk
-                LOGGER.log(TRACE, "Successfully backfilled gap {0}", chunk);
             }
 
             // Cooldown between batches
             Thread.sleep(backfillConfiguration.delayBetweenBatches());
         }
 
         LOGGER.log(
-                TRACE, "Completed backfilling of type {0} gap from {1} to {2}", backfillType, gap.start(), gap.end());
+                TRACE,
+                "Completed backfilling task (does not mean success or failure, just completion) of type {0} gap from {1} to {2} ",
+                backfillType,
+                gap.start(),
+                gap.end());
         if (backfillType.equals(BackfillType.ON_DEMAND)) {
             onDemandBackfillStartBlock.set(-1); // Reset on-demand start block after backfill
         }

block-node/backfill/src/main/java/org/hiero/block/node/backfill/client/BlockNodeClient.java

@@ -15,6 +15,7 @@
 public class BlockNodeClient {
     // Options definition for all gRPC services in the block node client
     private record Options(Optional<String> authority, String contentType) implements ServiceInterface.RequestOptions {}
+
     private static final BlockNodeClient.Options OPTIONS =
             new BlockNodeClient.Options(Optional.empty(), ServiceInterface.RequestOptions.APPLICATION_GRPC);