Handle wildcard parameter in Connect Clusters deletion API

src/main/java/com/michelin/ns4kafka/controller/connect/ConnectClusterController.java

@@ -135,9 +135,11 @@ public Mono<HttpResponse<ConnectCluster>> apply(String namespace, @Body @Valid C
      * @param connectCluster The current connect cluster name to delete
      * @param dryrun         Run in dry mode or not
      * @return A HTTP response
+     * @deprecated use {@link #bulkDelete(String, String, boolean)} instead.
      */
     @Status(HttpStatus.NO_CONTENT)
     @Delete("/{connectCluster}{?dryrun}")
+    @Deprecated(since = "1.13.0")
     public HttpResponse<Void> delete(String namespace, String connectCluster,
                                      @QueryValue(defaultValue = "false") boolean dryrun) {
         Namespace ns = getNamespace(namespace);
@@ -166,20 +168,66 @@ public HttpResponse<Void> delete(String namespace, String connectCluster,
             return HttpResponse.noContent();
         }
 
-        ConnectCluster connectClusterToDelete = optionalConnectCluster.get();
+        performDeletion(optionalConnectCluster.get());
+        return HttpResponse.noContent();
+    }
 
-        sendEventLog(
-            connectClusterToDelete,
-            ApplyStatus.deleted,
-            connectClusterToDelete.getSpec(),
-            null,
-            EMPTY_STRING
-        );
+    /**
+     * Delete Kafka Connect clusters.
+     *
+     * @param namespace      The current namespace
+     * @param name           The name parameter
+     * @param dryrun         Run in dry mode or not
+     * @return A HTTP response
+     */
+    @Status(HttpStatus.NO_CONTENT)
+    @Delete
+    public HttpResponse<Void> bulkDelete(String namespace, @QueryValue(defaultValue = "*") String name,
+                                     @QueryValue(defaultValue = "false") boolean dryrun) {
+        Namespace ns = getNamespace(namespace);
 
-        connectClusterService.delete(connectClusterToDelete);
+        List<ConnectCluster> connectClusters = connectClusterService.findByWildcardNameWithOwnerPermission(ns, name);
+
+        List<String> validationErrors = new ArrayList<>();
+        connectClusters.forEach(cc -> {
+            List<Connector> connectors = connectorService.findAllByConnectCluster(ns, cc.getMetadata().getName());
+            if (!connectors.isEmpty()) {
+                validationErrors.add(invalidConnectClusterDeleteOperation(cc.getMetadata().getName(), connectors));
+            }
+        });
+
+        if (!validationErrors.isEmpty()) {
+            throw new ResourceValidationException(CONNECT_CLUSTER, name, validationErrors);
+        }
+
+        if (connectClusters.isEmpty()) {
+            return HttpResponse.notFound();
+        }
+
+        if (dryrun) {
+            return HttpResponse.noContent();
+        }
+
+        connectClusters.forEach(this::performDeletion);
         return HttpResponse.noContent();
     }
 
+    /**
+     * Perform the deletion of the connectCluster and send an event log.
+     *
+     * @param connectCluster The connectCluster to delete
+     */
+    private void performDeletion(ConnectCluster connectCluster) {
+        sendEventLog(
+                connectCluster,
+                ApplyStatus.deleted,
+                connectCluster.getSpec(),
+                null,
+                EMPTY_STRING
+        );
+        connectClusterService.delete(connectCluster);
+    }
+
     /**
      * List vault Kafka Connect clusters by namespace.
      *

src/main/java/com/michelin/ns4kafka/service/ConnectClusterService.java

@@ -82,6 +82,7 @@ public Flux<ConnectCluster> findAll(boolean all) {
         if (all) {
             results.addAll(managedClusterProperties
                 .stream()
+                .filter(cluster -> cluster.getConnects() != null)
                 .map(config -> config.getConnects().entrySet()
                     .stream()
                     .map(entry ->

src/test/java/com/michelin/ns4kafka/controller/ConnectClusterControllerTest.java

@@ -159,6 +159,7 @@ void shouldGetConnectCluster() {
     }
 
     @Test
+    @SuppressWarnings("deprecation")
     void shouldNotDeleteConnectClusterWhenNotOwner() {
         Namespace ns = Namespace.builder()
             .metadata(Metadata.builder()
@@ -177,6 +178,7 @@ void shouldNotDeleteConnectClusterWhenNotOwner() {
     }
 
     @Test
+    @SuppressWarnings("deprecation")
     void shouldNotDeleteConnectClusterWhenNotFound() {
         Namespace ns = Namespace.builder()
             .metadata(Metadata.builder()
@@ -197,6 +199,7 @@ void shouldNotDeleteConnectClusterWhenNotFound() {
     }
 
     @Test
+    @SuppressWarnings("deprecation")
     void shouldDeleteConnectCluster() {
         Namespace ns = Namespace.builder()
             .metadata(Metadata.builder()
@@ -229,6 +232,7 @@ void shouldDeleteConnectCluster() {
     }
 
     @Test
+    @SuppressWarnings("deprecation")
     void shouldDeleteConnectClusterInDryRunMode() {
         Namespace ns = Namespace.builder()
             .metadata(Metadata.builder()
@@ -259,6 +263,7 @@ void shouldDeleteConnectClusterInDryRunMode() {
     }
 
     @Test
+    @SuppressWarnings("deprecation")
     void shouldNotDeleteConnectClusterWithConnectorsAssociated() {
         Namespace ns = Namespace.builder()
             .metadata(Metadata.builder()
@@ -290,6 +295,140 @@ void shouldNotDeleteConnectClusterWithConnectorsAssociated() {
             result.getValidationErrors().getFirst());
     }
 
+    @Test
+    void shouldDeleteConnectClusters() {
+        Namespace ns = Namespace.builder()
+                .metadata(Metadata.builder()
+                        .name("test")
+                        .cluster("local")
+                        .build())
+                .build();
+
+        ConnectCluster connectCluster1 = ConnectCluster.builder()
+                .metadata(Metadata.builder()
+                        .name("connect-cluster1")
+                        .build())
+                .build();
+
+        ConnectCluster connectCluster2 = ConnectCluster.builder()
+                .metadata(Metadata.builder()
+                        .name("connect-cluster2")
+                        .build())
+                .build();
+
+        when(namespaceService.findByName("test"))
+                .thenReturn(Optional.of(ns));
+        when(connectorService.findAllByConnectCluster(ns, "connect-cluster1"))
+                .thenReturn(List.of());
+        when(connectorService.findAllByConnectCluster(ns, "connect-cluster2"))
+                .thenReturn(List.of());
+        when(connectClusterService.findByWildcardNameWithOwnerPermission(ns, "connect-cluster*"))
+                .thenReturn(List.of(connectCluster1, connectCluster2));
+        doNothing().when(connectClusterService).delete(connectCluster1);
+        doNothing().when(connectClusterService).delete(connectCluster2);
+        when(securityService.username()).thenReturn(Optional.of("test-user"));
+        when(securityService.hasRole(ResourceBasedSecurityRule.IS_ADMIN)).thenReturn(false);
+        doNothing().when(applicationEventPublisher).publishEvent(any());
+
+        HttpResponse<Void> actual = connectClusterController.bulkDelete("test", "connect-cluster*", false);
+        assertEquals(HttpStatus.NO_CONTENT, actual.getStatus());
+    }
+
+    @Test
+    void shouldDeleteConnectClustersInDryRunMode() {
+        Namespace ns = Namespace.builder()
+                .metadata(Metadata.builder()
+                        .name("test")
+                        .cluster("local")
+                        .build())
+                .build();
+
+        ConnectCluster connectCluster = ConnectCluster.builder()
+                .metadata(Metadata.builder()
+                        .name("connect-cluster")
+                        .build())
+                .build();
+
+        when(namespaceService.findByName("test"))
+                .thenReturn(Optional.of(ns));
+        when(connectorService.findAllByConnectCluster(ns, "connect-cluster"))
+                .thenReturn(List.of());
+        when(connectClusterService.findByWildcardNameWithOwnerPermission(ns, "connect-cluster*"))
+                .thenReturn(List.of(connectCluster));
+
+        HttpResponse<Void> actual = connectClusterController.bulkDelete("test", "connect-cluster*", true);
+        assertEquals(HttpStatus.NO_CONTENT, actual.getStatus());
+
+        verify(connectClusterService, never()).delete(any());
+    }
+
+    @Test
+    void shouldNotDeleteConnectClustersWhenNotFound() {
+        Namespace ns = Namespace.builder()
+                .metadata(Metadata.builder()
+                        .name("test")
+                        .cluster("local")
+                        .build())
+                .build();
+
+        when(namespaceService.findByName("test"))
+                .thenReturn(Optional.of(ns));
+        when(connectClusterService.findByWildcardNameWithOwnerPermission(ns, "connect-cluster*"))
+                .thenReturn(List.of());
+
+        HttpResponse<Void> actual = connectClusterController.bulkDelete("test", "connect-cluster*", false);
+        assertEquals(HttpStatus.NOT_FOUND, actual.getStatus());
+    }
+
+    @Test
+    void shouldNotDeleteConnectClustersWithConnectorsAssociated() {
+        Namespace ns = Namespace.builder()
+                .metadata(Metadata.builder()
+                        .name("test")
+                        .cluster("local")
+                        .build())
+                .build();
+
+        Connector connector = Connector.builder()
+                .metadata(Metadata.builder()
+                        .name("connect1")
+                        .build())
+                .build();
+
+        ConnectCluster connectCluster1 = ConnectCluster.builder()
+                .metadata(Metadata.builder()
+                        .name("connect-cluster1")
+                        .build())
+                .build();
+
+        ConnectCluster connectCluster2 = ConnectCluster.builder()
+                .metadata(Metadata.builder()
+                        .name("connect-cluster2")
+                        .build())
+                .build();
+
+        when(namespaceService.findByName("test"))
+                .thenReturn(Optional.of(ns));
+        when(connectorService.findAllByConnectCluster(ns, "connect-cluster1"))
+                .thenReturn(List.of());
+        when(connectorService.findAllByConnectCluster(ns, "connect-cluster2"))
+                .thenReturn(List.of());
+        when(connectClusterService.findByWildcardNameWithOwnerPermission(ns, "connect-cluster*"))
+                .thenReturn(List.of(connectCluster1, connectCluster2));
+
+        when(connectorService.findAllByConnectCluster(ns, "connect-cluster2"))
+                .thenReturn(List.of(connector));
+
+        ResourceValidationException result = assertThrows(ResourceValidationException.class,
+                () -> connectClusterController.bulkDelete("test", "connect-cluster*", false));
+
+        assertEquals(1, result.getValidationErrors().size());
+        assertEquals(
+                "Invalid \"delete\" operation: The Kafka Connect \"connect-cluster2\" has 1 deployed connector(s): "
+                        + "connect1. Please remove the associated connector(s) before deleting it.",
+                result.getValidationErrors().getFirst());
+    }
+
     @Test
     void shouldCreateConnectCluster() {
         Namespace ns = Namespace.builder()