class SensitiveConfigScrubber {

    public Map<String, Object> filterSensitiveOptions(final Map<String, Object> config, final Cluster cluster) {
        Objects.requireNonNull(config);

        // Create a copy of the map
        final Map<String, Object> copy = new HashMap<>(config);

        // Filter sensitive fields
        final String[] sensitiveKeys = new String[] {
            SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG, SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG,
        };
        Arrays.stream(sensitiveKeys)
            .filter(copy::containsKey)
            .forEach((key) -> copy.put(key, "**HIDDEN**"));

        // Filter JAAS Config
        if (copy.containsKey(SaslConfigs.SASL_JAAS_CONFIG) && cluster != null) {
            final SaslProperties saslProperties = saslUtility.decodeProperties(cluster);

            // Only replace if plainPassword field set and non-empty.
            if (saslProperties.getPlainPassword() != null && !saslProperties.getPlainPassword().isEmpty()) {
                String jaasConfig = (String) copy.get(SaslConfigs.SASL_JAAS_CONFIG);
                jaasConfig = jaasConfig.replaceAll(saslProperties.getPlainPassword(), "**HIDDEN**");
                copy.put(SaslConfigs.SASL_JAAS_CONFIG, jaasConfig);
            }
        }

        // Return copy of the map.
        return copy;
    }

    @Autowired public  SensitiveConfigScrubber(final SaslUtility saslUtility);

}