How can i unit test a Kstream with an Kafka Binder?












0














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?










share|improve this question
























  • Is Kafka Streams binder on the class path? If you can share a small sample project on Github, we can take a look.
    – sobychacko
    Nov 13 '18 at 23:01










  • Yes kafka Streams binder is on the classpath in version 2.0.1.Release. Overall i am using springbootVersion 2.0.6. I can not share a sample project, i would have to alter almost everthying. Any more clues where i can start to look?
    – axelDiterta
    Nov 14 '18 at 6:27


















0














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?










share|improve this question
























  • Is Kafka Streams binder on the class path? If you can share a small sample project on Github, we can take a look.
    – sobychacko
    Nov 13 '18 at 23:01










  • Yes kafka Streams binder is on the classpath in version 2.0.1.Release. Overall i am using springbootVersion 2.0.6. I can not share a sample project, i would have to alter almost everthying. Any more clues where i can start to look?
    – axelDiterta
    Nov 14 '18 at 6:27
















0












0








0







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?










share|improve this question















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?







spring junit apache-kafka apache-kafka-streams spring-cloud-stream






share|improve this question















share|improve this question













share|improve this question




share|improve this question








edited Nov 13 '18 at 14:45









cricket_007

79.5k1142109




79.5k1142109










asked Nov 13 '18 at 6:30









axelDiterta

245




245












  • Is Kafka Streams binder on the class path? If you can share a small sample project on Github, we can take a look.
    – sobychacko
    Nov 13 '18 at 23:01










  • Yes kafka Streams binder is on the classpath in version 2.0.1.Release. Overall i am using springbootVersion 2.0.6. I can not share a sample project, i would have to alter almost everthying. Any more clues where i can start to look?
    – axelDiterta
    Nov 14 '18 at 6:27




















  • Is Kafka Streams binder on the class path? If you can share a small sample project on Github, we can take a look.
    – sobychacko
    Nov 13 '18 at 23:01










  • Yes kafka Streams binder is on the classpath in version 2.0.1.Release. Overall i am using springbootVersion 2.0.6. I can not share a sample project, i would have to alter almost everthying. Any more clues where i can start to look?
    – axelDiterta
    Nov 14 '18 at 6:27


















Is Kafka Streams binder on the class path? If you can share a small sample project on Github, we can take a look.
– sobychacko
Nov 13 '18 at 23:01




Is Kafka Streams binder on the class path? If you can share a small sample project on Github, we can take a look.
– sobychacko
Nov 13 '18 at 23:01












Yes kafka Streams binder is on the classpath in version 2.0.1.Release. Overall i am using springbootVersion 2.0.6. I can not share a sample project, i would have to alter almost everthying. Any more clues where i can start to look?
– axelDiterta
Nov 14 '18 at 6:27






Yes kafka Streams binder is on the classpath in version 2.0.1.Release. Overall i am using springbootVersion 2.0.6. I can not share a sample project, i would have to alter almost everthying. Any more clues where i can start to look?
– axelDiterta
Nov 14 '18 at 6:27














1 Answer
1






active

oldest

votes


















0














I missed to inject my Channels Interface as a MockBean. After i did that everything worked as expected.






share|improve this answer





















    Your Answer






    StackExchange.ifUsing("editor", function () {
    StackExchange.using("externalEditor", function () {
    StackExchange.using("snippets", function () {
    StackExchange.snippets.init();
    });
    });
    }, "code-snippets");

    StackExchange.ready(function() {
    var channelOptions = {
    tags: "".split(" "),
    id: "1"
    };
    initTagRenderer("".split(" "), "".split(" "), channelOptions);

    StackExchange.using("externalEditor", function() {
    // Have to fire editor after snippets, if snippets enabled
    if (StackExchange.settings.snippets.snippetsEnabled) {
    StackExchange.using("snippets", function() {
    createEditor();
    });
    }
    else {
    createEditor();
    }
    });

    function createEditor() {
    StackExchange.prepareEditor({
    heartbeatType: 'answer',
    autoActivateHeartbeat: false,
    convertImagesToLinks: true,
    noModals: true,
    showLowRepImageUploadWarning: true,
    reputationToPostImages: 10,
    bindNavPrevention: true,
    postfix: "",
    imageUploader: {
    brandingHtml: "Powered by u003ca class="icon-imgur-white" href="https://imgur.com/"u003eu003c/au003e",
    contentPolicyHtml: "User contributions licensed under u003ca href="https://creativecommons.org/licenses/by-sa/3.0/"u003ecc by-sa 3.0 with attribution requiredu003c/au003e u003ca href="https://stackoverflow.com/legal/content-policy"u003e(content policy)u003c/au003e",
    allowUrls: true
    },
    onDemand: true,
    discardSelector: ".discard-answer"
    ,immediatelyShowMarkdownHelp:true
    });


    }
    });














    draft saved

    draft discarded


















    StackExchange.ready(
    function () {
    StackExchange.openid.initPostLogin('.new-post-login', 'https%3a%2f%2fstackoverflow.com%2fquestions%2f53275071%2fhow-can-i-unit-test-a-kstream-with-an-kafka-binder%23new-answer', 'question_page');
    }
    );

    Post as a guest















    Required, but never shown

























    1 Answer
    1






    active

    oldest

    votes








    1 Answer
    1






    active

    oldest

    votes









    active

    oldest

    votes






    active

    oldest

    votes









    0














    I missed to inject my Channels Interface as a MockBean. After i did that everything worked as expected.






    share|improve this answer


























      0














      I missed to inject my Channels Interface as a MockBean. After i did that everything worked as expected.






      share|improve this answer
























        0












        0








        0






        I missed to inject my Channels Interface as a MockBean. After i did that everything worked as expected.






        share|improve this answer












        I missed to inject my Channels Interface as a MockBean. After i did that everything worked as expected.







        share|improve this answer












        share|improve this answer



        share|improve this answer










        answered Nov 23 '18 at 13:05









        axelDiterta

        245




        245






























            draft saved

            draft discarded




















































            Thanks for contributing an answer to Stack Overflow!


            • Please be sure to answer the question. Provide details and share your research!

            But avoid



            • Asking for help, clarification, or responding to other answers.

            • Making statements based on opinion; back them up with references or personal experience.


            To learn more, see our tips on writing great answers.





            Some of your past answers have not been well-received, and you're in danger of being blocked from answering.


            Please pay close attention to the following guidance:


            • Please be sure to answer the question. Provide details and share your research!

            But avoid



            • Asking for help, clarification, or responding to other answers.

            • Making statements based on opinion; back them up with references or personal experience.


            To learn more, see our tips on writing great answers.




            draft saved


            draft discarded














            StackExchange.ready(
            function () {
            StackExchange.openid.initPostLogin('.new-post-login', 'https%3a%2f%2fstackoverflow.com%2fquestions%2f53275071%2fhow-can-i-unit-test-a-kstream-with-an-kafka-binder%23new-answer', 'question_page');
            }
            );

            Post as a guest















            Required, but never shown





















































            Required, but never shown














            Required, but never shown












            Required, but never shown







            Required, but never shown

































            Required, but never shown














            Required, but never shown












            Required, but never shown







            Required, but never shown







            Popular posts from this blog

            Xamarin.iOS Cant Deploy on Iphone

            Glorious Revolution

            Dulmage-Mendelsohn matrix decomposition in Python