protected void updateKAFKAConfigs() throws AmbariException {
    AmbariManagementController ambariManagementController = injector.getInstance(AmbariManagementController.class);
    Clusters clusters = ambariManagementController.getClusters();
    if (clusters != null) {
      Map<String, Cluster> clusterMap = getCheckedClusterMap(clusters);
      if (clusterMap != null && !clusterMap.isEmpty()) {
        for (final Cluster cluster : clusterMap.values()) {
          Set<String> installedServices = cluster.getServices().keySet();

          if (installedServices.contains("KAFKA") && cluster.getSecurityType() == SecurityType.KERBEROS) {
            Config kafkaBroker = cluster.getDesiredConfigByType(KAFKA_BROKER_CONFIG);
            if (kafkaBroker != null) {
              String listenersPropertyValue = kafkaBroker.getProperties().get("listeners");
              if (StringUtils.isNotEmpty(listenersPropertyValue)) {
                String newListenersPropertyValue = listenersPropertyValue.replaceAll("\\bPLAINTEXT\\b", "PLAINTEXTSASL");
                if(!newListenersPropertyValue.equals(listenersPropertyValue)) {
                  updateConfigurationProperties(KAFKA_BROKER_CONFIG, Collections.singletonMap("listeners", newListenersPropertyValue), true, false);
                }
              }
            }
          }
        }
      }
    }
  }
Ensure that the updates from Ambari 2.4.0 are applied in the event the initial version is
Ambari 2.5.0, since this Kafka change failed to make it into Ambari 2.5.0.

If the base version was before Ambari 2.5.0, this method should wind up doing nothing.
@throws AmbariException
ambari-server/src/main/java/org/apache/ambari/server/upgrade/UpgradeCatalog251.java