Fix misleading WARN when polling connector status after KafkaConnector delete (#12710)

Signed-off-by: Parameshwaran Krishnasamy <Parameshwaran.K@ibm.com>

Author: Paramesh324

cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/assembly/KafkaConnectApiImpl.java

@@ -131,8 +131,16 @@ private <T> CompletableFuture<T> doGet(Reconciliation reconciliation, String hos
                         LOGGER.debugCr(reconciliation, "Got {} response to GET request to {}: {}", statusCode, path, json.asText());
                         return CompletableFuture.completedFuture(OBJECT_MAPPER.convertValue(json, type));
                     } else {
-                        LOGGER.debugCr(reconciliation, "Got {} response to GET request to {}", statusCode, path);
-                        return CompletableFuture.failedFuture(new ConnectRestException(response, tryToExtractErrorMessage(reconciliation, response.body())));
+                        LOGGER.debugCr(reconciliation, "Got unexpected {} response to GET request to {}. Ok status codes: {}",
+                                statusCode, path, okStatusCodes);
+                        String message;
+                        if (statusCode >= 200 && statusCode < 300) {
+                            // Unlisted 2xx: poll may get 200 before 404 after delete.
+                            message = "Unexpected HTTP status code " + statusCode;
+                        } else {
+                            message = tryToExtractErrorMessage(reconciliation, response.body());
+                        }
+                        return CompletableFuture.failedFuture(new ConnectRestException(response, message));
                     }
                 });
     }

cluster-operator/src/test/java/io/strimzi/operator/cluster/operator/assembly/KafkaConnectApiImplTest.java

@@ -15,7 +15,9 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
+import java.util.Collections;
 import java.util.Map;
+import java.util.concurrent.ExecutionException;
 
 import static com.github.tomakehurst.wiremock.client.WireMock.aResponse;
 import static com.github.tomakehurst.wiremock.client.WireMock.get;
@@ -26,6 +28,7 @@
 import static org.hamcrest.Matchers.allOf;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.hasEntry;
+import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.is;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 
@@ -86,6 +89,45 @@ public void testFeatureCompletionWithWellFormattedError() {
                 .whenComplete((res, ex) -> assertThat(ex.getMessage(), containsString("This is the error")))::join);
     }
 
+    @Test
+    public void testStatusDoesNotTreatOkBodyAsErrorMessage() throws Exception {
+        // Poll /status for 404 only; 200 with connector status is normal until removal.
+        String statusBody = "{\"name\":\"my-connector\",\"connector\":{\"state\":\"RUNNING\"},"
+                + "\"tasks\":[]}";
+        server.stubFor(get(urlPathMatching(".*/connectors/my-connector/status"))
+                .willReturn(aResponse()
+                        .withStatus(200)
+                        .withBody(statusBody)));
+
+        KafkaConnectApi api = new KafkaConnectApiImpl();
+
+        ExecutionException ex = assertThrows(ExecutionException.class,
+                () -> api.status(
+                        Reconciliation.DUMMY_RECONCILIATION, "127.0.0.1", server.port(),
+                        "my-connector", Collections.singleton(404))
+                        .get());
+        assertThat(ex.getCause(), instanceOf(ConnectRestException.class));
+        assertThat(ex.getCause().getMessage(), containsString("Unexpected HTTP status code 200"));
+    }
+
+    @Test
+    public void testStatusWithValidErrorBody() throws Exception {
+        server.stubFor(get(urlPathMatching(".*/connectors/c/status"))
+                .willReturn(aResponse()
+                        .withStatus(500)
+                        .withBody("{\"message\": \"Connect cluster error\"}")));
+
+        KafkaConnectApi api = new KafkaConnectApiImpl();
+
+        ExecutionException ex = assertThrows(ExecutionException.class,
+                () -> api.status(
+                        Reconciliation.DUMMY_RECONCILIATION, "127.0.0.1", server.port(),
+                        "c", Collections.singleton(200))
+                        .get());
+        assertThat(ex.getCause(), instanceOf(ConnectRestException.class));
+        assertThat(ex.getCause().getMessage(), containsString("Connect cluster error"));
+    }
+
     @Test
     public void testListConnectLoggersWithLevel() throws Exception {
         server.stubFor(get(urlPathMatching(".*"))