search

spring-cloud / spring-cloud-stream

Framework for building Event-Driven Microservices
http://cloud.spring.io/spring-cloud-stream
Apache License 2.0
998 stars 609 forks source link

Consumer not consuming Object #1201

Closed pavankjadda closed 6 years ago

pavankjadda commented 6 years ago

I setup microservices with docker. My application has one microservice and gateway. Selected Kafka option when I setup microservice. Here is the problem. I created new package inside microservice to send an object from producer and receive it on the consumer. Both producer and consumer present in the same package. I am able to send the object from the producer based on logs, but the receiver is not receiving it. I checked Staack overflow and GitHub issues, nothing helped me so far. microservice, gateway, kafka, zookeeper run as different containers in docker

ProducerChannel.java

package com.projects.messaging;

import org.springframework.cloud.stream.annotation.Output;
import org.springframework.messaging.MessageChannel;

public interface ProducerChannel
{

    String CHANNEL = "messageChannel";

    @Output
    MessageChannel messageChannel();
}

ProducerResource.java

package com.projects.messaging;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

import com.projects.domain.Project;

@Service
public class ProducerResource
{

    private MessageChannel channel;

    @Autowired
    public ProducerResource(ProducerChannel channel)
    {
        System.out.println("***************************************** Inside ProducerResource Constructor  *****************************************");
        this.channel = channel.messageChannel();
    }

    public void produce(Project project)
    {
        System.out.println("***************************************** Inside produce method  *****************************************");
        System.out.println("***************************************** project: "+project.getProjectName());
        Integer count=10;

        System.out.println("***************************************** Inside produce method  while loop*****************************************");
        channel.send(MessageBuilder.withPayload("Project Sending: "+project.toString()).build());

    }

}

ConsumerChannel.java

package com.projects.messaging;

import org.springframework.cloud.stream.annotation.Input;
import org.springframework.messaging.SubscribableChannel;

public interface ConsumerChannel
{

    String CHANNEL = "subscribableChannel";

    @Input
    SubscribableChannel subscribableChannel();
}

ConsumerService.java

package com.projects.messaging;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.stereotype.Service;

import com.projects.domain.Project;

//@Service
@EnableBinding(ConsumerChannel.class)
public class ConsumerService
{

    private final Logger log = LoggerFactory.getLogger(ConsumerService.class);

    @StreamListener(ConsumerChannel.CHANNEL)
    public void consume(Project project)
    {
        log.info("Received Project Info: {}.", project.toString());
    }
}

ProjectResource.java : whenever new object created/updated, producer sends message to consumer

package com.projects.web.rest;

import com.codahale.metrics.annotation.Timed;
import com.projects.domain.Project;
import com.projects.messaging.*;

import com.projects.repository.ProjectRepository;
import com.projects.web.rest.errors.BadRequestAlertException;
import com.projects.web.rest.util.HeaderUtil;
import com.projects.web.rest.util.PaginationUtil;
import io.github.jhipster.web.util.ResponseUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.Pageable;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.messaging.MessageChannel;
import org.springframework.web.bind.annotation.*;

import javax.validation.Valid;
import java.net.URI;
import java.net.URISyntaxException;

import java.util.List;
import java.util.Optional;

/**
 * REST controller for managing Project.
 */
@RestController
@RequestMapping("/api")
public class ProjectResource
{

    private final Logger log = LoggerFactory.getLogger(ProjectResource.class);

    private static final String ENTITY_NAME = "project";

    private final ProjectRepository projectRepository;

    @Autowired
    private ProducerResource producerResource;

    public ProjectResource(ProjectRepository projectRepository)
    {
        this.projectRepository = projectRepository;
    }

    /**
     * POST /projects : Create a new project.
     *
     * @param project
     *            the project to create
     * @return the ResponseEntity with status 201 (Created) and with body the new
     *         project, or with status 400 (Bad Request) if the project has already
     *         an ID
     * @throws URISyntaxException
     *             if the Location URI syntax is incorrect
     */
    @PostMapping("/projects")
    @Timed
    public ResponseEntity<Project> createProject(@Valid @RequestBody Project project) throws URISyntaxException
    {
        log.debug("REST request to save Project : {}", project);
        if (project.getId() != null)
        {
            throw new BadRequestAlertException("A new project cannot already have an ID", ENTITY_NAME, "idexists");
        }
        Project result = projectRepository.save(project);
        producerResource.produce(result);
        return ResponseEntity.created(new URI("/api/projects/" + result.getId()))
                .headers(HeaderUtil.createEntityCreationAlert(ENTITY_NAME, result.getId().toString())).body(result);
    }

    /**
     * PUT /projects : Updates an existing project.
     *
     * @param project
     *            the project to update
     * @return the ResponseEntity with status 200 (OK) and with body the updated
     *         project, or with status 400 (Bad Request) if the project is not
     *         valid, or with status 500 (Internal Server Error) if the project
     *         couldn't be updated
     * @throws URISyntaxException
     *             if the Location URI syntax is incorrect
     */
    @PutMapping("/projects")
    @Timed
    public ResponseEntity<Project> updateProject(@Valid @RequestBody Project project) throws URISyntaxException
    {
        log.debug("REST request to update Project : {}", project);
        if (project.getId() == null)
        {
            return createProject(project);
        }
        Project result = projectRepository.save(project);
        producerResource.produce(result);
        return ResponseEntity.ok().headers(HeaderUtil.createEntityUpdateAlert(ENTITY_NAME, project.getId().toString()))
                .body(result);
    }

}

docker ps command output


