chore: wait for process termination (#21986) (#22217)

Signed-off-by: Michael Heinrichs <[email protected]>
Co-authored-by: Michael Heinrichs <[email protected]>

Author: poulok

platform-sdk/consensus-otter-docker-app/src/testFixtures/java/org/hiero/consensus/otter/docker/app/ConsensusNodeMain.java

@@ -1,6 +1,7 @@
 // SPDX-License-Identifier: Apache-2.0
 package org.hiero.consensus.otter.docker.app;
 
+import static com.swirlds.logging.legacy.LogMarker.STARTUP;
 import static org.hiero.otter.fixtures.container.utils.ContainerConstants.CONTAINER_APP_WORKING_DIR;
 
 import com.hedera.hapi.platform.state.NodeId;
@@ -39,6 +40,11 @@ public class ConsensusNodeMain {
     /** Logger */
     private static final Logger log = LogManager.getLogger(ConsensusNodeMain.class);
 
+    /**
+     * Main method to start the Consensus Node application.
+     *
+     * @param args command line arguments; expects a single argument representing the node's ID
+     */
     public static void main(final String[] args) {
         if (args.length != 1) {
             throw new IllegalArgumentException("Usage: ConsensusNodeMain <selfId>");
@@ -50,7 +56,7 @@ public static void main(final String[] args) {
 
         final NodeCommunicationService nodeCommunicationService = new NodeCommunicationService(selfId);
 
-        log.info("Starting ConsensusNodeMain");
+        log.info(STARTUP.getMarker(), "Starting ConsensusNodeMain");
         // Start the consensus node manager gRPC server
         final Server nodeGrpcServer = ServerBuilder.forPort(NODE_COMM_SERVICE_PORT)
                 .addService(nodeCommunicationService)
@@ -60,11 +66,11 @@ public static void main(final String[] args) {
             writeStartedMarkerFile();
             nodeGrpcServer.awaitTermination();
         } catch (final IOException ie) {
-            log.error("Failed to start the gRPC server for the consensus node manager", ie);
+            log.error(STARTUP.getMarker(), "Failed to start the gRPC server for the consensus node manager", ie);
             System.exit(-1);
         } catch (final InterruptedException e) {
             // Only warn, because we expect this exception when we interrupt the thread on a kill request
-            log.warn("Interrupted while running the consensus node manager gRPC server", e);
+            log.warn(STARTUP.getMarker(), "Interrupted while running the consensus node manager gRPC server", e);
             Thread.currentThread().interrupt();
             System.exit(-1);
         }
@@ -76,12 +82,18 @@ public static void main(final String[] args) {
     private static void writeStartedMarkerFile() {
         try {
             if (new File(STARTED_MARKER_FILE.toString()).createNewFile()) {
-                log.info("Node Communication Service marker file written to {}", STARTED_MARKER_FILE);
+                log.info(
+                        STARTUP.getMarker(),
+                        "Node Communication Service marker file written to {}",
+                        STARTED_MARKER_FILE);
             } else {
-                log.info("Node Communication Service marker file already exists at {}", STARTED_MARKER_FILE);
+                log.info(
+                        STARTUP.getMarker(),
+                        "Node Communication Service marker file already exists at {}",
+                        STARTED_MARKER_FILE);
             }
         } catch (final IOException e) {
-            log.error("Failed to write Node Communication Service marker file", e);
+            log.error(STARTUP.getMarker(), "Failed to write Node Communication Service marker file", e);
             throw new RuntimeException("Failed to write Node Communication Service marker file", e);
         }
     }

platform-sdk/consensus-otter-docker-app/src/testFixtures/java/org/hiero/consensus/otter/docker/app/DockerManager.java

@@ -120,7 +120,11 @@ public synchronized void init(
                 responseObserver.onNext(Empty.getDefaultInstance());
                 responseObserver.onCompleted();
             } else {
-                log.error("Consensus node process started, but marker file was not detected in the allowed time");
+                if (!process.isAlive()) {
+                    log.error("Consensus node stopped prematurely. Errorcode: {}", process.exitValue());
+                } else {
+                    log.error("Consensus node process started, but marker file was not detected in the allowed time");
+                }
                 responseObserver.onError(new IllegalStateException(
                         "Consensus node process started, but marker file was not detected in the allowed time"));
             }
@@ -188,9 +192,25 @@ public synchronized void killImmediately(
         log.info("Received kill request: {}", request);
         if (process != null) {
             process.destroyForcibly();
+            try {
+                if (process.waitFor(request.getTimeoutSeconds(), TimeUnit.SECONDS)) {
+                    responseObserver.onNext(Empty.getDefaultInstance());
+                    responseObserver.onCompleted();
+                } else {
+                    log.error("Failed to terminate the consensus node process within the timeout period.");
+                    responseObserver.onError(new IllegalStateException(
+                            "Failed to terminate the consensus node process within the timeout period."));
+                }
+            } catch (final InterruptedException e) {
+                log.error("Interrupted while waiting for the consensus node process to terminate.", e);
+                Thread.currentThread().interrupt();
+                responseObserver.onError(new InterruptedException(
+                        "Interrupted while waiting for the consensus node process to terminate."));
+            }
+        } else {
+            responseObserver.onNext(Empty.getDefaultInstance());
+            responseObserver.onCompleted();
         }
-        responseObserver.onNext(Empty.getDefaultInstance());
-        responseObserver.onCompleted();
         log.info("Kill request completed.");
     }
 

platform-sdk/consensus-otter-docker-app/src/testFixtures/java/org/hiero/consensus/otter/docker/app/logging/DockerLogConfigBuilder.java

@@ -1,6 +1,7 @@
 // SPDX-License-Identifier: Apache-2.0
 package org.hiero.consensus.otter.docker.app.logging;
 
+import static com.swirlds.logging.legacy.LogMarker.STARTUP;
 import static java.util.Objects.requireNonNull;
 import static org.hiero.otter.fixtures.logging.internal.LogConfigHelper.DEFAULT_PATTERN;
 import static org.hiero.otter.fixtures.logging.internal.LogConfigHelper.combineFilters;
@@ -100,6 +101,7 @@ public static void configure(@NonNull final Path baseDir, @NonNull final NodeId
 
         Configurator.reconfigure(builder.build());
 
-        LogManager.getLogger(DockerLogConfigBuilder.class).info("Logging configuration (re)initialized");
+        LogManager.getLogger(DockerLogConfigBuilder.class)
+                .info(STARTUP.getMarker(), "Logging configuration (re)initialized");
     }
 }

platform-sdk/consensus-otter-tests/src/testFixtures/java/org/hiero/otter/fixtures/container/ContainerNode.java

@@ -265,7 +265,9 @@ protected void doKillImmediately(@NonNull final Duration timeout) {
             // conditions with the stream observer receiving an error.
             lifeCycle = SHUTDOWN;
 
-            final KillImmediatelyRequest request = KillImmediatelyRequest.getDefaultInstance();
+            final KillImmediatelyRequest request = KillImmediatelyRequest.newBuilder()
+                    .setTimeoutSeconds((int) timeout.getSeconds())
+                    .build();
             // Unary call – will throw if server returns an error.
             containerControlBlockingStub.withDeadlineAfter(timeout).killImmediately(request);
             platformStatus = null;

platform-sdk/consensus-otter-tests/src/testFixtures/proto/container_communication.proto

@@ -16,7 +16,7 @@ message InitRequest {
 
 // Request to kill the application immediately.
 message KillImmediatelyRequest {
-  google.protobuf.Empty ignore = 1;
+  uint32 timeout_seconds = 1;
 }
 
 message PingResponse {