Optimization improvements over the PR Review process

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

Author: AlfredoG87

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

@@ -22,6 +22,7 @@
  * @param perBlockProcessingTimeout Timeout in milliseconds for processing each block, to avoid blocking the backfill
  *                                  process indefinitely in case something unexpected happens, this would allow for self-recovery
  * @param grpcOverallTimeout single timeout configuration for gRPC Client construction, connectTimeout, readTimeout and pollWaitTime
+ * @param enableTLS if enabled will assume block-node client supports tls connection.
  */
 @ConfigData("backfill")
 public record BackfillConfiguration(
@@ -35,4 +36,5 @@ public record BackfillConfiguration(
         @Loggable @ConfigProperty(defaultValue = "1000") @Min(100) int delayBetweenBatches,
         @Loggable @ConfigProperty(defaultValue = "15000") @Min(5) int initialDelay,
         @Loggable @ConfigProperty(defaultValue = "1000") @Min(500) int perBlockProcessingTimeout,
-        @Loggable @ConfigProperty(defaultValue = "30000") @Min(10000) int grpcOverallTimeout) {}
+        @Loggable @ConfigProperty(defaultValue = "30000") @Min(10000) int grpcOverallTimeout,
+        @Loggable @ConfigProperty(defaultValue = "false") boolean enableTLS) {}

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

@@ -52,10 +52,10 @@ public class BackfillGrpcClient {
      * This is used for exponential backoff in case of failures.
      */
     private final int initialRetryDelayMs;
-
     /** Connection timeout in milliseconds for gRPC calls to block nodes. */
     private final int connectionTimeoutSeconds;
-
+    /** Enable TLS for secure connections to block nodes. */
+    private final boolean enableTls;
     /** Current status of the Block Node Clients */
     private ConcurrentHashMap<BackfillSourceConfig, Status> nodeStatusMap = new ConcurrentHashMap<>();
     /**
@@ -74,13 +74,15 @@ public BackfillGrpcClient(
             int maxRetries,
             Counter backfillRetriesCounter,
             int retryInitialDelayMs,
-            int connectionTimeoutSeconds)
+            int connectionTimeoutSeconds,
+            boolean enableTls)
             throws IOException, ParseException {
         this.blockNodeSource = BackfillSource.JSON.parse(Bytes.wrap(Files.readAllBytes(blockNodePreferenceFilePath)));
         this.maxRetries = maxRetries;
         this.initialRetryDelayMs = retryInitialDelayMs;
         this.backfillRetries = backfillRetriesCounter;
         this.connectionTimeoutSeconds = connectionTimeoutSeconds;
+        this.enableTls = enableTls;
 
         for (BackfillSourceConfig node : blockNodeSource.nodes()) {
             LOGGER.log(INFO, "Address: {0}, Port: {1}, Priority: {2}", node.address(), node.port(), node.priority());
@@ -207,7 +209,7 @@ public List<BlockUnparsed> fetchBlocks(LongRange blockRange) {
      */
     private BlockNodeClient getNodeClient(BackfillSourceConfig node) {
         return nodeClientMap.computeIfAbsent(
-                node, BlockNodeClient -> new BlockNodeClient(node, connectionTimeoutSeconds));
+                node, BlockNodeClient -> new BlockNodeClient(node, connectionTimeoutSeconds, enableTls));
     }
 
     /**

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

@@ -1,6 +1,7 @@
 // SPDX-License-Identifier: Apache-2.0
 package org.hiero.block.node.backfill;
 
+import static java.lang.System.Logger.Level.DEBUG;
 import static java.lang.System.Logger.Level.INFO;
 import static java.lang.System.Logger.Level.TRACE;
 
@@ -181,7 +182,8 @@ public void init(BlockNodeContext context, ServiceBuilder serviceBuilder) {
                     backfillConfiguration.maxRetries(),
                     this.backfillRetries,
                     backfillConfiguration.initialRetryDelay(),
-                    backfillConfiguration.grpcOverallTimeout());
+                    backfillConfiguration.grpcOverallTimeout(),
+                    backfillConfiguration.enableTLS());
             LOGGER.log(TRACE, "Initialized gRPC client with sources path: {0}", blockNodeSourcesPath);
         } catch (Exception e) {
             LOGGER.log(INFO, "Failed to initialize gRPC client: {0}", e.getMessage());
@@ -334,7 +336,7 @@ private void backfillGap(LongRange gap, BackfillType backfillType) throws Interr
             getLatch(backfillType).set(new CountDownLatch(batchOfBlocks.size()));
 
             if (batchOfBlocks.isEmpty()) {
-                LOGGER.log(TRACE, "No blocks fetched for gap {0}, skipping", chunk);
+                LOGGER.log(DEBUG, "No blocks fetched for gap {0}, skipping", chunk);
                 continue; // Skip empty batches
             }
 

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

@@ -14,7 +14,8 @@
 
 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 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);
@@ -28,11 +29,11 @@ private record Options(Optional<String> authority, String contentType) implement
      *
      * @param blockNodeConfig the configuration for the block node, including address and port
      */