CONTAINER ID        IMAGE                               COMMAND                   CREATED             STATUS              PORTS                                  NAMES
647110ce5857        projects                            "/bin/sh -c 'echo \"T…"   15 minutes ago      Up 15 minutes       8081/tcp                               dockercompose_projects-app_1
04624ca664dd        wurstmeister/kafka:1.0.0            "start-kafka.sh"          About an hour ago   Up 21 minutes       0.0.0.0:9092->9092/tcp                 dockercompose_kafka_1
3aa6b7f15ec8        gateway                             "/bin/sh -c 'echo \"T…"   2 hours ago         Up 22 minutes       5701/udp, 0.0.0.0:8080->8080/tcp       dockercompose_gateway-app_1
23d4b241a36f        jhipster/jhipster-registry:v3.2.4   "/bin/sh -c 'java   …"    2 hours ago         Up 22 minutes       0.0.0.0:8761->8761/tcp                 dockercompose_jhipster-registry_1
9c8368f6093d        mysql:5.7.20                        "docker-entrypoint.s…"    2 hours ago         Up 22 minutes       3306/tcp                               dockercompose_projects-mysql_1
3c8658816d0c        mysql:5.7.20                        "docker-entrypoint.s…"    2 hours ago         Up 22 minutes       3306/tcp                               dockercompose_gateway-mysql_1
b610f4c71463        wurstmeister/zookeeper:3.4.6        "/bin/sh -c '/usr/sb…"    2 hours ago         Up 22 minutes       22/tcp, 2181/tcp, 2888/tcp, 3888/tcp   dockercompose_zookeeper_1

Console output when create/update object

gateway-app_1        | 2018-01-30 22:06:58.148  WARN 9 --- [nfoReplicator-0] o.a.k.clients.consumer.ConsumerConfig    : The configuration 'value.serializer' was supplied but isn't a known config.
gateway-app_1        | 2018-01-30 22:06:58.148  WARN 9 --- [nfoReplicator-0] o.a.k.clients.consumer.ConsumerConfig    : The configuration 'key.serializer' was supplied but isn't a known config.
projects-app_1       | 2018-01-30 22:07:04.482  WARN 7 --- [nfoReplicator-0] o.a.k.clients.consumer.ConsumerConfig    : The configuration 'value.serializer' was supplied but isn't a known config.
projects-app_1       | 2018-01-30 22:07:04.483  WARN 7 --- [nfoReplicator-0] o.a.k.clients.consumer.ConsumerConfig    : The configuration 'key.serializer' was supplied but isn't a known config.
gateway-app_1        | 2018-01-30 22:07:08.251  WARN 9 --- [nfoReplicator-0] o.a.k.clients.consumer.ConsumerConfig    : The configuration 'value.serializer' was supplied but isn't a known config.
gateway-app_1        | 2018-01-30 22:07:08.251  WARN 9 --- [nfoReplicator-0] o.a.k.clients.consumer.ConsumerConfig    : The configuration 'key.serializer' was supplied but isn't a known config.
projects-app_1       | ***************************************** Inside produce method  *****************************************
projects-app_1       | ***************************************** project: PostIT
projects-app_1       | ***************************************** Inside produce method  while loop*****************************************
projects-app_1       | 2018-01-30 22:07:14.587  WARN 7 --- [nfoReplicator-0] o.a.k.clients.consumer.ConsumerConfig    : The configuration 'value.serializer' was supplied but isn't a known config.
projects-app_1       | 2018-01-30 22:07:14.587  WARN 7 --- [nfoReplicator-0] o.a.k.clients.consumer.ConsumerConfig    : The configuration 'key.serializer' was supplied but isn't a known config.
gateway-app_1        | 2018-01-30 22:07:18.354  WARN 9 --- [nfoReplicator-0] o.a.k.clients.consumer.ConsumerConfig    : The configuration 'value.serializer' was supplied but isn't a known confi

application-dev.yml

# ===================================================================
# Spring Boot configuration for the "dev" profile.
#
# This configuration overrides the application.yml file.
#
# More information on profiles: http://www.jhipster.tech/profiles/
# More information on configuration properties: http://www.jhipster.tech/common-application-properties/
# ===================================================================

# ===================================================================
# Standard Spring Boot properties.
# Full reference is available at:
# http://docs.spring.io/spring-boot/docs/current/reference/html/common-application-properties.html
# ===================================================================

logging:
    level:
        ROOT: DEBUG
        com.gateway: DEBUG
        io.github.jhipster: DEBUG

eureka:
    instance:
        prefer-ip-address: true
    client:
        service-url:
            defaultZone: http://admin:${jhipster.registry.password}@localhost:8761/eureka/

spring:
    profiles:
        active: dev
        include: swagger
    devtools:
        restart:
            enabled: true
        livereload:
            enabled: false # we use gulp + BrowserSync for livereload
    jackson:
        serialization.indent_output: true
    cloud:
        stream:
            kafka:
                binder:
                    brokers: localhost
                    zk-nodes: localhost
                    auto-add-partitions: true
                    auto-create-topics: true
                    min-partition-count: 1
                    offsetUpdateCount: true
            bindings:
               messageChannel:
                  destination: project
                  group: group1
                  producer:
                      partitionCount: 1
                      partitioned: true     
               subscribableChannel:
                  destination: project
                  group: group1
                  consumer:
                      concurrency: 1
                      partitioned: true
                      startOffset: earliest

    datasource:
        type: com.zaxxer.hikari.HikariDataSource
        url: jdbc:mysql://localhost:3306/gateway?useUnicode=true&characterEncoding=utf8&useSSL=false
        username: root
        password:
        hikari:
            data-source-properties:
                cachePrepStmts: true
                prepStmtCacheSize: 250
                prepStmtCacheSqlLimit: 2048
                useServerPrepStmts: true
    jpa:
        database-platform: org.hibernate.dialect.MySQL5InnoDBDialect
        database: MYSQL
        show-sql: true
        properties:
            hibernate.id.new_generator_mappings: true
            hibernate.cache.use_second_level_cache: false
            hibernate.cache.use_query_cache: false
            hibernate.generate_statistics: true
    mail:
        host: localhost
        port: 25
        username:
        password:
    messages:
        cache-seconds: 1
    thymeleaf:
        cache: false
    zipkin: # Use the "zipkin" Maven profile to have the Spring Cloud Zipkin dependencies
        base-url: http://localhost:9411
        enabled: false
        locator:
            discovery:
                enabled: true

liquibase:
    contexts: dev

# ===================================================================
# To enable SSL, generate a certificate using:
# keytool -genkey -alias gateway -storetype PKCS12 -keyalg RSA -keysize 2048 -keystore keystore.p12 -validity 3650
#
# You can also use Let's Encrypt:
# https://maximilian-boehm.com/hp2121/Create-a-Java-Keystore-JKS-from-Let-s-Encrypt-Certificates.htm
#
# Then, modify the server.ssl properties so your "server" configuration looks like:
#
# server:
#    port: 8443
#    ssl:
#        key-store: keystore.p12
#        key-store-password: <your-password>
#        key-store-type: PKCS12
#        key-alias: gateway
# ===================================================================
server:
    port: 8080

