import org.apache.kafka.common.Configurable; //导入依赖的package包/类
/**
 * Get a list of configured instances of the given class specified by the given configuration key. The configuration
 * may specify either null or an empty string to indicate no configured instances. In both cases, this method
 * returns an empty list to indicate no configured instances.
 * @param key The configuration key for the class
 * @param t The interface the class should implement
 * @return The list of configured instances
 */
public <T> List<T> getConfiguredInstances(String key, Class<T> t) {
    List<String> klasses = getList(key);
    List<T> objects = new ArrayList<T>();
    if (klasses == null)
        return objects;
    for (String klass : klasses) {
        Object o;
        try {
            o = Utils.newInstance(klass, t);
        } catch (ClassNotFoundException e) {
            throw new KafkaException(klass + " ClassNotFoundException exception occured", e);
        }
        if (!t.isInstance(o))
            throw new KafkaException(klass + " is not an instance of " + t.getName());
        if (o instanceof Configurable)
            ((Configurable) o).configure(this.originals);
        objects.add(t.cast(o));
    }
    return objects;
}
 

import org.apache.kafka.common.Configurable; //导入依赖的package包/类
public <T> T getConfiguredInstance(String key, Class<T> t, Producer<byte[], byte[]> producer) {
  Class<?> c = getClass(key);
  if (c == null) {
    return null;
  }
  Object o = Utils.newInstance(c);

  if (!t.isInstance(o)) {
    throw new KafkaException(c.getName() + " is not an instance of " + t.getName());
  }

  if (o instanceof Configurable) {
    ((Configurable) o).configure(configsWithCurrentProducer(producer));
  }

  return t.cast(o);
}
 

import org.apache.kafka.common.Configurable; //导入依赖的package包/类
protected static <T> T newConfiguredPlugin(AbstractConfig config, Class<T> klass) {
    T plugin = Utils.newInstance(klass);
    if (plugin instanceof Configurable) {
        ((Configurable) plugin).configure(config.originals());
    }
    return plugin;
}
 

import org.apache.kafka.common.Configurable; //导入依赖的package包/类
/**
 * Get a configured instance of the give class specified by the given configuration key. If the object implements
 * Configurable configure it using the configuration.
 *
 * @param key The configuration key for the class
 * @param t The interface the class should implement
 * @return A configured instance of the class
 */
public <T> T getConfiguredInstance(String key, Class<T> t) {
    Class<?> c = getClass(key);
    if (c == null)
        return null;
    Object o = Utils.newInstance(c);
    if (!t.isInstance(o))
        throw new KafkaException(c.getName() + " is not an instance of " + t.getName());
    if (o instanceof Configurable)
        ((Configurable) o).configure(originals());
    return t.cast(o);
}
 

import org.apache.kafka.common.Configurable; //导入依赖的package包/类
/**
 * Get a list of configured instances of the given class specified by the given configuration key. The configuration
 * may specify either null or an empty string to indicate no configured instances. In both cases, this method
 * returns an empty list to indicate no configured instances.
 * @param key The configuration key for the class
 * @param t The interface the class should implement
 * @param configOverrides Configuration overrides to use.
 * @return The list of configured instances
 */
public <T> List<T> getConfiguredInstances(String key, Class<T> t, Map<String, Object> configOverrides) {
    List<String> klasses = getList(key);
    List<T> objects = new ArrayList<T>();
    if (klasses == null)
        return objects;
    Map<String, Object> configPairs = originals();
    configPairs.putAll(configOverrides);
    for (Object klass : klasses) {
        Object o;
        if (klass instanceof String) {
            try {
                o = Utils.newInstance((String) klass, t);
            } catch (ClassNotFoundException e) {
                throw new KafkaException(klass + " ClassNotFoundException exception occurred", e);
            }
        } else if (klass instanceof Class<?>) {
            o = Utils.newInstance((Class<?>) klass);
        } else
            throw new KafkaException("List contains element of type " + klass.getClass().getName() + ", expected String or Class");
        if (!t.isInstance(o))
            throw new KafkaException(klass + " is not an instance of " + t.getName());
        if (o instanceof Configurable)
            ((Configurable) o).configure(configPairs);
        objects.add(t.cast(o));
    }
    return objects;
}
 

import org.apache.kafka.common.Configurable; //导入依赖的package包/类
/**
 * Get a configured instance of the give class specified by the given configuration key. If the object implements
 * Configurable configure it using the configuration.
 *
 * @param key The configuration key for the class
 * @param t The interface the class should implement
 * @return A configured instance of the class
 */
public <T> T getConfiguredInstance(String key, Class<T> t) {
    Class<?> c = getClass(key);
    if (c == null)
        return null;
    Object o = Utils.newInstance(c);
    if (!t.isInstance(o))
        throw new KafkaException(c.getName() + " is not an instance of " + t.getName());
    if (o instanceof Configurable)
        ((Configurable) o).configure(this.originals);
    return t.cast(o);
}
 

import org.apache.kafka.common.Configurable; //导入依赖的package包/类
@Override
@SuppressWarnings("unchecked")
protected Binding<KStream<Object, Object>> doBindProducer(String name, KStream<Object, Object> outboundBindTarget,
														ExtendedProducerProperties<KStreamProducerProperties> properties) {
	ExtendedProducerProperties<KafkaProducerProperties> extendedProducerProperties = new ExtendedProducerProperties<KafkaProducerProperties>(
			new KafkaProducerProperties());
	this.kafkaTopicProvisioner.provisionProducerDestination(name, extendedProducerProperties);
	outboundBindTarget = outboundBindTarget
			.map((k, v) -> KeyValue.pair(k, ((Message<Object>) v).getPayload()));

	Serde<?> keySerde = Serdes.ByteArray();
	Serde<?> valueSerde = Serdes.ByteArray();
	if (properties.isUseNativeEncoding()) {
		outboundBindTarget.to(name, Produced.with((Serde<Object>) keySerde, (Serde<Object>) valueSerde));
	}
	else {
		try {
			if (StringUtils.hasText(properties.getExtension().getKeySerde())) {
				keySerde = Utils.newInstance(properties.getExtension().getKeySerde(), Serde.class);
				if (keySerde instanceof Configurable) {
					((Configurable) keySerde).configure(streamsConfig.originals());
				}
			}
			else {
				keySerde = this.binderConfigurationProperties.getConfiguration().containsKey("key.serde") ?
						Utils.newInstance(this.binderConfigurationProperties.getConfiguration().get("key.serde"), Serde.class) : Serdes.ByteArray();
			}

			if (StringUtils.hasText(properties.getExtension().getValueSerde())) {
				valueSerde = Utils.newInstance(properties.getExtension().getValueSerde(), Serde.class);
				if (valueSerde instanceof Configurable) {
					((Configurable) valueSerde).configure(streamsConfig.originals());
				}
			}
			outboundBindTarget.to(name, Produced.with((Serde<Object>) keySerde, (Serde<Object>) valueSerde));
		} catch (ClassNotFoundException e) {
			throw new IllegalStateException("Serde class not found: ", e);
		}
	}
	return new DefaultBinding<>(name, null, outboundBindTarget, null);
}