I've a simple spring bean/component with one method and one int property (initialized to 0) as shown below and this method (Kafka consumer) is being called multiple times in a short span of time.
@Component
public class MyKafkaConsumer{
public int messageCount = 0;
@KafkaListener
public void consumeMessage(MyPayload payload){
messageCount++;
System.out.println("messageCount - "+ messageCount);
}
}
Now when I see the output of this method, it's coming as below and I'm surprised to see this output.
messageCount - 1
messageCount - 2
messageCount - 3
messageCount - 4
messageCount - 1
messageCount - 2
...
Here is my configuration
@Configuration
@ComponentScan(basePackages = {
"mypackages"})
@EnableAutoConfiguration(exclude = EmbeddedMongoAutoConfiguration.class)
@EnableDiscoveryClient
public class MyApplication {
public static void main(String[] args) {
SpringApplication.run(MyApplication.class, args);
System.out.println("MyApplication is ready........");
}
}
This behavior is not consistent but intermittent. Can someone help me understand why is it behaving like this? Am i missing something basic here?
I tried with synchronized block as shown below
@Component
public class MyKafkaConsumer{
public int messageCount = 0;
@KafkaListener
public void consumeMessage(MyPayload payload){
System.out.println("Thread - "+Thread.currentThread().getName()+" count "+messageCount);
synchronized (this){
messageCount++;
}
}
}
and here is the output and we can clearly see there is only one thread operating there and the count is expected to be 7 for but it's not
Thread - org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1 count 0
Thread - org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1 count 1
Thread - org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1 count 2
Thread - org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1 count 5
Thread - org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1 count 3
Thread - org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1 count 4
Thread - org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1 count 5