# ===================================================================
# JHipster specific properties
#
# Full reference is available at: http://www.jhipster.tech/common-application-properties/
# ===================================================================

jhipster:
    gateway:
        rate-limiting:
            enabled: false
            limit: 100000
            duration-in-seconds: 3600
        authorized-microservices-endpoints: # Access Control Policy, if left empty for a route, all endpoints will be accessible
            app1: /api,/v2/api-docs # recommended dev configuration
    http:
        version: V_1_1 # To use HTTP/2 you will need SSL support (see above the "server.ssl" configuration)
    cache: # Cache configuration
        hazelcast: # Hazelcast distributed cache
            time-to-live-seconds: 3600
            backup-count: 1
            management-center: # Full reference is available at: http://docs.hazelcast.org/docs/management-center/3.9/manual/html/Deploying_and_Starting.html
                enabled: false
                update-interval: 3
                url: http://localhost:8180/mancenter
    # CORS is only enabled by default with the "dev" profile, so BrowserSync can access the API
    cors:
        allowed-origins: "*"
        allowed-methods: "*"
        allowed-headers: "*"
        exposed-headers: "Authorization,Link,X-Total-Count"
        allow-credentials: true
        max-age: 1800
    security:
        authentication:
            jwt:
                secret: my-secret-token-to-change-in-production
                # Token is valid 24 hours
                token-validity-in-seconds: 86400
                token-validity-in-seconds-for-remember-me: 2592000
    mail: # specific JHipster mail property, for standard properties see MailProperties
        from: gateway@localhost
        base-url: http://127.0.0.1:8080
    metrics: # DropWizard Metrics configuration, used by MetricsConfiguration
        jmx.enabled: true
        graphite: # Use the "graphite" Maven profile to have the Graphite dependencies
            enabled: false
            host: localhost
            port: 2003
            prefix: gateway
        prometheus: # Use the "prometheus" Maven profile to have the Prometheus dependencies
            enabled: false
            endpoint: /prometheusMetrics
        logs: # Reports Dropwizard metrics in the logs
            enabled: false
            report-frequency: 60 # in seconds
    logging:
        logstash: # Forward logs to logstash over a socket, used by LoggingConfiguration
            enabled: false
            host: localhost
            port: 5000
            queue-size: 512
        spectator-metrics: # Reports Spectator Circuit Breaker metrics in the logs
            enabled: false
            # edit spring.metrics.export.delay-millis to set report frequency

# ===================================================================
# Application specific properties
# Add your own application properties here, see the ApplicationProperties class
# to have type-safe configuration, like in the JHipsterProperties above
#
# More documentation is available at:
# http://www.jhipster.tech/common-application-properties/
# ===================================================================

application:
garyrussell commented 6 years ago

You don't show your application properties. Does the subscribableChannel binding have a group? Anonymous consumers read from the end of the topic. Consumers with a group consume from where they left off or from the beginning first time.

pavankjadda commented 6 years ago

My bad. Added application-dev.ymlfile. I do not have any group. Is there a way to specify the offset or group name based my application-dev.yml file.

pavankjadda commented 6 years ago

@garyrussell Based on spring cloud documentation, I updated my application-dev.yml to the following.

Application-dev.yml

# ===================================================================
# Spring Boot configuration for the "dev" profile.
#
# This configuration overrides the application.yml file.
#
# More information on profiles: http://www.jhipster.tech/profiles/
# More information on configuration properties: http://www.jhipster.tech/common-application-properties/
# ===================================================================

# ===================================================================
# Standard Spring Boot properties.
# Full reference is available at:
# http://docs.spring.io/spring-boot/docs/current/reference/html/common-application-properties.html
# ===================================================================

logging:
    level:
        ROOT: DEBUG
        com.gateway: DEBUG
        io.github.jhipster: DEBUG

eureka:
    instance:
        prefer-ip-address: true
    client:
        service-url:
            defaultZone: http://admin:${jhipster.registry.password}@localhost:8761/eureka/

spring:
    profiles:
        active: dev
        include: swagger
    devtools:
        restart:
            enabled: true
        livereload:
            enabled: false # we use gulp + BrowserSync for livereload
    jackson:
        serialization.indent_output: true
    cloud:
        stream:
            kafka:
                binder:
                    brokers: localhost
                    zk-nodes: localhost
                    auto-add-partitions: true
                    auto-create-topics: true
                    min-partition-count: 1
                    offsetUpdateCount: true
            bindings:
               messageChannel:
                  destination: project
                  group: group1
                  producer:
                      partitionCount: 1
                      partitioned: true     
               subscribableChannel:
                  destination: project
                  group: group1
                  consumer:
                      concurrency: 1
                      partitioned: true
                      startOffset: earliest

    datasource:
        type: com.zaxxer.hikari.HikariDataSource
        url: jdbc:mysql://localhost:3306/gateway?useUnicode=true&characterEncoding=utf8&useSSL=false
        username: root
        password:
        hikari:
            data-source-properties:
                cachePrepStmts: true
                prepStmtCacheSize: 250
                prepStmtCacheSqlLimit: 2048
                useServerPrepStmts: true
    jpa:
        database-platform: org.hibernate.dialect.MySQL5InnoDBDialect
        database: MYSQL
        show-sql: true
        properties:
            hibernate.id.new_generator_mappings: true
            hibernate.cache.use_second_level_cache: false
            hibernate.cache.use_query_cache: false
            hibernate.generate_statistics: true
    mail:
        host: localhost
        port: 25
        username:
        password:
    messages:
        cache-seconds: 1
    thymeleaf:
        cache: false
    zipkin: # Use the "zipkin" Maven profile to have the Spring Cloud Zipkin dependencies
        base-url: http://localhost:9411
        enabled: false
        locator:
            discovery:
                enabled: true

liquibase:
    contexts: dev

