src/main/java/com/purbon/kafka/topology/AccessControlManager.java
/**
   * Build a list of actions required to create or delete necessary bindings
   *
   * @param actions List of pre computed actions based on a topology
   * @param bindings List of current bindings available in the cluster
   * @return List<Action> list of actions necessary to update the cluster
   */
Build a list of actions required to create or delete necessary bindings
@param actions List of pre computed actions based on a topology
@param bindings List of current bindings available in the cluster
@return List list of actions necessary to update the cluster
Build a list of actions required to create or delete necessary bindings @param actions List of pre computed actions based on a topology @param bindings List of current bindings available in the cluster @return List list of actions necessary to update the cluster
public List<Action> buildUpdateBindingsActions(
      List<Action> actions, Set<TopologyAclBinding> bindings) throws IOException {
    List<Action> updateActions = new ArrayList<>();
    List<Either> eitherStreamOrError =
        actions.stream()
            .map(
                action -> {
                  try {
                    action.run();
                    return Either.Left(action.getBindings().stream());
                  } catch (IOException e) {
                    return Either.Right(e);
                  }
                })
            .collect(Collectors.toList());
    List<IOException> errors =
        eitherStreamOrError.stream()
            .filter(Either::isRight)
            .map(e -> (IOException) e.getRight().get())
            .collect(Collectors.toList());
    if (!errors.isEmpty()) {
      for (IOException err : errors) {
        LOGGER.error(err);
      }
      throw errors.get(0);
    }
    Set<TopologyAclBinding> allFinalBindings =
        eitherStreamOrError.stream()
            .filter(Either::isLeft)
            .flatMap(e -> (Stream<TopologyAclBinding>) e.getLeft().get())
            .collect(Collectors.toSet());
    Set<TopologyAclBinding> bindingsToBeCreated =
        allFinalBindings.stream()
            .filter(Objects::nonNull)
            .filter(this::matchesManagedPrefixList)
            .filter(binding -> !bindings.contains(binding))
            .collect(Collectors.toSet());
    if (!bindingsToBeCreated.isEmpty()) {
      CreateBindings createBindings = new CreateBindings(controlProvider, bindingsToBeCreated);
      updateActions.add(createBindings);
    }
    if (config.isAllowDeleteBindings()) {
      Set<TopologyAclBinding> bindingsToDelete =
          bindings.stream()
              .filter(binding -> !allFinalBindings.contains(binding))
              .collect(Collectors.toSet());
      if (!bindingsToDelete.isEmpty()) {
        ClearBindings clearBindings = new ClearBindings(controlProvider, bindingsToDelete);
        updateActions.add(clearBindings);
      }
    }
    return updateActions;
  }
public List < Action > buildUpdateBindingsActions ( List < Action > actions , Set < TopologyAclBinding > bindings ) throws IOException { List < Action > updateActions = new ArrayList < > ( ) ; List < Either > eitherStreamOrError = actions . stream ( ) . map ( action -> { try { action . run ( ) ; return Either . Left ( action . getBindings ( ) . stream ( ) ) ; } catch ( IOException e ) { return Either . Right ( e ) ; } } ) . collect ( Collectors . toList ( ) ) ; List < IOException > errors = eitherStreamOrError . stream ( ) . filter ( Either :: isRight ) . map ( e -> ( IOException ) e . getRight ( ) . get ( ) ) . collect ( Collectors . toList ( ) ) ; if ( ! errors . isEmpty ( ) ) { for ( IOException err : errors ) { LOGGER . error ( err ) ; } throw errors . get ( 0 ) ; } Set < TopologyAclBinding > allFinalBindings = eitherStreamOrError . stream ( ) . filter ( Either :: isLeft ) . flatMap ( e -> ( Stream < TopologyAclBinding > ) e . getLeft ( ) . get ( ) ) . collect ( Collectors . toSet ( ) ) ; Set < TopologyAclBinding > bindingsToBeCreated = allFinalBindings . stream ( ) . filter ( Objects :: nonNull ) . filter ( this :: matchesManagedPrefixList ) . filter ( binding -> ! bindings . contains ( binding ) ) . collect ( Collectors . toSet ( ) ) ; if ( ! bindingsToBeCreated . isEmpty ( ) ) { CreateBindings createBindings = new CreateBindings ( controlProvider , bindingsToBeCreated ) ; updateActions . add ( createBindings ) ; } if ( config . isAllowDeleteBindings ( ) ) { Set < TopologyAclBinding > bindingsToDelete = bindings . stream ( ) . filter ( binding -> ! allFinalBindings . contains ( binding ) ) . collect ( Collectors . toSet ( ) ) ; if ( ! bindingsToDelete . isEmpty ( ) ) { ClearBindings clearBindings = new ClearBindings ( controlProvider , bindingsToDelete ) ; updateActions . add ( clearBindings ) ; } } return updateActions ; }
Build a list of actions required to create or delete necessary bindings
 @param actions List of pre computed actions based on a topology
 @param bindings List of current bindings available in the cluster
 @return List<Action> list of actions necessary to update the cluster
Build a list of actions required to create or delete necessary bindings @param actions List of pre computed actions based on a topology @param bindings List of current bindings available in the cluster @return List<Action > list of actions necessary to update the cluster
// Only create what we manage // Diff of bindings, so we only create what is not already created in the cluster. // clear acls that does not appear anymore in the new generated list, // but where previously created