introduced executor service for pipeline operations using virtual threads

Signed-off-by: Alex Kehayov <aleks.kehayov@limechain.tech>

Author: AlexKehayov

hedera-node/docs/design/app/blocks/BlockNodeConnection.md

@@ -173,21 +173,23 @@ To detect unresponsive block nodes during message transmission, the connection i
 
 #### Timeout Behavior
 
-Pipeline operations (`onNext()` and `onComplete()`) are blocking I/O operations that are executed on a separate executor thread with timeout enforcement using `Future.get(timeout)`.
+Pipeline operations (`onNext()` and `onComplete()`) are blocking I/O operations that are executed on a dedicated virtual thread executor with timeout enforcement using `Future.get(timeout)`. Each connection instance creates its own virtual thread executor to isolate pipeline operations from other tasks and prevent blocking.
 
-- **onNext() timeout**: When sending block items via `sendRequest()`, the operation is submitted to an executor and the calling thread blocks waiting for completion with a timeout. If the operation does not complete within the configured timeout period:
+- **onNext() timeout**: When sending block items via `sendRequest()`, the operation is submitted to the connection's dedicated executor and the calling thread blocks waiting for completion with a timeout. If the operation does not complete within the configured timeout period:
   - The Future is cancelled to interrupt the blocked operation
   - The timeout metric is incremented
   - `handleStreamFailure()` is triggered (only if connection is still ACTIVE)
   - The connection follows standard failure handling with exponential backoff retry
   - The connection manager will select a different block node for the next attempt if one is available
   - `TimeoutException` is caught and handled internally
-- **onComplete() timeout**: When closing the stream via `closePipeline()`, the operation is submitted to an executor with the same timeout mechanism. If the operation does not complete within the configured timeout period:
+- **onComplete() timeout**: When closing the stream via `closePipeline()`, the operation is submitted to the same dedicated executor with the same timeout mechanism. If the operation does not complete within the configured timeout period:
   - The Future is cancelled to interrupt the blocked operation
   - The timeout metric is incremented
   - Since the connection is already in CLOSING state, only the timeout is logged
   - The connection completes the close operation normally
 
+**Note**: The dedicated virtual thread executor is shut down when the connection closes, ensuring no resource leaks.
+
 #### Exception Handling
 
 The implementation handles multiple exception scenarios:

hedera-node/hedera-app/src/main/java/com/hedera/node/app/blocks/impl/streaming/BlockNodeConnection.java

@@ -31,6 +31,8 @@
 import java.util.Objects;
 import java.util.Optional;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.concurrent.Flow;
 import java.util.concurrent.Future;
 import java.util.concurrent.ScheduledExecutorService;
@@ -124,6 +126,13 @@ private record Options(Optional<String> authority, String contentType) implement
      * Scheduled executor service that is used to schedule periodic reset of the stream to help ensure stream health.
      */
     private final ScheduledExecutorService executorService;