# ===================================================================
# To enable SSL, generate a certificate using:
# keytool -genkey -alias gateway -storetype PKCS12 -keyalg RSA -keysize 2048 -keystore keystore.p12 -validity 3650
#
# You can also use Let's Encrypt:
# https://maximilian-boehm.com/hp2121/Create-a-Java-Keystore-JKS-from-Let-s-Encrypt-Certificates.htm
#
# Then, modify the server.ssl properties so your "server" configuration looks like:
#
# server:
#    port: 8443
#    ssl:
#        key-store: keystore.p12
#        key-store-password: <your-password>
#        key-store-type: PKCS12
#        key-alias: gateway
# ===================================================================
server:
    port: 8080

# ===================================================================
# JHipster specific properties
#
# Full reference is available at: http://www.jhipster.tech/common-application-properties/
# ===================================================================

jhipster:
    gateway:
        rate-limiting:
            enabled: false
            limit: 100000
            duration-in-seconds: 3600
        authorized-microservices-endpoints: # Access Control Policy, if left empty for a route, all endpoints will be accessible
            app1: /api,/v2/api-docs # recommended dev configuration
    http:
        version: V_1_1 # To use HTTP/2 you will need SSL support (see above the "server.ssl" configuration)
    cache: # Cache configuration
        hazelcast: # Hazelcast distributed cache
            time-to-live-seconds: 3600
            backup-count: 1
            management-center: # Full reference is available at: http://docs.hazelcast.org/docs/management-center/3.9/manual/html/Deploying_and_Starting.html
                enabled: false
                update-interval: 3
                url: http://localhost:8180/mancenter
    # CORS is only enabled by default with the "dev" profile, so BrowserSync can access the API
    cors:
        allowed-origins: "*"
        allowed-methods: "*"
        allowed-headers: "*"
        exposed-headers: "Authorization,Link,X-Total-Count"
        allow-credentials: true
        max-age: 1800
    security:
        authentication:
            jwt:
                secret: my-secret-token-to-change-in-production
                # Token is valid 24 hours
                token-validity-in-seconds: 86400
                token-validity-in-seconds-for-remember-me: 2592000
    mail: # specific JHipster mail property, for standard properties see MailProperties
        from: gateway@localhost
        base-url: http://127.0.0.1:8080
    metrics: # DropWizard Metrics configuration, used by MetricsConfiguration
        jmx.enabled: true
        graphite: # Use the "graphite" Maven profile to have the Graphite dependencies
            enabled: false
            host: localhost
            port: 2003
            prefix: gateway
        prometheus: # Use the "prometheus" Maven profile to have the Prometheus dependencies
            enabled: false
            endpoint: /prometheusMetrics
        logs: # Reports Dropwizard metrics in the logs
            enabled: false
            report-frequency: 60 # in seconds
    logging:
        logstash: # Forward logs to logstash over a socket, used by LoggingConfiguration
            enabled: false
            host: localhost
            port: 5000
            queue-size: 512
        spectator-metrics: # Reports Spectator Circuit Breaker metrics in the logs
            enabled: false
            # edit spring.metrics.export.delay-millis to set report frequency

# ===================================================================
# Application specific properties
# Add your own application properties here, see the ApplicationProperties class
# to have type-safe configuration, like in the JHipsterProperties above
#
# More documentation is available at:
# http://www.jhipster.tech/common-application-properties/
# ===================================================================

application:

Now I get following error

org.springframework.context.ApplicationContextException: Failed to start bean 'inputBindingLifecycle'; nested exception is org.springframework.cloud.stream.binder.BinderException: Exception thrown while starting consumer:

Caused by: java.lang.IllegalArgumentException: A list of partitions must be provided

Error Log:

