multiple kafka consumers with same group id

Viewed 47

i'm new to kafka. I have created a kafka consumer with spring boot (spring-kafka dependency). In my app i have used consumerFactory and producerfactory beans for config. So in my application i have created the kafka consumer like below.

@RetryableTopic(
        attempts = "3",
        backoff = @Backoff(delay = 1000, multiplier = 2.0),
        autoCreateTopics = "false")
  @KafkaListener(topics = "myTopic", groupId = "myGroupId")
  public void consume(@Payload(required = false) String message) {

       processMessage(message);
  
}

My configs are like below

@Bean
  public ConsumerFactory<String, Object> consumerFactory() {
    Map<String, Object> config = new HashMap<>();
    config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, env.getProperty("kafka.consumer.bootstrap.servers"));
    config.put(ConsumerConfig.GROUP_ID_CONFIG, env.getProperty("kafka.consumer.group"));
    config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

    return new DefaultKafkaConsumerFactory<>(config);
  }

  @Bean
  public ConcurrentKafkaListenerContainerFactory<String, Object> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, Object> factory =
            new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    factory.getContainerProperties().setCommitLogLevel(LogIfLevelEnabled.Level.DEBUG);
    factory.getContainerProperties().setMissingTopicsFatal(false);
    return factory;
  }

  @Bean
  public ProducerFactory<String, String> producerFactory() {
    Map<String, Object> config = new HashMap<>();
    config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, env.getProperty("kafka.consumer.bootstrap.servers"));
    config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    return new DefaultKafkaProducerFactory<>(config);
  }

  @Bean
  public KafkaTemplate<String, String> kafkaTemplate() {
    return new KafkaTemplate<>(producerFactory());
  }

So i want to consume parallelly since i may get more messages. What i found about consuming parallelly topics is that i need to create multiple partitions for a topic and i need to create a consumer for each partition. Let´s say i have 10 partitions for my topic, then i can have 10 consumers in the same consumer group reading one partition each. I understand this behavior. But my concern is how can i create several consumers in my application.

Do i have to write multiple kafka consumer using @KafkaListener with the same functionality ? In that case do i have to write below method X amount of times if i need X amount of same consumers.

@RetryableTopic(
            attempts = "3",
            backoff = @Backoff(delay = 1000, multiplier = 2.0),
            autoCreateTopics = "false")
      @KafkaListener(topics = "myTopic", groupId = "myGroupId")
      public void consume(@Payload(required = false) String message) {
    
           processMessage(message);
      
    }

What are the options or configs that i need to achieve parallel consuming with multiple consumers ?

Thank you in advance.

1 Answers

The @KafkaListener has this option:

/**
 * Override the container factory's {@code concurrency} setting for this listener. May
 * be a property placeholder or SpEL expression that evaluates to a {@link Number}, in
 * which case {@link Number#intValue()} is used to obtain the value.
 * <p>SpEL {@code #{...}} and property place holders {@code ${...}} are supported.
 * @return the concurrency.
 * @since 2.2
 */
String concurrency() default "";

See more in docs: https://docs.spring.io/spring-kafka/reference/html/#kafka-listener-annotation

Related