...
Code Block | ||||
---|---|---|---|---|
| ||||
public class SpringAwareStreamsConfig extends StreamsConfig { private final List<ProducerInterceptor<?, ?>> producerInterceptors; private final List<ConsumerInterceptor<?, ?>> consumerInterceptors; public SpringAwareStreamsConfig(Map<?, ?> props, boolean doLog, List<ProducerInterceptor<?, ?>> producerInterceptors, List<ConsumerInterceptor<?, ?>> consumerInterceptors) { super(props, doLog); this.producerInterceptors = producerInterceptors; this.consumerInterceptors = consumerInterceptors; } @Override public <T> List<T> getConfiguredInstances(List<String> classNames, Class<T> t, Map<String, Object> configOverrides) { final var configuredInstances = super.getConfiguredInstances(classNames, t, configOverrides); if (ProducerInterceptor.class.equals(t)) { configuredInstances.addAll((Collection<? extends T>) producerInterceptors); } if (ConsumerInterceptor.class.equals(t)) { configuredInstances.addAll((Collection<? extends T>) consumerInterceptors); } return configuredInstances; } } |
Streams may also use another approach based on a custom KafkaClientSupplier and reusing this KIP
Code Block | ||||
---|---|---|---|---|
| ||||
public class CustomKafkaClientSupplier extends DefaultKafkaClientSupplier {
// additional dependencies given by constructor injection.
@Override
public Producer<byte[], byte[]> getProducer(final Map<String, Object> config) {
return new KafkaProducer<>(<CUSTOM_CONFIG>, new ByteArraySerializer(), new ByteArraySerializer());
}
@Override
public Consumer<byte[], byte[]> getConsumer(final Map<String, Object> config) {
return new KafkaConsumer<>(<CUSTOM_CONFIG>, new ByteArrayDeserializer(), new ByteArrayDeserializer());
}
@Override
public Consumer<byte[], byte[]> getRestoreConsumer(final Map<String, Object> config) {
return new KafkaConsumer<>(<CUSTOM_CONFIG>, new ByteArrayDeserializer(), new ByteArrayDeserializer());
}
@Override
public Consumer<byte[], byte[]> getGlobalConsumer(final Map<String, Object> config) {
return new KafkaConsumer<>(<CUSTOM_CONFIG>, new ByteArrayDeserializer(), new ByteArrayDeserializer());
}
}
...
new KafkaStreams(<TOPOLOGY>, <PROPERTIES>, <CUSTOM_KAFKA_CLIENT_SUPPLIER>);
... |
See another similar KIP for additional information related to the current KIP but for streams.
...