projects-app_1       | 2018-01-31 17:55:26.849  INFO 7 --- [           main] com.netflix.discovery.DiscoveryClient    : Saw local status change event StatusChangeEvent [timestamp=1517421326849, current=UP, previous=STARTING]
projects-app_1       | 2018-01-31 17:55:26.853  INFO 7 --- [nfoReplicator-0] com.netflix.discovery.DiscoveryClient    : DiscoveryClient_PROJECTS/projects:9fb41d03534ebf1f8a1ea5064e0050a6: registering service...
projects-app_1       | 2018-01-31 17:55:26.933  INFO 7 --- [nfoReplicator-0] com.netflix.discovery.DiscoveryClient    : DiscoveryClient_PROJECTS/projects:9fb41d03534ebf1f8a1ea5064e0050a6 - registration status: 204
zookeeper_1          | 2018-01-31 17:55:26,997 [myid:] - INFO  [NIOServerCxn.Factory:0.0.0.0/0.0.0.0:2181:NIOServerCnxnFactory@197] - Accepted socket connection from /172.18.0.7:54080
zookeeper_1          | 2018-01-31 17:55:26,998 [myid:] - INFO  [NIOServerCxn.Factory:0.0.0.0/0.0.0.0:2181:ZooKeeperServer@868] - Client attempting to establish new session at /172.18.0.7:54080
zookeeper_1          | 2018-01-31 17:55:27,018 [myid:] - INFO  [SyncThread:0:ZooKeeperServer@617] - Established session 0x1614d22063b004a with negotiated timeout 10000 for client /172.18.0.7:54080
projects-app_1       | 2018-01-31 17:55:27.027  WARN 7 --- [nfoReplicator-0] o.a.k.clients.consumer.ConsumerConfig    : The configuration 'value.serializer' was supplied but isn't a known config.
projects-app_1       | 2018-01-31 17:55:27.030  WARN 7 --- [nfoReplicator-0] o.a.k.clients.consumer.ConsumerConfig    : The configuration 'key.serializer' was supplied but isn't a known config.
zookeeper_1          | 2018-01-31 17:55:27,035 [myid:] - INFO  [ProcessThread(sid:0 cport:-1)::PrepRequestProcessor@494] - Processed session termination for sessionid: 0x1614d22063b004a
zookeeper_1          | 2018-01-31 17:55:27,039 [myid:] - INFO  [NIOServerCxn.Factory:0.0.0.0/0.0.0.0:2181:NIOServerCnxn@1007] - Closed socket connection for client /172.18.0.7:54080 which had sessionid 0x1614d22063b004a
zookeeper_1          | 2018-01-31 17:55:27,045 [myid:] - INFO  [NIOServerCxn.Factory:0.0.0.0/0.0.0.0:2181:NIOServerCnxnFactory@197] - Accepted socket connection from /172.18.0.7:54084
zookeeper_1          | 2018-01-31 17:55:27,046 [myid:] - INFO  [NIOServerCxn.Factory:0.0.0.0/0.0.0.0:2181:ZooKeeperServer@868] - Client attempting to establish new session at /172.18.0.7:54084
zookeeper_1          | 2018-01-31 17:55:27,049 [myid:] - INFO  [SyncThread:0:ZooKeeperServer@617] - Established session 0x1614d22063b004b with negotiated timeout 10000 for client /172.18.0.7:54084
projects-app_1       | 2018-01-31 17:55:27.061  WARN 7 --- [           main] o.a.k.clients.consumer.ConsumerConfig    : The configuration 'value.serializer' was supplied but isn't a known config.
projects-app_1       | 2018-01-31 17:55:27.065  WARN 7 --- [           main] o.a.k.clients.consumer.ConsumerConfig    : The configuration 'key.serializer' was supplied but isn't a known config.
projects-app_1       | 2018-01-31 17:55:27.171  WARN 7 --- [           main] o.s.c.s.b.k.p.KafkaTopicProvisioner      : The number of expected partitions was: 1, but 0 has been found instead.There will be 1 idle consumers
projects-app_1       | 2018-01-31 17:55:27.173  WARN 7 --- [           main] ationConfigEmbeddedWebApplicationContext : Exception encountered during context initialization - cancelling refresh attempt: org.springframework.context.ApplicationContextException: Failed to start bean 'inputBindingLifecycle'; nested exception is org.springframework.cloud.stream.binder.BinderException: Exception thrown while starting consumer: 
projects-app_1       | 2018-01-31 17:55:27.182  WARN 7 --- [           main] com.netflix.discovery.DiscoveryClient    : Saw local status change event StatusChangeEvent [timestamp=1517421327182, current=DOWN, previous=UP]
projects-app_1       | 2018-01-31 17:55:27.186  WARN 7 --- [           main] c.n.discovery.InstanceInfoReplicator     : Ignoring onDemand update due to rate limiter
projects-app_1       | 2018-01-31 17:55:27.189  INFO 7 --- [           main] com.netflix.discovery.DiscoveryClient    : Shutting down DiscoveryClient ...
projects-app_1       | 2018-01-31 17:55:27.191  INFO 7 --- [           main] com.netflix.discovery.DiscoveryClient    : Unregistering ...
projects-app_1       | 2018-01-31 17:55:27.197  INFO 7 --- [           main] com.netflix.discovery.DiscoveryClient    : DiscoveryClient_PROJECTS/projects:9fb41d03534ebf1f8a1ea5064e0050a6 - deregister  status: 200
projects-app_1       | 2018-01-31 17:55:27.200  INFO 7 --- [           main] com.netflix.discovery.DiscoveryClient    : Completed shut down of DiscoveryClient
projects-app_1       | 2018-01-31 17:55:27.276 ERROR 7 --- [           main] o.s.boot.SpringApplication               : Application startup failed
projects-app_1       | 
projects-app_1       | org.springframework.context.ApplicationContextException: Failed to start bean 'inputBindingLifecycle'; nested exception is org.springframework.cloud.stream.binder.BinderException: Exception thrown while starting consumer: 
projects-app_1       |  at org.springframework.context.support.DefaultLifecycleProcessor.doStart(DefaultLifecycleProcessor.java:178)
projects-app_1       |  at org.springframework.context.support.DefaultLifecycleProcessor.access$200(DefaultLifecycleProcessor.java:50)
projects-app_1       |  at org.springframework.context.support.DefaultLifecycleProcessor$LifecycleGroup.start(DefaultLifecycleProcessor.java:348)
projects-app_1       |  at org.springframework.context.support.DefaultLifecycleProcessor.startBeans(DefaultLifecycleProcessor.java:151)
projects-app_1       |  at org.springframework.context.support.DefaultLifecycleProcessor.onRefresh(DefaultLifecycleProcessor.java:114)
projects-app_1       |  at org.springframework.context.support.AbstractApplicationContext.finishRefresh(AbstractApplicationContext.java:880)
projects-app_1       |  at org.springframework.boot.context.embedded.EmbeddedWebApplicationContext.finishRefresh(EmbeddedWebApplicationContext.java:144)
projects-app_1       |  at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:546)
projects-app_1       |  at org.springframework.boot.context.embedded.EmbeddedWebApplicationContext.refresh(EmbeddedWebApplicationContext.java:122)
projects-app_1       |  at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:693)
projects-app_1       |  at org.springframework.boot.SpringApplication.refreshContext(SpringApplication.java:360)
projects-app_1       |  at org.springframework.boot.SpringApplication.run(SpringApplication.java:303)
projects-app_1       |  at com.projects.ProjectsApp.main(ProjectsApp.java:68)
projects-app_1       |  at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
projects-app_1       |  at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
projects-app_1       |  at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
projects-app_1       |  at java.lang.reflect.Method.invoke(Method.java:498)
projects-app_1       |  at org.springframework.boot.loader.MainMethodRunner.run(MainMethodRunner.java:48)
projects-app_1       |  at org.springframework.boot.loader.Launcher.launch(Launcher.java:87)
projects-app_1       |  at org.springframework.boot.loader.Launcher.launch(Launcher.java:50)
projects-app_1       |  at org.springframework.boot.loader.WarLauncher.main(WarLauncher.java:59)
projects-app_1       | Caused by: org.springframework.cloud.stream.binder.BinderException: Exception thrown while starting consumer: 
projects-app_1       |  at org.springframework.cloud.stream.binder.AbstractMessageChannelBinder.doBindConsumer(AbstractMessageChannelBinder.java:258)
projects-app_1       |  at org.springframework.cloud.stream.binder.AbstractMessageChannelBinder.doBindConsumer(AbstractMessageChannelBinder.java:57)
projects-app_1       |  at org.springframework.cloud.stream.binder.AbstractBinder.bindConsumer(AbstractBinder.java:145)
projects-app_1       |  at org.springframework.cloud.stream.binding.BindingService.bindConsumer(BindingService.java:97)
projects-app_1       |  at org.springframework.cloud.stream.binding.BindableProxyFactory.bindInputs(BindableProxyFactory.java:221)
projects-app_1       |  at org.springframework.cloud.stream.binding.InputBindingLifecycle.start(InputBindingLifecycle.java:55)
projects-app_1       |  at org.springframework.context.support.DefaultLifecycleProcessor.doStart(DefaultLifecycleProcessor.java:175)
projects-app_1       |  ... 20 common frames omitted
projects-app_1       | Caused by: java.lang.IllegalArgumentException: A list of partitions must be provided
projects-app_1       |  at org.springframework.util.Assert.isTrue(Assert.java:92)
projects-app_1       |  at org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder.createConsumerEndpoint(KafkaMessageChannelBinder.java:241)
projects-app_1       |  at org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder.createConsumerEndpoint(KafkaMessageChannelBinder.java:88)
projects-app_1       |  at org.springframework.cloud.stream.binder.AbstractMessageChannelBinder.doBindConsumer(AbstractMessageChannelBinder.java:217)
projects-app_1       |  ... 26 common frames omitted
projects-app_1       | 
zookeeper_1          | 2018-01-31 17:55:27,666 [myid:] - WARN  [NIOServerCxn.Factory:0.0.0.0/0.0.0.0:2181:NIOServerCnxn@357] - caught end of stream exception
zookeeper_1          | EndOfStreamException: Unable to read additional data from client sessionid 0x1614d22063b004b, likely client has closed socket
zookeeper_1          |  at org.apache.zookeeper.server.NIOServerCnxn.doIO(NIOServerCnxn.java:228)
zookeeper_1          |  at org.apache.zookeeper.server.NIOServerCnxnFactory.run(NIOServerCnxnFactory.java:208)
zookeeper_1          |  at java.lang.Thread.run(Thread.java:745)
zookeeper_1          | 2018-01-31 17:55:27,667 [myid:] - INFO  [NIOServerCxn.Factory:0.0.0.0/0.0.0.0:2181:NIOServerCnxn@1007] - Closed socket connection for client /172.18.0.7:54084 which had sessionid 0x1614d22063b004b
zookeeper_1          | 2018-01-31 17:55:27,667 [myid:] - WARN  [NIOServerCxn.Factory:0.0.0.0/0.0.0.0:2181:NIOServerCnxn@357] - caught end of stream exception
zookeeper_1          | EndOfStreamException: Unable to read additional data from client sessionid 0x1614d22063b0047, likely client has closed socket
zookeeper_1          |  at org.apache.zookeeper.server.NIOServerCnxn.doIO(NIOServerCnxn.java:228)
zookeeper_1          |  at org.apache.zookeeper.server.NIOServerCnxnFactory.run(NIOServerCnxnFactory.java:208)
zookeeper_1          |  at java.lang.Thread.run(Thread.java:745)
zookeeper_1          | 2018-01-31 17:55:27,668 [myid:] - INFO  [NIOServerCxn.Factory:0.0.0.0/0.0.0.0:2181:NIOServerCnxn@1007] - Closed socket connection for client /172.18.0.7:54068 which had sessionid 0x1614d22063b0047
zookeeper_1          | 2018-01-31 17:55:27,668 [myid:] - WARN  [NIOServerCxn.Factory:0.0.0.0/0.0.0.0:2181:NIOServerCnxn@357] - caught end of stream exception
zookeeper_1          | EndOfStreamException: Unable to read additional data from client sessionid 0x1614d22063b0049, likely client has closed socket
zookeeper_1          |  at org.apache.zookeeper.server.NIOServerCnxn.doIO(NIOServerCnxn.java:228)
zookeeper_1          |  at org.apache.zookeeper.server.NIOServerCnxnFactory.run(NIOServerCnxnFactory.java:208)
zookeeper_1          |  at java.lang.Thread.run(Thread.java:745)
zookeeper_1          | 2018-01-31 17:55:27,668 [myid:] - INFO  [NIOServerCxn.Factory:0.0.0.0/0.0.0.0:2181:NIOServerCnxn@1007] - Closed socket connection for client /172.18.0.7:54074 which had sessionid 0x1614d22063b0049
dockercompose_projects-app_1 exited with code 1
gateway-app_1        | 2018-01-31 17:55:30.850  WARN 8 --- [nfoReplicator-0] o.a.k.clients.consumer.ConsumerConfig    : The configuration 'value.serializer' was supplied but isn't a known config.
gateway-app_1        | 2018-01-31 17:55:30.850  WARN 8 --- [nfoReplicator-0] o.a.k.clients.consumer.ConsumerConfig    : The configuration 'key.serializer' was supplied but isn't a known config.
zookeeper_1          | 2018-01-31 17:55:36,000 [myid:] - INFO  [SessionTracker:ZooKeeperServer@347] - Expiring session 0x1614d22063b0049, timeout of 10000ms exceeded
zookeeper_1          | 2018-01-31 17:55:36,001 [myid:] - INFO  [SessionTracker:ZooKeeperServer@347] - Expiring session 0x1614d22063b0047, timeout of 10000ms exceeded
zookeeper_1          | 2018-01-31 17:55:36,001 [myid:] - INFO  [ProcessThread(sid:0 cport:-1)::PrepRequestProcessor@494] - Processed session termination for sessionid: 0x1614d22063b0049
zookeeper_1          | 2018-01-31 17:55:36,002 [myid:] - INFO  [ProcessThread(sid:0 cport:-1)::PrepRequestProcessor@494] - Processed session termination for sessionid: 0x1614d22063b0047
zookeeper_1          | 2018-01-31 17:55:38,000 [myid:] - INFO  [SessionTracker:ZooKeeperServer@347] - Expiring session 0x1614d22063b004b, timeout of 10000ms exceeded
zookeeper_1          | 2018-01-31 17:55:38,001 [myid:] - INFO  [ProcessThread(sid:0 cport:-1)::PrepRequestProcessor@494] - Processed session termination for sessionid: 0x1614d22063b004b
gateway-app_1        | 2018-01-31 17:55:40.956  WARN 8 --- [nfoReplicator-0] o.a.k.clients.consumer.ConsumerConfig    : The configuration 'value.serializer' was supplied but isn't a known config.
gateway-app_1        | 2018-01-31 17:55:40.956  WARN 8 --- [nfoReplicator-0] o.a.k.clients.consumer.ConsumerConfig    : The configuration 'key.serializer' was supplied but isn't a known config.
gateway-app_1        | 2018-01-31 17:55:51.065  WARN 8 --- [nfoReplicator-0] o.a.k.clients.consumer.ConsumerConfig    : The configuration 'value.serializer' was supplied but isn't a known config.
gateway-app_1        | 2018-01-31 17:55:51.065  WARN 8 --- [nfoReplicator-0] o.a.k.clients.consumer.ConsumerConfig    : The configuration 'key.serializer' was supplied but isn't a known config.
garyrussell commented 6 years ago

