Repository: apache/kafka
File: test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/ClusterInstance.java (Java)

Code before:

    default <K, V> Consumer<K, V> consumer(Map<String, Object> configs) {
        Map<String, Object> props = new HashMap<>(configs);
        props.putIfAbsent(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
        props.putIfAbsent(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
        props.putIfAbsent(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.putIfAbsent(ConsumerConfig.GROUP_ID_CONFIG, "group_" + TestUtils.randomString(5));
        props.putIfAbsent(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers());
        return new KafkaConsumer<>(setClientSaslConfig(props));
    }

    default <K, V> Consumer<K, V> consumer() {
        return consumer(Map.of());
    }

    default <K, V> ShareConsumer<K, V> shareConsumer() {
        return shareConsumer(Map.of());
    }

    default <K, V> ShareConsumer<K, V> shareConsumer(Map<String, Object> configs) {
        return new KafkaShareConsumer<>(configs, null, null);
    }

    default <K, V> ShareConsumer<K, V> shareConsumer(Map<String, Object> configs, Deserializer<K> keyDeserializer, Deserializer<V> valueDeserializer) {
        Map<String, Object> props = new HashMap<>(configs);
        props.putIfAbsent(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
        props.putIfAbsent(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
        props.putIfAbsent(ConsumerConfig.GROUP_ID_CONFIG, "group_" + TestUtils.randomString(5));
        props.putIfAbsent(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers());
        return new KafkaShareConsumer<>(setClientSaslConfig(props), keyDeserializer, valueDeserializer);
    }

    default Admin admin(Map<String, Object> configs, boolean usingBootstrapControllers) {
        Map<String, Object> props = new HashMap<>(configs);
        if (usingBootstrapControllers) {
            props.putIfAbsent(AdminClientConfig.BOOTSTRAP_CONTROLLERS_CONFIG, bootstrapControllers());
            props.remove(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG);
        } else {
            props.putIfAbsent(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers());
            props.remove(AdminClientConfig.BOOTSTRAP_CONTROLLERS_CONFIG);
        }
        return Admin.create(setClientSaslConfig(props));
    }

    default Map<String, Object> setClientSaslConfig(Map<String, Object> configs) {
        Map<String, Object> props = new HashMap<>(configs);
        if (config().brokerSecurityProtocol() == SecurityProtocol.SASL_PLAINTEXT) {
            props.putIfAbsent(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, SecurityProtocol.SASL_PLAINTEXT.name);
            props.putIfAbsent(SaslConfigs.SASL_MECHANISM, "PLAIN");
            props.putIfAbsent(
                SaslConfigs.SASL_JAAS_CONFIG,

Review comment (suggestion):
If you're providing instances of the deserializers, I don't think you should be setting the class names in the config.

Code after:

    default <K, V> Consumer<K, V> consumer(Map<String, Object> configs) {
        Map<String, Object> props = new HashMap<>(configs);
        props.putIfAbsent(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
        props.putIfAbsent(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
        props.putIfAbsent(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.putIfAbsent(ConsumerConfig.GROUP_ID_CONFIG, "group_" + TestUtils.randomString(5));
        props.putIfAbsent(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers());
        return new KafkaConsumer<>(setClientSaslConfig(props));
    }

    default <K, V> Consumer<K, V> consumer() {
        return consumer(Map.of());
    }

    default <K, V> ShareConsumer<K, V> shareConsumer() {
        return shareConsumer(Map.of());
    }

    default <K, V> ShareConsumer<K, V> shareConsumer(Map<String, Object> configs) {
        return shareConsumer(configs, null, null);
    }

    default <K, V> ShareConsumer<K, V> shareConsumer(Map<String, Object> configs, Deserializer<K> keyDeserializer, Deserializer<V> valueDeserializer) {
        Map<String, Object> props = new HashMap<>(configs);
        if (keyDeserializer == null) {
            props.putIfAbsent(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
        }
        if (valueDeserializer == null) {
            props.putIfAbsent(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
        }
        props.putIfAbsent(ConsumerConfig.GROUP_ID_CONFIG, "group_" + TestUtils.randomString(5));
        props.putIfAbsent(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers());
        return new KafkaShareConsumer<>(setClientSaslConfig(props), keyDeserializer, valueDeserializer);
    }

    default Admin admin(Map<String, Object> configs, boolean usingBootstrapControllers) {
        Map<String, Object> props = new HashMap<>(configs);
        if (usingBootstrapControllers) {
            props.putIfAbsent(AdminClientConfig.BOOTSTRAP_CONTROLLERS_CONFIG, bootstrapControllers());
            props.remove(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG);
        } else {
            props.putIfAbsent(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers());
            props.remove(AdminClientConfig.BOOTSTRAP_CONTROLLERS_CONFIG);
        }
        return Admin.create(setClientSaslConfig(props));
    }

    default Map<String, Object> setClientSaslConfig(Map<String, Object> configs) {
        Map<String, Object> props = new HashMap<>(configs);
        if (config().brokerSecurityProtocol() == SecurityProtocol.SASL_PLAINTEXT) {