static Map<String, Object> appendSerializerToConfig(Map<String, Object> configs,
            Serializer<?> keySerializer,
            Serializer<?> valueSerializer) {
        // validate serializer configuration, if the passed serializer instance is null, the user must explicitly set a valid serializer configuration value
        Map<String, Object> newConfigs = new HashMap<>(configs);
        if (keySerializer != null)
            newConfigs.put(KEY_SERIALIZER_CLASS_CONFIG, keySerializer.getClass());
        else if (newConfigs.get(KEY_SERIALIZER_CLASS_CONFIG) == null)
            throw new ConfigException(KEY_SERIALIZER_CLASS_CONFIG, null, "must be non-null.");
        if (valueSerializer != null)
            newConfigs.put(VALUE_SERIALIZER_CLASS_CONFIG, valueSerializer.getClass());
        else if (newConfigs.get(VALUE_SERIALIZER_CLASS_CONFIG) == null)
            throw new ConfigException(VALUE_SERIALIZER_CLASS_CONFIG, null, "must be non-null.");
        return newConfigs;
    }
@Test
    public void testAppendSerializerToConfigWithException() {
        Map<String, Object> configs = new HashMap<>();
        configs.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, null);
        configs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, valueSerializerClass);
        assertThrows(ConfigException.class, () -> ProducerConfig.appendSerializerToConfig(configs, null, valueSerializer));

        configs.clear();
        configs.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, keySerializerClass);
        configs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, null);
        assertThrows(ConfigException.class, () -> ProducerConfig.appendSerializerToConfig(configs, keySerializer, null));
    }