[GitHub] [kafka] C0urante commented on a change in pull request #10629: BlockingConnectorTest improvements to verify Connectors and Tasks are successfully deleted

[GitBox <gi...@apache.org>]

C0urante commented on a change in pull request #10629:
URL: https://github.com/apache/kafka/pull/10629#discussion_r659903390



##########
File path: connect/runtime/src/test/java/org/apache/kafka/connect/util/clusters/EmbeddedConnectClusterAssertions.java
##########
@@ -392,6 +392,48 @@ protected boolean checkConnectorAndTasksAreStopped(String connectorName) {
                 && info.tasks().stream().noneMatch(s -> s.state().equals(AbstractStatus.State.RUNNING.toString()));
     }
 
+    /**
+     * Assert that a connector and its tasks are deleted.
+     *
+     * @param connectorName the connector name
+     * @param detailMessage the assertion message
+     * @throws InterruptedException
+     */
+    public void assertConnectorAndTasksAreDeleted(String connectorName, String detailMessage)

Review comment:
       Nit: do we really need the `AndTasks` section? Might be fine to use `assertConnectorIsDeleted` instead.

##########
File path: connect/runtime/src/test/java/org/apache/kafka/connect/util/clusters/EmbeddedConnectClusterAssertions.java
##########
@@ -392,6 +392,48 @@ protected boolean checkConnectorAndTasksAreStopped(String connectorName) {
                 && info.tasks().stream().noneMatch(s -> s.state().equals(AbstractStatus.State.RUNNING.toString()));
     }
 
+    /**
+     * Assert that a connector and its tasks are deleted.
+     *
+     * @param connectorName the connector name
+     * @param detailMessage the assertion message
+     * @throws InterruptedException
+     */
+    public void assertConnectorAndTasksAreDeleted(String connectorName, String detailMessage)
+            throws InterruptedException {
+        try {
+            waitForCondition(
+                () -> checkConnectorAndTasksAreDeleted(connectorName),
+                CONNECTOR_SETUP_DURATION_MS,
+                "At least the connector or one of its tasks still exists.");
+        } catch (AssertionError e) {
+            throw new AssertionError(detailMessage, e);
+        }
+    }
+
+    /**
+     * Check whether the connector or any of its tasks still exist.
+     *
+     * @param connectorName the connector
+     * @return true if the connector and all the tasks are not in RUNNING state; false otherwise
+     */
+    protected boolean checkConnectorAndTasksAreDeleted(String connectorName) {
+        ConnectorStateInfo info;
+        try {
+            info = connect.connectorStatus(connectorName);
+        } catch (ConnectRestException e) {
+            return e.statusCode() == Response.Status.NOT_FOUND.getStatusCode();
+        } catch (Exception e) {
+            log.error("Could not check connector state info.", e);
+            return false;
+        }
+        if (info == null) {
+            return true;
+        }

Review comment:
       Do we know when this might happen? My understanding is that part of the contract we're trying to test here is that Connect gives back a 404 response when a stalled connector is deleted; do we want to relax our expectations to also permit this case?

##########
File path: connect/runtime/src/test/java/org/apache/kafka/connect/integration/BlockingConnectorTest.java
##########
@@ -596,6 +662,8 @@ public void start(Map<String, String> props) {
             @Override
             public List<SourceRecord> poll() {
                 block.maybeBlockOn(SOURCE_TASK_POLL);
+                // even when not blocking, pause to prevent a tight loop
+                Utils.sleep(1000);

Review comment:
       Good call 👍 




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org