Ytukyg

I'm creating a microservice using JHipster and I'm facing an strange problem when trying receive a message from kafka. I'm using 3 computer on internal LAN to make this test:

• Kafka server in 192.168.0.200


• Gateway microservice in 192.168.0.7


• Blog microservice in 192.168.0.9

When I change a user in gateway microservice, it can send the message correctly to kafka, and one second later I receive the following error in blog microservice:

2018-11-23 09:33:22.628 ERROR 21063 --- [container-0-C-1] o.s.integration.handler.LoggingHandler   : org.springframework.messaging.MessageDeliveryException: Dispatcher has no subscribers for channel 'blog-1.userChannel'.; nested exception is org.springframework.integration.MessageDispatchingException: Dispatcher has no subscribers, failedMessage=GenericMessage [payload=byte[188], headers={kafka_offset=3, scst_nativeHeadersPresent=true, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@33bfc8cb, deliveryAttempt=3, kafka_timestampType=CREATE_TIME, kafka_receivedMessageKey=null, kafka_receivedPartitionId=0, contentType=application/json, kafka_receivedTopic=user-data-change, kafka_receivedTimestamp=1542972799625}], failedMessage=GenericMessage [payload=byte[188], headers={kafka_offset=3, scst_nativeHeadersPresent=true, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@33bfc8cb, deliveryAttempt=3, kafka_timestampType=CREATE_TIME, kafka_receivedMessageKey=null, kafka_receivedPartitionId=0, contentType=application/json, kafka_receivedTopic=user-data-change, kafka_receivedTimestamp=1542972799625}]

at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:77)

at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:445)

at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:394)

at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:181)

at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:160)

at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:47)

at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:108)

at org.springframework.integration.endpoint.MessageProducerSupport.sendMessage(MessageProducerSupport.java:203)

at org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter.access$300(KafkaMessageDrivenChannelAdapter.java:70)

at org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter$IntegrationRecordMessageListener.onMessage(KafkaMessageDrivenChannelAdapter.java:387)

at org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter$IntegrationRecordMessageListener.onMessage(KafkaMessageDrivenChannelAdapter.java:364)

at org.springframework.kafka.listener.adapter.RetryingMessageListenerAdapter.lambda$onMessage$0(RetryingMessageListenerAdapter.java:120)

at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:287)

at org.springframework.retry.support.RetryTemplate.execute(RetryTemplate.java:211)

at org.springframework.kafka.listener.adapter.RetryingMessageListenerAdapter.onMessage(RetryingMessageListenerAdapter.java:114)

at org.springframework.kafka.listener.adapter.RetryingMessageListenerAdapter.onMessage(RetryingMessageListenerAdapter.java:40)

at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeRecordListener(KafkaMessageListenerContainer.java:1077)

at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeWithRecords(KafkaMessageListenerContainer.java:1057)

at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeRecordListener(KafkaMessageListenerContainer.java:999)

at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeListener(KafkaMessageListenerContainer.java:867)

at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:725)

at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)

at java.util.concurrent.FutureTask.run$$$capture(FutureTask.java:266)

at java.util.concurrent.FutureTask.run(FutureTask.java)

at java.lang.Thread.run(Thread.java:748)

Caused by: org.springframework.integration.MessageDispatchingException: Dispatcher has no subscribers, failedMessage=GenericMessage [payload=byte[188], headers={kafka_offset=3, scst_nativeHeadersPresent=true, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@33bfc8cb, deliveryAttempt=3, kafka_timestampType=CREATE_TIME, kafka_receivedMessageKey=null, kafka_receivedPartitionId=0, contentType=application/json, kafka_receivedTopic=user-data-change, kafka_receivedTimestamp=1542972799625}]

at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:138)

at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:105)

at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:73)

... 24 more

This is my kafka configuration:

spring

    cloud:

        stream:

            kafka:

                binder:

                    brokers: 192.168.0.200

                    zk-nodes: 192.168.0.200

                bindings:

                    gatewayChannel:

                        consumer:

                            enableDlq: true

                            dlqName: unprocessed

                            autoCommitOnError: true

                            autoCommitOffset: true

                    userChannel:

                        consumer:

                            enableDlq: true

                            dlqName: unprocessed

                            autoCommitOnError: true

                            autoCommitOffset: true

            bindings:

                gatewayChannel:

                    group: blog

                    destination: schema

                    content-type: application/json

                userChannel:

                    group: blog

                    destination: user-data-change

                    content-type: application/json

And these are the classes that consumes the messages:

@EnableBinding(value = {ConsumerChannel.class})

public class MessagingConfiguration {



    private final CustomLiquibaseService customLiquibaseService;



    public MessagingConfiguration(CustomLiquibaseService customLiquibaseService) {

        this.customLiquibaseService = customLiquibaseService;

    }



    @StreamListener("gatewayChannel")

    public void createSchema(DefaultMessage message) {

        customLiquibaseService.createSchema(message.getMessage());

    }



    @StreamListener("userChannel")

    public void processUser(Object message) throws Exception {

        throw new Exception("test");

    }

}



public interface ConsumerChannel {



    @Input

    SubscribableChannel gatewayChannel();



    @Input

    SubscribableChannel userChannel();

}

As I this error only happens when gateway microservice add a message to topic, I suspect that my microservice is listening but can't process the message. What am I doing wrong?