+    /**
+     * Dedicated executor service for pipeline operations using virtual threads.
+     * This isolates pipeline operations from the shared scheduled executor to prevent
+     * blocking when the connection manager is busy with other tasks. Virtual threads
+     * make this approach lightweight despite creating one executor per connection.
+     */
+    private final ExecutorService pipelineExecutor;
     /**
      * This task runs every 24 hours (initial delay of 24 hours) when a connection is active.
      * The task helps maintain stream stability by forcing periodic reconnections.
@@ -204,6 +213,7 @@ public BlockNodeConnection(
         this.blockStreamMetrics = requireNonNull(blockStreamMetrics, "blockStreamMetrics must not be null");
         this.connectionState = new AtomicReference<>(ConnectionState.UNINITIALIZED);
         this.executorService = requireNonNull(executorService, "executorService must not be null");
+        this.pipelineExecutor = Executors.newVirtualThreadPerTaskExecutor();
         final var blockNodeConnectionConfig =
                 configProvider.getConfiguration().getConfigData(BlockNodeConnectionConfig.class);
         this.streamResetPeriod = blockNodeConnectionConfig.streamResetPeriod();
@@ -646,7 +656,7 @@ public void endTheStreamWith(final PublishStreamRequest.EndStream.Code code) {
                 highestAckedBlockNumber);
         try {
             sendRequest(endStream);
-        } catch (RuntimeException e) {
+        } catch (final RuntimeException e) {
             logger.warn(formatLogMessage("Error sending EndStream request", this), e);
         }
         close(true);
@@ -686,10 +696,10 @@ public void sendRequest(@NonNull final PublishStreamRequest request) {
                 }
                 final long startMs = System.currentTimeMillis();
 
-                Future<?> future = executorService.submit(() -> pipeline.onNext(request));
+                final Future<?> future = pipelineExecutor.submit(() -> pipeline.onNext(request));
                 try {
                     future.get(pipelineOperationTimeout.toMillis(), TimeUnit.MILLISECONDS);
-                } catch (TimeoutException e) {
+                } catch (final TimeoutException e) {
                     future.cancel(true); // Cancel the task if it times out
                     if (getConnectionState() == ConnectionState.ACTIVE) {
                         logWithContext(
@@ -701,10 +711,10 @@ public void sendRequest(@NonNull final PublishStreamRequest request) {
                         blockStreamMetrics.recordPipelineOperationTimeout();
                         handleStreamFailure();
                     }
-                } catch (InterruptedException e) {
+                } catch (final InterruptedException e) {
                     Thread.currentThread().interrupt(); // Restore interrupt status
                     throw new RuntimeException("Interrupted while waiting for pipeline.onNext()", e);
-                } catch (ExecutionException e) {
+                } catch (final ExecutionException e) {
                     throw new RuntimeException("Error executing pipeline.onNext()", e.getCause());
                 }
 
@@ -783,6 +793,16 @@ public void close(final boolean callOnComplete) {
             } catch (final Exception e) {
                 logger.error(formatLogMessage("Error occurred while closing gRPC client.", this), e);
             }
+            try {
+                pipelineExecutor.shutdown();
+                if (!pipelineExecutor.awaitTermination(5, TimeUnit.SECONDS)) {
+                    pipelineExecutor.shutdownNow();
+                }
+            } catch (final InterruptedException e) {
+                Thread.currentThread().interrupt();
+                pipelineExecutor.shutdownNow();
+                logger.error(formatLogMessage("Error occurred while shutting down pipeline executor.", this), e);
+            }
             blockStreamMetrics.recordConnectionClosed();
             blockStreamMetrics.recordActiveConnectionIp(-1L);
             // regardless of outcome, mark the connection as closed
@@ -800,11 +820,11 @@ private void closePipeline(final boolean callOnComplete) {
             try {
                 final ConnectionState state = getConnectionState();
                 if (state == ConnectionState.CLOSING && callOnComplete) {
-                    Future<?> future = executorService.submit(pipeline::onComplete);
+                    final Future<?> future = pipelineExecutor.submit(pipeline::onComplete);
                     try {
                         future.get(pipelineOperationTimeout.toMillis(), TimeUnit.MILLISECONDS);
                         logWithContext(logger, DEBUG, this, "Request pipeline successfully closed.");
-                    } catch (TimeoutException e) {
+                    } catch (final TimeoutException e) {
                         future.cancel(true); // Cancel the task if it times out
                         logWithContext(
                                 logger,
@@ -814,10 +834,10 @@ private void closePipeline(final boolean callOnComplete) {
                                 pipelineOperationTimeout.toMillis());
                         blockStreamMetrics.recordPipelineOperationTimeout();
                         // Connection is already closing, just log the timeout
-                    } catch (InterruptedException e) {
+                    } catch (final InterruptedException e) {
                         Thread.currentThread().interrupt(); // Restore interrupt status
                         logWithContext(logger, DEBUG, this, "Interrupted while waiting for pipeline.onComplete()");
-                    } catch (ExecutionException e) {
+                    } catch (final ExecutionException e) {
                         logWithContext(logger, DEBUG, this, "Error executing pipeline.onComplete()", e.getCause());
                     }
                 }