Nothing stands out except you should probably use difference groups for each consumer; also it makes no sense to have partitioned: true with only one partition.

If neither of those help; next step would be a DEBUG log covering the app initialization.

pavankjadda commented 6 years ago

I made suggested changes (removed unused properties) and clean installed the project. I do not get any errors now. But the consumer is not consuming. @garyrussell any ideas?

application-dev.yml

# ===================================================================
# Spring Boot configuration for the "dev" profile.
#
# This configuration overrides the application.yml file.
#
# More information on profiles: http://www.jhipster.tech/profiles/
# More information on configuration properties: http://www.jhipster.tech/common-application-properties/
# ===================================================================

# ===================================================================
# Standard Spring Boot properties.
# Full reference is available at:
# http://docs.spring.io/spring-boot/docs/current/reference/html/common-application-properties.html
# ===================================================================

logging:
    level:
        ROOT: DEBUG
        com.gateway: DEBUG
        io.github.jhipster: DEBUG

eureka:
    instance:
        prefer-ip-address: true
    client:
        service-url:
            defaultZone: http://admin:${jhipster.registry.password}@localhost:8761/eureka/

spring:
    profiles:
        active: dev
        include: swagger
    devtools:
        restart:
            enabled: true
        livereload:
            enabled: false # we use gulp + BrowserSync for livereload
    jackson:
        serialization.indent_output: true
    cloud:
        stream:
            kafka:
                binder:
                    brokers: localhost
                    zk-nodes: localhost
            bindings:
               messageChannel:
                  destination: project
                  group: group1
                  producer:
                      sync: false 
               subscribableChannel:
                  destination: project
                  group: group2
                  consumer:
                      startOffset: earliest

    datasource:
        type: com.zaxxer.hikari.HikariDataSource
        url: jdbc:mysql://localhost:3306/gateway?useUnicode=true&characterEncoding=utf8&useSSL=false
        username: root
        password:
        hikari:
            data-source-properties:
                cachePrepStmts: true
                prepStmtCacheSize: 250
                prepStmtCacheSqlLimit: 2048
                useServerPrepStmts: true
    jpa:
        database-platform: org.hibernate.dialect.MySQL5InnoDBDialect
        database: MYSQL
        show-sql: true
        properties:
            hibernate.id.new_generator_mappings: true
            hibernate.cache.use_second_level_cache: false
            hibernate.cache.use_query_cache: false
            hibernate.generate_statistics: true
    mail:
        host: localhost
        port: 25
        username:
        password:
    messages:
        cache-seconds: 1
    thymeleaf:
        cache: false
    zipkin: # Use the "zipkin" Maven profile to have the Spring Cloud Zipkin dependencies
        base-url: http://localhost:9411
        enabled: false
        locator:
            discovery:
                enabled: true

