Issue
i want to unit test an Kafka Stream Aggregate and i am totally confused which method to use. I read about the TestSupportBinder but i do not think this is working in my case, therefore i use the KafkaEmbedded Method. This is how i initialize the embedded Kafka.
@Before
public void setUp() throws Exception{
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("group-id", "false", embeddedKafka);
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
DefaultKafkaConsumerFactory<Object, LoggerMessage> cf = new DefaultKafkaConsumerFactory<>(consumerProps);
consumer = cf.createConsumer();
embeddedKafka.consumeFromAnEmbeddedTopic(consumer, OUTPUT_TOPIC);
}
What i want to test is the following:
public interface Channels {
String LOGGER_IN_STREAM = "logger-topic-in-stream";
String LOGGER_IN = "logger-topic-in";
String LOGGERDATAVALIDATED_OUT = "loggerDataValidated-topic-out";
@Input(Channels.LOGGER_IN)
SubscribableChannel processMessage();
@Input(Channels.LOGGER_IN_STREAM)
KStream<Object, LoggerMessage> loggerKstreamIn();
@Output(Channels.LOGGERDATAVALIDATED_OUT)
MessageChannel validateLoggerData();
}
And i get following error message
org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'some.domain.Channels': Invocation of init method failed; nested exception is java.lang.IllegalStateException: No factory found for binding target type: org.apache.kafka.streams.kstream.KStream among registered factories: channelFactory,messageSourceFactory
Caused by: java.lang.IllegalStateException: No factory found for binding target type: org.apache.kafka.streams.kstream.KStream among registered factories: channelFactory,messageSourceFactory
What am i doing wrong?
Solution
I missed to inject my Channels Interface as a MockBean. After i did that everything worked as expected.
Answered By - axelDiterta