Consumer concurrency not working (all consumed message by same thread)

[lubos-bistak]

Hello,

i built an application that writes and reads Java objects (serialized/deserialized using Avro) into a Kafka topic having 3 partitions by using Spring Cloud Stream. I want to read in parallel from this topic, but it seems not be working as always same thread handles the Java objects from this topic sequential.

Code to read from the topic looks like this:

@StreamListener(InputSink.INPUT_TOPIC1)
    public void readFromTopic1(SomeObject obj) {
        log.debug("[] Read from Kafka: {} ", Thread.currentThread().getName(),obj.toString());
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

In logs i always see the same thread name "-L-1":
-L-1 Read from Kafka: ...

Application start-up logs:

[           -C-1] o.s.c.s.b.k.KafkaMessageChannelBinder$3  : partitions assigned:[topic1-2]
[           -C-1] o.s.c.s.b.k.KafkaMessageChannelBinder$3  : partitions assigned:[topic1-0]
[           -C-1] o.s.c.s.b.k.KafkaMessageChannelBinder$3  : partitions assigned:[topic1-1]

Configuration in application.yml:

spring:
  cloud:
    stream:
      default:
        contentType: application/*+avro
      bindings:
        outputTopic1
          destination: topic1
        inputTopic1:
          destination: topic1
          group: group-inputTopic1
          consumer:
            concurrency: 3
            partitioned: true
      schema:
        avro:
          dynamic-schema-generation-enabled: true

Spring dependencies:

• spring-cloud-stream-reactive:1.3.0.RELEASE
• spring-could-stream-schema:1.3.0.RELEASE
• spring-cloud-starter-stream-kafka:1.3.0.RELEASE
• spring-cloud-stream:1.3.0.RELEASE
• spring-boot-starter-web:1.5.10.RELEASE

Thanks a lot for any advice !!!

[garyrussell]

This is just a thread naming problem; 1.3.x inherits a very old version of spring-kafka from boot. The messages are actually processed on different threads...

@SpringBootApplication
@EnableBinding(MyChannels.class)
public class Kbgh356Application {

	private final Logger logger = LoggerFactory.getLogger(Kbgh356Application.class);

	public static void main(String[] args) {
		SpringApplication.run(Kbgh356Application.class, args);
	}

	@Bean
	public ApplicationRunner runner(MessageChannel output) {
		return args -> {
			Thread.sleep(10_000);
			output.send(new GenericMessage<>("foo", Collections.singletonMap(KafkaHeaders.PARTITION_ID, 0)));
			output.send(new GenericMessage<>("bar", Collections.singletonMap(KafkaHeaders.PARTITION_ID, 1)));
			output.send(new GenericMessage<>("baz", Collections.singletonMap(KafkaHeaders.PARTITION_ID, 2)));
		};
	}

	@StreamListener("input")
	public void listen(String in, @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition) {
		logger.info(in + " from partition " + partition + " thread: " + Thread.currentThread().getId());
	}

}

interface MyChannels {

	@Input
	MessageChannel input();

	@Output
	MessageChannel output();

}

2018-04-16 11:15:55.299  INFO 74942 --- [           -C-1] o.s.c.s.b.k.KafkaMessageChannelBinder$3  : partitions assigned:[topic1-0]
2018-04-16 11:15:55.299  INFO 74942 --- [           -C-1] o.s.c.s.b.k.KafkaMessageChannelBinder$3  : partitions assigned:[topic1-2]
2018-04-16 11:15:55.299  INFO 74942 --- [           -C-1] o.s.c.s.b.k.KafkaMessageChannelBinder$3  : partitions assigned:[topic1-1]
...
2018-04-16 11:16:02.451  INFO 74942 --- [           -L-1] com.example.Kbgh356Application           : bar from partition 1 thread: 61
2018-04-16 11:16:02.451  INFO 74942 --- [           -L-1] com.example.Kbgh356Application           : baz from partition 2 thread: 60
2018-04-16 11:16:02.451  INFO 74942 --- [           -L-1] com.example.Kbgh356Application           : foo from partition 0 thread: 59

Notice that although the threads have the same name, they have different IDs.

In any case, I would recommend upgrading to a newer release of the kafka libraries:

		<dependency>
			<groupId>org.springframework.kafka</groupId>
			<artifactId>spring-kafka</artifactId>
			<version>1.3.4.RELEASE</version>
		</dependency>
		<dependency>
			<groupId>org.springframework.integration</groupId>
			<artifactId>spring-integration-kafka</artifactId>
			<version>2.3.0.RELEASE</version>
		</dependency>
		<dependency>
			<groupId>org.apache.kafka</groupId>
			<artifactId>kafka-clients</artifactId>
			<version>0.11.0.1</version>
		</dependency>

This has a better threading model (thanks to KIP-62) - there are no longer any -L-x threads, just -C-x.

(Also 1.3.2 of the spring-cloud-stream jars from Edgware.SR3).

However, it still suffers from the thread naming problem.

I just made a change to spring-kafka to improve things...

2018-04-16 11:30:27.349  INFO 77754 --- [ consumer-0-C-1] com.example.Kbgh356Application           : foo from partition 0 thread: 26
2018-04-16 11:30:27.349  INFO 77754 --- [ consumer-1-C-1] com.example.Kbgh356Application           : bar from partition 1 thread: 28
2018-04-16 11:30:27.349  INFO 77754 --- [ consumer-2-C-1] com.example.Kbgh356Application           : baz from partition 2 thread: 30

[lubos-bistak]

Thank you for your replay, i updated dependencies as you suggested, but now i get following WARNing and programs stops:

o.s.c.s.b.k.p.KafkaTopicProvisioner      : Auto creation of topics is enabled, but Kafka AdminUtils class is not present on the classpath. No topic will be created by the binder
2018-04-19 14:30:04.838  WARN [....,,,] 42482 --- [ad | producer-1] org.apache.kafka.clients.NetworkClient   : Error while fetching metadata with correlation id 1 : {businessPartner=LEADER_NOT_AVAILABLE}

[garyrussell]

Please provide your complete pom.