-    public BlockNodeClient(BackfillSourceConfig blockNodeConfig, int timeoutMs) {
+    public BlockNodeClient(BackfillSourceConfig blockNodeConfig, int timeoutMs, boolean enableTls) {
 
         final Duration timeoutDuration = Duration.ofMillis(timeoutMs);
 
-        final Tls tls = Tls.builder().enabled(false).build();
+        final Tls tls = Tls.builder().enabled(enableTls).build();
         final PbjGrpcClientConfig grpcConfig =
                 new PbjGrpcClientConfig(timeoutDuration, tls, Optional.of(""), "application/grpc");
 

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

@@ -1,14 +1,13 @@
 // SPDX-License-Identifier: Apache-2.0
 package org.hiero.block.node.backfill.client;
 
+import static java.lang.System.Logger.Level.DEBUG;
 import static java.lang.System.Logger.Level.TRACE;
 import static java.util.Objects.requireNonNull;
 import static org.hiero.block.api.BlockStreamSubscribeServiceInterface.FULL_NAME;
 
 import com.hedera.hapi.block.stream.output.BlockHeader;
-import com.hedera.pbj.runtime.Codec;
 import com.hedera.pbj.runtime.ParseException;
-import com.hedera.pbj.runtime.UncheckedParseException;
 import com.hedera.pbj.runtime.grpc.GrpcCall;
 import com.hedera.pbj.runtime.grpc.GrpcClient;
 import com.hedera.pbj.runtime.grpc.Pipeline;
@@ -44,12 +43,11 @@ public class BlockStreamSubscribeUnparsedClient {
 
     // From constructor
     private final GrpcClient grpcClient;
-    private final ServiceInterface.RequestOptions requestOptions;
 
     public BlockStreamSubscribeUnparsedClient(
             @NonNull final GrpcClient grpcClient, @NonNull final ServiceInterface.RequestOptions requestOptions) {
         this.grpcClient = requireNonNull(grpcClient);
-        this.requestOptions = requireNonNull(requestOptions);
+        ServiceInterface.RequestOptions requestOptions1 = requireNonNull(requestOptions);
     }
 
     /**
@@ -77,72 +75,13 @@ public List<BlockUnparsed> getBatchOfBlocks(long startBlockNumber, long endBlock
                 .build();
 
         // Create a per-request pipeline that closes over `ctx`
-        final Pipeline<SubscribeStreamResponseUnparsed> pipeline = new Pipeline<>() {
-            @Override
-            public void onSubscribe(Flow.Subscription subscription) {
-                LOGGER.log(TRACE, "received onSubscribe confirmation");
-                // No backpressure negotiation needed for this pattern.
-            }
-
-            @Override
-            public void onNext(SubscribeStreamResponseUnparsed subscribeStreamResponse) {
-                if (subscribeStreamResponse.hasBlockItems()) {
-                    final List<BlockItemUnparsed> blockItems =
-                            subscribeStreamResponse.blockItems().blockItems();
-
-                    // Start of a new block
-                    if (blockItems.getFirst().hasBlockHeader()) {
-                        final long expected = ctx.expectedBlockNumber;
-                        final long actual = extractBlockNumberFromBlockHeader(blockItems.getFirst());
-                        if (actual != expected) {
-                            ctx.fail(new IllegalStateException(
-                                    "Expected block number " + expected + " but received " + actual));
-                            return;
-                        }
-                        // Begin a new block with the header and following items in this frame
-                        ctx.currentBlockItems = new ArrayList<>(blockItems);
-                    } else {
-                        // Continuation of the current block
-                        ctx.currentBlockItems.addAll(blockItems);
-                    }
-
-                    // End of block
-                    if (blockItems.getLast().hasBlockProof()) {
-                        ctx.blocks.add(BlockUnparsed.newBuilder()
-                                .blockItems(ctx.currentBlockItems)
-                                .build());
-                        ctx.currentBlockItems = new ArrayList<>();
-                        ctx.expectedBlockNumber++;
-                    }
-
-                } else if (subscribeStreamResponse.hasStatus()) {
-                    final SubscribeStreamResponse.Code code = subscribeStreamResponse.status();
-                    if (code != SubscribeStreamResponse.Code.SUCCESS) {
-                        ctx.fail(new RuntimeException("Received error code: " + code));
-                    }
-                } else {
-                    ctx.fail(new RuntimeException("Received unexpected response without block items or code"));
-                }
-            }
-
-            @Override
-            public void onError(Throwable throwable) {
-                LOGGER.log(TRACE, "received onError", throwable);
-                ctx.fail(throwable);
-            }
-
-            @Override
-            public void onComplete() {
-                LOGGER.log(TRACE, "received onComplete");
-                ctx.complete();
-            }
-        };
+        final Pipeline<SubscribeStreamResponseUnparsed> pipeline = new SubscribePipeline(ctx);
 
         // Issue the call using the per-request pipeline
         final GrpcCall<SubscribeStreamRequest, SubscribeStreamResponseUnparsed> call = grpcClient.createCall(
                 FULL_NAME + "/subscribeBlockStream",
-                getSubscribeStreamRequestCodec(requestOptions),
-                getSubscribeStreamResponseUnparsedCodec(requestOptions),
+                SubscribeStreamRequest.PROTOBUF,
+                SubscribeStreamResponseUnparsed.PROTOBUF,
                 pipeline);
 
         call.sendRequest(request, true);
@@ -154,35 +93,8 @@ public void onComplete() {
     /**
      * Extracts the block number from a block header item.
      */
-    private static long extractBlockNumberFromBlockHeader(BlockItemUnparsed itemUnparsed) {
-        try {
-            return BlockHeader.PROTOBUF.parse(itemUnparsed.blockHeaderOrThrow()).number();
-        } catch (ParseException e) {
-            throw new UncheckedParseException(e);
-        }
-    }
-
-    private static Codec<SubscribeStreamRequest> getSubscribeStreamRequestCodec(
-            @NonNull final ServiceInterface.RequestOptions options) {
-        requireNonNull(options);
-        // Default to protobuf, and don't error out if both are set:
-        if (options.isJson() && !options.isProtobuf()) {
-            return SubscribeStreamRequest.JSON;
-        } else {
-            return SubscribeStreamRequest.PROTOBUF;
-        }
-    }
-
-    @NonNull
-    private static Codec<SubscribeStreamResponseUnparsed> getSubscribeStreamResponseUnparsedCodec(
-            @NonNull final ServiceInterface.RequestOptions options) {
-        requireNonNull(options);
-        // Default to protobuf, and don't error out if both are set:
-        if (options.isJson() && !options.isProtobuf()) {
-            return SubscribeStreamResponseUnparsed.JSON;
-        } else {
-            return SubscribeStreamResponseUnparsed.PROTOBUF;
-        }
+    private static long extractBlockNumberFromBlockHeader(BlockItemUnparsed itemUnparsed) throws ParseException {
+        return BlockHeader.PROTOBUF.parse(itemUnparsed.blockHeaderOrThrow()).number();
     }
 
     /**
@@ -214,12 +126,85 @@ List<BlockUnparsed> await() {
                 done.await();
             } catch (InterruptedException e) {
                 Thread.currentThread().interrupt();
-                throw new RuntimeException("Interrupted while waiting for blocks", e);
             }
             if (error != null) {
                 throw new RuntimeException("Error fetching blocks", error);
             }
             return blocks;
         }
     }
+
+    private static final class SubscribePipeline implements Pipeline<SubscribeStreamResponseUnparsed> {
+        private final RequestContext ctx;
+
+        SubscribePipeline(RequestContext ctx) {
+            this.ctx = ctx;
+        }
+
+        @Override
+        public void onSubscribe(Flow.Subscription subscription) {
+            LOGGER.log(TRACE, "received onSubscribe confirmation");
+            // No backpressure negotiation needed for this pattern.
+        }
+
+        @Override
+        public void onNext(SubscribeStreamResponseUnparsed resp) {
+            try {
+                if (resp.hasBlockItems()) {
+                    final List<BlockItemUnparsed> frame = resp.blockItems().blockItems();
+
+                    if (frame.getFirst().hasBlockHeader()) {
+                        final long expected = ctx.expectedBlockNumber;
+                        final long actual = extractBlockNumberFromBlockHeader(frame.getFirst());
+                        if (actual != expected) {
+                            ctx.fail(new IllegalStateException(
+                                    "Expected block number " + expected + " but received " + actual));
+                            return;
+                        }
+                        // Start a new block: reuse the buffer and populate it.
+                        ctx.currentBlockItems.clear();
+                        ctx.currentBlockItems.addAll(frame);
+                    } else {
+                        // Continuation: append to the same buffer.
+                        ctx.currentBlockItems.addAll(frame);
+                    }
+
+                    if (frame.getLast().hasBlockProof()) {
+                        // Snapshot the current items to avoid retaining the large buffer in the finished block.
+                        final List<BlockItemUnparsed> snapshot = List.copyOf(ctx.currentBlockItems);
+                        ctx.blocks.add(
+                                BlockUnparsed.newBuilder().blockItems(snapshot).build());
+                        ctx.currentBlockItems.clear();
+                        ctx.expectedBlockNumber++;
+                    }
+
+                } else if (resp.hasStatus()) {
+                    final SubscribeStreamResponse.Code code = resp.status();
+                    if (code != SubscribeStreamResponse.Code.SUCCESS) {
+                        ctx.fail(new RuntimeException("Received error code: " + code));
+                    }
+                } else {
+                    ctx.fail(new RuntimeException("Received unexpected response without block items or code"));
+                }
+            } catch (ParseException e) {
+                LOGGER.log(DEBUG, "Parse error in block item", e);
+                ctx.fail(e);
+            } catch (RuntimeException e) {
+                LOGGER.log(DEBUG, "Runtime error processing SubscribeStreamResponseUnparsed", e);
+                ctx.fail(e);
+            }
+        }
+
+        @Override
+        public void onError(Throwable throwable) {
+            LOGGER.log(TRACE, "received onError", throwable);
+            ctx.fail(throwable);
+        }
+
+        @Override
+        public void onComplete() {
+            LOGGER.log(TRACE, "received onComplete");
+            ctx.complete();
+        }
+    }
 }