liquibase:
    contexts: dev

# ===================================================================
# To enable SSL, generate a certificate using:
# keytool -genkey -alias gateway -storetype PKCS12 -keyalg RSA -keysize 2048 -keystore keystore.p12 -validity 3650
#
# You can also use Let's Encrypt:
# https://maximilian-boehm.com/hp2121/Create-a-Java-Keystore-JKS-from-Let-s-Encrypt-Certificates.htm
#
# Then, modify the server.ssl properties so your "server" configuration looks like:
#
# server:
#    port: 8443
#    ssl:
#        key-store: keystore.p12
#        key-store-password: <your-password>
#        key-store-type: PKCS12
#        key-alias: gateway
# ===================================================================
server:
    port: 8080

# ===================================================================
# JHipster specific properties
#
# Full reference is available at: http://www.jhipster.tech/common-application-properties/
# ===================================================================

jhipster:
    gateway:
        rate-limiting:
            enabled: false
            limit: 100000
            duration-in-seconds: 3600
        authorized-microservices-endpoints: # Access Control Policy, if left empty for a route, all endpoints will be accessible
            app1: /api,/v2/api-docs # recommended dev configuration
    http:
        version: V_1_1 # To use HTTP/2 you will need SSL support (see above the "server.ssl" configuration)
    cache: # Cache configuration
        hazelcast: # Hazelcast distributed cache
            time-to-live-seconds: 3600
            backup-count: 1
            management-center: # Full reference is available at: http://docs.hazelcast.org/docs/management-center/3.9/manual/html/Deploying_and_Starting.html
                enabled: false
                update-interval: 3
                url: http://localhost:8180/mancenter
    # CORS is only enabled by default with the "dev" profile, so BrowserSync can access the API
    cors:
        allowed-origins: "*"
        allowed-methods: "*"
        allowed-headers: "*"
        exposed-headers: "Authorization,Link,X-Total-Count"
        allow-credentials: true
        max-age: 1800
    security:
        authentication:
            jwt:
                secret: my-secret-token-to-change-in-production
                # Token is valid 24 hours
                token-validity-in-seconds: 86400
                token-validity-in-seconds-for-remember-me: 2592000
    mail: # specific JHipster mail property, for standard properties see MailProperties
        from: gateway@localhost
        base-url: http://127.0.0.1:8080
    metrics: # DropWizard Metrics configuration, used by MetricsConfiguration
        jmx.enabled: true
        graphite: # Use the "graphite" Maven profile to have the Graphite dependencies
            enabled: false
            host: localhost
            port: 2003
            prefix: gateway
        prometheus: # Use the "prometheus" Maven profile to have the Prometheus dependencies
            enabled: false
            endpoint: /prometheusMetrics
        logs: # Reports Dropwizard metrics in the logs
            enabled: false
            report-frequency: 60 # in seconds
    logging:
        logstash: # Forward logs to logstash over a socket, used by LoggingConfiguration
            enabled: false
            host: localhost
            port: 5000
            queue-size: 512
        spectator-metrics: # Reports Spectator Circuit Breaker metrics in the logs
            enabled: false
            # edit spring.metrics.export.delay-millis to set report frequency

