Introduce the concept of an hybrid ccloud provider, to set acls via admin client and translation via api

src/main/java/com/purbon/kafka/topology/AccessControlProviderFactory.java

@@ -6,6 +6,7 @@
 import com.purbon.kafka.topology.api.mds.MDSApiClient;
 import com.purbon.kafka.topology.api.mds.MDSApiClientBuilder;
 import com.purbon.kafka.topology.roles.CCloudAclsProvider;
+import com.purbon.kafka.topology.roles.HybridCCloudAclsProvider;
 import com.purbon.kafka.topology.roles.RBACProvider;
 import com.purbon.kafka.topology.roles.SimpleAclsProvider;
 import com.purbon.kafka.topology.roles.acls.AclsBindingsBuilder;
@@ -44,6 +45,11 @@ public AccessControlProvider get() throws IOException {
               clazz.getConstructor(TopologyBuilderAdminClient.class, Configuration.class);
           return (CCloudAclsProvider)
               ccloudProviderConstructor.newInstance(builderAdminClient, config);
+        case CONFLUENT_HYBRID_CLOUD_CONTROL_CLASS:
+          Constructor<?> hybridCcloudProviderConstructor =
+                  clazz.getConstructor(TopologyBuilderAdminClient.class, Configuration.class);
+          return (HybridCCloudAclsProvider)
+                  hybridCcloudProviderConstructor.newInstance(builderAdminClient, config);
         case RBAC_ACCESS_CONTROL_CLASS:
           Constructor<?> rbacProviderConstructor = clazz.getConstructor(MDSApiClient.class);
           MDSApiClient apiClient = apiClientLogIn();

src/main/java/com/purbon/kafka/topology/Constants.java

@@ -13,6 +13,9 @@ public class Constants {
   public static final String CONFLUENT_CLOUD_CONTROL_CLASS =
       "com.purbon.kafka.topology.roles.CCloudAclsProvider";
 
+  public static final String CONFLUENT_HYBRID_CLOUD_CONTROL_CLASS =
+          "com.purbon.kafka.topology.roles.HybridCCloudAclsProvider";
+
   public static final String RBAC_ACCESS_CONTROL_CLASS =
       "com.purbon.kafka.topology.roles.RBACProvider";
 

src/main/java/com/purbon/kafka/topology/roles/HybridCCloudAclsProvider.java

@@ -0,0 +1,81 @@
+package com.purbon.kafka.topology.roles;
+
+import com.purbon.kafka.topology.AccessControlProvider;
+import com.purbon.kafka.topology.Configuration;
+import com.purbon.kafka.topology.api.adminclient.TopologyBuilderAdminClient;
+import com.purbon.kafka.topology.api.ccloud.CCloudApi;
+import com.purbon.kafka.topology.utils.CCloudUtils;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+public class HybridCCloudAclsProvider extends SimpleAclsProvider implements AccessControlProvider {
+
+  private static final Logger LOGGER = LogManager.getLogger(HybridCCloudAclsProvider.class);
+
+  private final CCloudApi cli;
+  private final String clusterId;
+  private final Configuration config;
+  private CCloudUtils cCloudUtils;
+
+  public HybridCCloudAclsProvider(
+      final TopologyBuilderAdminClient adminClient, final Configuration config) throws IOException {
+    super(adminClient);
+    this.cli = new CCloudApi(config.getConfluentCloudClusterUrl(), config);
+    this.clusterId = config.getConfluentCloudClusterId();
+    this.config = config;
+    this.cCloudUtils = new CCloudUtils(config);
+  }
+
+  @Override
+  public void createBindings(Set<TopologyAclBinding> bindings) throws IOException {
+    var serviceAccountIdByNameMap = cCloudUtils.initializeLookupTable(this.cli);
+    var mayBeTranslated = bindings
+            .stream()
+            .map(binding -> {
+      try {
+        return cCloudUtils.translateIfNecessary(binding, serviceAccountIdByNameMap);
+      } catch (IOException e) {
+        LOGGER.error(e);
+        return binding;
+      }
+    }).collect(Collectors.toSet());
+    super.createBindings(mayBeTranslated);
+  }
+
+  @Override
+  public void clearBindings(Set<TopologyAclBinding> bindings) throws IOException {
+    var serviceAccountIdByNameMap = cCloudUtils.initializeLookupTable(this.cli);
+    for (TopologyAclBinding binding : bindings) {
+      adminClient.clearAcls(cCloudUtils.translateIfNecessary(binding, serviceAccountIdByNameMap));
+    }
+  }
+
+  @Override
+  public Map<String, List<TopologyAclBinding>> listAcls() {
+    try {
+      Map<String, List<TopologyAclBinding>> bindings = new HashMap<>();
+      for (TopologyAclBinding binding : cli.listAcls(clusterId)) {
+        String resourceName = binding.getResourceName();
+        if (!bindings.containsKey(resourceName)) {
+          bindings.put(resourceName, new ArrayList<>());
+        }
+        bindings.get(resourceName).add(binding);
+      }
+      return bindings;
+    } catch (IOException e) {
+      LOGGER.warn(e);
+      return Collections.emptyMap();
+    }
+  }
+}

src/main/java/com/purbon/kafka/topology/roles/SimpleAclsProvider.java

@@ -16,7 +16,7 @@ public class SimpleAclsProvider implements AccessControlProvider {
 
   private static final Logger LOGGER = LogManager.getLogger(SimpleAclsProvider.class);
 
-  private final TopologyBuilderAdminClient adminClient;
+  protected final TopologyBuilderAdminClient adminClient;
 
   public SimpleAclsProvider(final TopologyBuilderAdminClient adminClient) {
     this.adminClient = adminClient;