Attemp to fix flakiness CoreFeaturesIT (strimzi#11758)

Signed-off-by: see-quick <maros.orsak159@gmail.com>

Author: see-quick

.azure/build-pipeline.yaml

@@ -32,7 +32,8 @@ stages:
           artifactRunVersion: ''
           artifactRunId: ''
     variables:
-      STRIMZI_TEST_CONTAINER_LOGGING_ENABLED: false
+      # set to false when none error occurs during 2 weeks - https://github.com/strimzi/strimzi-kafka-operator/issues/11839
+      STRIMZI_TEST_CONTAINER_LOGGING_ENABLED: true
 
   # Builds Strimzi docs
   - stage: build_docs

topic-operator/src/main/java/io/strimzi/operator/topic/TopicOperatorMain.java

@@ -51,7 +51,7 @@ public class TopicOperatorMain implements Liveness, Readiness {
     /* test */ final BatchingTopicController controller;
 
     private SharedIndexInformer<KafkaTopic> informer; // guarded by this
-    Thread shutdownHook; // guarded by this
+    /* test */ volatile Thread shutdownHook;
 
     private final ResourceEventHandler<KafkaTopic> resourceEventHandler;
     private final HealthCheckAndMetricsServer healthAndMetricsServer;
@@ -93,7 +93,7 @@ synchronized void start() {
             throw new IllegalStateException();
         }
 
-        shutdownHook = new Thread(this::shutdown, "TopicOperator-shutdown-hook");
+        shutdownHook = new Thread(this::stop, "TopicOperator-shutdown-hook");
         LOGGER.infoOp("Installing shutdown hook");
         Runtime.getRuntime().addShutdownHook(shutdownHook);
         LOGGER.infoOp("Starting health and metrics");
@@ -119,28 +119,17 @@ synchronized void start() {
     }
 
     synchronized void stop() {
+        // Execute the actual shutdown sequence (idempotent).
         if (shutdownHook == null) {
-            throw new IllegalStateException();
+            LOGGER.debugOp("Already shut down.");
+            return;
         }
-        // shutdown(), will be be invoked indirectly by calling
-        // hook.run() has the side effect of nullifying this.shutdownHook
-        // so retain a reference now so we have something to call
-        // removeShutdownHook() with.
-        var hook = shutdownHook;
-        // Call run (not start()) on the thread so that shutdown() is executed
-        // on this thread.
-        shutdown();
-        // stop() is _not_ called from the shutdown hook, so calling
-        // removeShutdownHook() should not cause IAE.
-        Runtime.getRuntime().removeShutdownHook(hook);
-    }
-
-    private synchronized void shutdown() {
+        shutdownHook = null;
+        LOGGER.infoOp("Shutdown initiated");
         // Note: This method can be invoked on either via the shutdown hook thread or
         // on the thread(s) on which stop()/start() are called
-        LOGGER.infoOp("Shutdown initiated");
         try {
-            shutdownHook = null;
+            // Idempotent resource teardown.
             if (informer != null) {
                 informer.stop();
                 informer = null;
@@ -152,6 +141,7 @@ private synchronized void shutdown() {
             this.healthAndMetricsServer.stop();
             LOGGER.infoOp("Shutdown completed normally");
         } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
             LOGGER.infoOp("Interrupted during shutdown");
             throw new RuntimeException(e);
         }
@@ -170,7 +160,7 @@ public static void main(String[] args) {
     public boolean isAlive() {
         boolean running;
         synchronized (this) {
-            running = informer.isRunning();
+            running = informer != null && informer.isRunning();
         }
         if (!running) {
             LOGGER.infoOp("isAlive returning false because informer is not running");
@@ -184,7 +174,7 @@ public boolean isAlive() {
     public boolean isReady() {
         boolean running;
         synchronized (this) {
-            running = informer.isRunning();
+            running = informer != null && informer.isRunning();
         }
         if (!running) {
             LOGGER.infoOp("isReady returning false because informer is not running");

topic-operator/src/test/java/io/strimzi/operator/topic/CoreFeaturesIT.java

@@ -143,7 +143,8 @@ public static void afterAll() {
 
     @AfterEach
     public void afterEach() {
-        if (operator != null) {
+        // Only check operator state if it wasn't intentionally shut down
+        if (operator != null && operator.shutdownHook != null) {
             assertTrue(operator.queue.isAlive());
             assertTrue(operator.queue.isReady());
         }
@@ -158,6 +159,11 @@ public void afterEach() {
             kafkaAdminClient = null;
         }
 
+        if (kafkaAdminClientOp != null) {
+            kafkaAdminClientOp.close();
+            kafkaAdminClientOp = null;
+        }
+
         if (kafkaCluster != null) {
             kafkaCluster.stop();
             kafkaCluster = null;
@@ -2095,9 +2101,11 @@ public void shouldTerminateIfQueueFull() throws InterruptedException, TimeoutExc
         // then
         assertNull(operator.shutdownHook, "Expect the operator to shutdown");
 
-        // finally, because the @After method of this class asserts that the operator is running
-        // we start a new operator
-        kafkaAdminClientOp = null;
+        // clean up the shutdown operator and start a new one
+        if (kafkaAdminClientOp != null) {
+            kafkaAdminClientOp.close();
+            kafkaAdminClientOp = null;
+        }
         operator = null;
         maybeStartOperator(topicOperatorConfig(NAMESPACE, kafkaCluster));
     }