# ===================================================================
# Application specific properties
# Add your own application properties here, see the ApplicationProperties class
# to have type-safe configuration, like in the JHipsterProperties above
#
# More documentation is available at:
# http://www.jhipster.tech/common-application-properties/
# ===================================================================

application:

kafka.yml

version: '2'
services:
    zookeeper:
        image: wurstmeister/zookeeper:3.4.6
    kafka:
        image: wurstmeister/kafka:1.0.0
        environment:
            KAFKA_ADVERTISED_HOST_NAME: 192.168.1.64
            KAFKA_ADVERTISED_PORT: 9092
            KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
            KAFKA_CREATE_TOPICS: "project:1:1"
        ports:
            - 9092:9092  # Uncomment to make Kafka available externally

docker-compose.yml

version: '2'
services:
    gateway-app:
        image: gateway
        environment:
            - SPRING_PROFILES_ACTIVE=prod,swagger
            - EUREKA_CLIENT_SERVICE_URL_DEFAULTZONE=http://admin:$${jhipster.registry.password}@jhipster-registry:8761/eureka
            - SPRING_CLOUD_CONFIG_URI=http://admin:$${jhipster.registry.password}@jhipster-registry:8761/config
            - SPRING_DATASOURCE_URL=jdbc:mysql://gateway-mysql:3306/gateway?useUnicode=true&characterEncoding=utf8&useSSL=false
            - JHIPSTER_SLEEP=30
            - SPRING_CLOUD_STREAM_KAFKA_BINDER_BROKERS=kafka
            - SPRING_CLOUD_STREAM_KAFKA_BINDER_ZK_NODES=zookeeper
            - JHIPSTER_REGISTRY_PASSWORD=admin
        ports:
            - 8080:8080
    gateway-mysql:
        image: mysql:5.7.20
        environment:
            - MYSQL_USER=root
            - MYSQL_ALLOW_EMPTY_PASSWORD=yes
            - MYSQL_DATABASE=gateway
        command:             mysqld --lower_case_table_names=1 --skip-ssl --character_set_server=utf8
            --explicit_defaults_for_timestamp

    projects-app:
        image: projects
        environment:
            - SPRING_PROFILES_ACTIVE=prod,swagger
            - EUREKA_CLIENT_SERVICE_URL_DEFAULTZONE=http://admin:$${jhipster.registry.password}@jhipster-registry:8761/eureka
            - SPRING_CLOUD_CONFIG_URI=http://admin:$${jhipster.registry.password}@jhipster-registry:8761/config
            - SPRING_DATASOURCE_URL=jdbc:mysql://projects-mysql:3306/projects?useUnicode=true&characterEncoding=utf8&useSSL=false
            - JHIPSTER_SLEEP=30
            - SPRING_CLOUD_STREAM_KAFKA_BINDER_BROKERS=kafka
            - SPRING_CLOUD_STREAM_KAFKA_BINDER_ZK_NODES=zookeeper
            - JHIPSTER_REGISTRY_PASSWORD=admin
    projects-mysql:
        image: mysql:5.7.20
        environment:
            - MYSQL_USER=root
            - MYSQL_ALLOW_EMPTY_PASSWORD=yes
            - MYSQL_DATABASE=projects
        command:             mysqld --lower_case_table_names=1 --skip-ssl --character_set_server=utf8
            --explicit_defaults_for_timestamp

    zookeeper:
        extends:
            file: kafka.yml
            service: zookeeper
    kafka:
        extends:
            file: kafka.yml
            service: kafka
        environment:
            KAFKA_ADVERTISED_HOST_NAME: kafka
    jhipster-registry:
        extends:
            file: jhipster-registry.yml
            service: jhipster-registry
garyrussell commented 6 years ago

Turn on DEBUG logging for org.springframework.kafka.

pavankjadda commented 6 years ago

I did. When I manually login to Kafka container and sent a message with the command bin/kafka-console-producer.sh --broker-list 172.19.0.5:9092 --topic greetings, I got an error message WARN Error while fetching metadata with correlation id 39 : {greetings=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient). I looked up online and opened server.properties file. It does not haveadvertised.host.name set and it makes sense as it is deprecated in new versions. When I set the value manually to localhost, it worked. @garyrussell is this a bug in version?

bash-4.3# bin/kafka-topics.sh --list --zookeeper 172.19.0.6:2181
__consumer_offsets
greetings
messageChannel
subscribableChannel
topic-jhipster
bash-4.3# bin/kafka-topics.sh --describe --zookeeper 172.19.0.6:2181 --topic greetings
Topic:greetings PartitionCount:1    ReplicationFactor:1 Configs:
    Topic: greetings    Partition: 0    Leader: 999 Replicas: 999   Isr: 999

bash-4.3# bin/kafka-console-producer.sh --broker-list 172.19.0.5:9092 --topic greetings
>Hello
WARN Error while fetching metadata with correlation id 39 : 
     {greetings=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
garyrussell commented 6 years ago

is this a bug in version?

That's a question for the Kafka folks, not us; sorry.

Closing this; please reopen if you think there is a problem with our code.

pavankjadda commented 6 years ago

@garyrussell I was able to fix the issue with advertised.host.name entry. Now I am able to send and receive messages from producer to consumer through following commands.

$bin/kafka-console-producer.sh --broker-list 172.18.0.7:9092 --topic greetings

$bin/kafka-console-consumer.sh --bootstrap-server 172.18.0.7:9092 --topic greetings --from-beginning

Now the real Spring Cloud Stream problem came back. When I send a message from the producer, I am able to see the send confirmation on the console, but the consumer is not receiving it. I think Spring Cloud Stream failed to send this to Kafka. Please advice.

garyrussell commented 6 years ago

I think Spring Cloud Stream failed to send this to Kafka. Please advice.

That makes no sense if you see the message on the console.

One more time - you need to enable DEBUG logging to determine what's going on; I would suggest DEBUG for all of org.springframework.

pavankjadda commented 6 years ago

Setting autoCreateTopics: false in application-dev.yml file solved my issue. It took a while to figure the issue.