spring cloud stream - consumer group bound

Question

My consumer is bound to anonymous consumer group instead of consumer group I specified.

spring:
  cloud:
    stream:
      kafka:
        binder:
          brokers: localhost
          defaultBrokerPort: 9092
          zkNodes: localhost
          defaultZkPort: 2181
        bindings:
          inEvent:
            group: eventin
            destination: event
          outEvent:
            group: eventout
            destination: processevent

My Spring boot applicaition

@SpringBootApplication
@EnableBinding(EventStream.class)
public class ConsumerApplication {
    public static void main(String[] args) {
        SpringApplication.run(ConsumerApplication.class, args);
    }
    @StreamListener(value = "inEvent")
    public void getEvent(Event event){
        System.out.println(event.name);
    }
}

My input output channel interface

public interface EventStream {
    @Input("inEvent")
    SubscribableChannel inEvent();
    @Output("outEvent")
    MessageChannel outEvent();
}

my console log--

: Started ConsumerApplication in 3.233 seconds (JVM running for 4.004) : [Consumer clientId=consumer-3, groupId=anonymous.0d0c87d6-ef39-4bfe-b475-4491c40caf6d] Discovered group coordinator singh:9092 (id: 2147483647 rack: null) : [Consumer clientId=consumer-3, groupId=anonymous.0d0c87d6-ef39-4bfe-b475-4491c40caf6d] Revoking previously assigned partitions [] : partitions revoked: [] : [Consumer clientId=consumer-3, groupId=anonymous.0d0c87d6-ef39-4bfe-b475-4491c40caf6d] (Re-)joining group : [Consumer clientId=consumer-3, groupId=anonymous.0d0c87d6-ef39-4bfe-b475-4491c40caf6d] Successfully joined group with generation 1 : [Consumer clientId=consumer-3, groupId=anonymous.0d0c87d6-ef39-4bfe-b475-4491c40caf6d] Setting newly assigned partitions [inEvent-0] : [Consumer clientId=consumer-3, groupId=anonymous.0d0c87d6-ef39-4bfe-b475-4491c40caf6d] Resetting offset for partition inEvent-0 to offset 2. : partitions assigned: [inEvent-0]

Artem Bilan Artem Bilan · Accepted Answer · 2019-02-25T13:55:28

The group property must not be in the kafka tree. It has to be like this:

My consumer is bound to anonymous consumer group instead of consumer group I specified.

spring:
  cloud:
    stream:
       bindings:
          inEvent:
            group: eventin
            destination: event

See more info in the Docs: http://cloud.spring.io/spring-cloud-static/spring-cloud-stream/2.1.1.RELEASE/single/spring-cloud-stream.html#consumer-groups

The group is a common property, so it is the same independently of the binder implementation. The kafka is for Apache Kafka specific properties, exposed on its binder implementation level.

spring cloud stream - consumer group bound

1 Answers