@SuppressWarnings("unchecked")
    static Map<String, Object> channelBuilderConfigs(final AbstractConfig config, final ListenerName listenerName) {
        Map<String, Object> parsedConfigs;
        if (listenerName == null)
            parsedConfigs = (Map<String, Object>) config.values();
        else
            parsedConfigs = config.valuesWithPrefixOverride(listenerName.configPrefix());

        config.originals().entrySet().stream()
            .filter(e -> !parsedConfigs.containsKey(e.getKey())) // exclude already parsed configs
            // exclude already parsed listener prefix configs
            .filter(e -> !(listenerName != null && e.getKey().startsWith(listenerName.configPrefix()) &&
                parsedConfigs.containsKey(e.getKey().substring(listenerName.configPrefix().length()))))
            // exclude keys like `{mechanism}.some.prop` if "listener.name." prefix is present and key `some.prop` exists in parsed configs.
            .filter(e -> !(listenerName != null && parsedConfigs.containsKey(e.getKey().substring(e.getKey().indexOf('.') + 1))))
            .forEach(e -> parsedConfigs.put(e.getKey(), e.getValue()));
        return parsedConfigs;
    }
@Test
    public void testChannelBuilderConfigs() {
        Properties props = new Properties();
        props.put("listener.name.listener1.gssapi.sasl.kerberos.service.name", "testkafka");
        props.put("listener.name.listener1.sasl.kerberos.service.name", "testkafkaglobal");
        props.put("plain.sasl.server.callback.handler.class", "callback");
        props.put("listener.name.listener1.gssapi.config1.key", "custom.config1");
        props.put("custom.config2.key", "custom.config2");
        TestSecurityConfig securityConfig = new TestSecurityConfig(props);

        // test configs with listener prefix
        Map<String, Object> configs = ChannelBuilders.channelBuilderConfigs(securityConfig, new ListenerName("listener1"));

        assertNull(configs.get("listener.name.listener1.gssapi.sasl.kerberos.service.name"));
        assertFalse(securityConfig.unused().contains("listener.name.listener1.gssapi.sasl.kerberos.service.name"));

        assertEquals(configs.get("gssapi.sasl.kerberos.service.name"), "testkafka");
        assertFalse(securityConfig.unused().contains("gssapi.sasl.kerberos.service.name"));

        assertEquals(configs.get("sasl.kerberos.service.name"), "testkafkaglobal");
        assertFalse(securityConfig.unused().contains("sasl.kerberos.service.name"));

        assertNull(configs.get("listener.name.listener1.sasl.kerberos.service.name"));
        assertFalse(securityConfig.unused().contains("listener.name.listener1.sasl.kerberos.service.name"));

        assertNull(configs.get("plain.sasl.server.callback.handler.class"));
        assertFalse(securityConfig.unused().contains("plain.sasl.server.callback.handler.class"));

        assertEquals(configs.get("listener.name.listener1.gssapi.config1.key"), "custom.config1");
        assertFalse(securityConfig.unused().contains("listener.name.listener1.gssapi.config1.key"));

        assertEquals(configs.get("custom.config2.key"), "custom.config2");
        assertFalse(securityConfig.unused().contains("custom.config2.key"));

        // test configs without listener prefix
        securityConfig = new TestSecurityConfig(props);
        configs = ChannelBuilders.channelBuilderConfigs(securityConfig, null);

        assertEquals(configs.get("listener.name.listener1.gssapi.sasl.kerberos.service.name"), "testkafka");
        assertFalse(securityConfig.unused().contains("listener.name.listener1.gssapi.sasl.kerberos.service.name"));

        assertNull(configs.get("gssapi.sasl.kerberos.service.name"));
        assertFalse(securityConfig.unused().contains("gssapi.sasl.kerberos.service.name"));

        assertEquals(configs.get("listener.name.listener1.sasl.kerberos.service.name"), "testkafkaglobal");
        assertFalse(securityConfig.unused().contains("listener.name.listener1.sasl.kerberos.service.name"));

        assertNull(configs.get("sasl.kerberos.service.name"));
        assertFalse(securityConfig.unused().contains("sasl.kerberos.service.name"));

        assertEquals(configs.get("plain.sasl.server.callback.handler.class"), "callback");
        assertFalse(securityConfig.unused().contains("plain.sasl.server.callback.handler.class"));

        assertEquals(configs.get("listener.name.listener1.gssapi.config1.key"), "custom.config1");
        assertFalse(securityConfig.unused().contains("listener.name.listener1.gssapi.config1.key"));

        assertEquals(configs.get("custom.config2.key"), "custom.config2");
        assertFalse(securityConfig.unused().contains("custom.config2.key"));
    }