How to pause/resume consumers in spring-kafka when containers are defined manually and listener annotations are not used

user_4206960 0
Viewed 1282

I'm creating a generic Kafka module that handles all the Kafka related configuration code and just needs to be added as a dependency in the parent application. The application then just needs to extend a class and override its method and write only necessary business logic. For having custom error handler and recovery, I hadn't used any annotations.

public abstract class AbstractKafkaConsumerImpl<K, V> {

  @Value("${kafka.fixed-back-off.interval}")
  private long fixedBackOffInterval;

  @Value("${kafka.fixed-back-off.max-attempts}")
  private long fixedBackOffMaxAttempts;

  @Autowired
  private KafkaConsumerConfiguration defaultConsumerPropsConfiguration;

  @Autowired
  private KafkaProducerConfiguration kafkaProducerConfiguration;


  private KafkaTemplate<K, V> dlqSender;

  @PostConstruct
  public void init() {


    DefaultKafkaProducerFactory<K, V> producerFactory =
        new DefaultKafkaProducerFactory<>(kafkaProducerConfiguration.getProducer());
    dlqSender = new KafkaTemplate<>(producerFactory);
    getLogger().info("DLQ sender initialised for topic: {}", getTopicName());


  }

  @EventListener(ApplicationReadyEvent.class)
  public void consume() {
    final String consumerTopicName = getTopicName();

    Map<String, Object> consumerProps =
        new HashMap<>(defaultConsumerPropsConfiguration.getConsumer());
    configureConsumerProperties(consumerProps);

    DefaultKafkaConsumerFactory<K, V> kafkaConsumerFactory =
        new DefaultKafkaConsumerFactory<>(consumerProps);


    SeekToCurrentErrorHandler seekToCurrentErrorHandler;

    DeadLetterPublishingRecoverer recoverer =
        new DeadLetterPublishingRecoverer(dlqSender, (r, e) -> {
          getLogger().error("Retries exhausted for consumer topic: {}. Sending to DLQ", r.topic());
          return new TopicPartition(consumerTopicName + "_DLQ", r.partition());
        });
    seekToCurrentErrorHandler = new SeekToCurrentErrorHandler(recoverer,
        new FixedBackOff(fixedBackOffInterval, fixedBackOffMaxAttempts));

    seekToCurrentErrorHandler.addNotRetryableException(NonRetryableException.class);
    ContainerProperties containerProperties = new ContainerProperties(consumerTopicName);


    AcknowledgingMessageListener<K, V> messageListener = new AcknowledgingMessageListener<K, V>() {
      @Override
      public void onMessage(ConsumerRecord<K, V> data, Acknowledgment acknowledgment) {
        processMessage(data);
        acknowledge(data, acknowledgment);
      }
    };

    containerProperties.setMessageListener(messageListener);
    containerProperties.setAckMode(AckMode.MANUAL_IMMEDIATE);
    seekToCurrentErrorHandler.setCommitRecovered(true);

    KafkaMessageListenerContainer<K, V> container =
        new KafkaMessageListenerContainer<>(kafkaConsumerFactory, containerProperties);
    container.setErrorHandler(seekToCurrentErrorHandler);
    container.start();
  }

  private void acknowledge(ConsumerRecord<K, V> receivedRecord, Acknowledgment acknowledgment) {
    acknowledgment.acknowledge();
    getLogger().debug("Topic: {}. Offset acknowledged: {}.", receivedRecord.topic(),
        receivedRecord.offset());
  }

  protected void configureConsumerProperties(Map<String, Object> consumerProps) {
    consumerProps.put(JsonDeserializer.KEY_DEFAULT_TYPE, getKeyClass());
    consumerProps.put(JsonDeserializer.VALUE_DEFAULT_TYPE, getValueClass());
  }


  protected abstract void processMessage(ConsumerRecord<K, V> receivedRecord);

  protected abstract Logger getLogger();

  protected abstract String getTopicName();

  protected abstract Class<K> getKeyClass();

  protected abstract Class<V> getValueClass();

}

Since each consumer extending the class will have different Key and Value Type, I had to create different consumer factory for each.

Now there's a requirement to add the functionality of pausing/resuming any consumer(s) dynamically.

I came across KafkaListenerEndpointRegistry, but it registers only Kafka Listener annotated methods.

Is there a way I can register the manually created containers to the registry and use its pause/resume calls?

Also, on a side note, does the above practice of creating factories for each consumer fine? Keeping in mind that the code should be generic and compatible with all Key and Value types.

spring-kafka kafka-consumer-api apache-kafka spring-boot java
1 Answers

You should not be creating the factory and containers like that; they must be a bean to work properly, unless you take over the responsibility of Springs container initialization work (injection of event publisher, etc, etc).

If you must create them manually, you should register them as beans using GenericApplicationContext.registerBean() then Spring will initialize them.

The framework really wasn't designed to be used outside of Spring; perhaps ask another question with the problem you are trying to solve, and someone can provide a better solution than the one you have come up with.

container.start();

You need to keep a reference to the container to pause and resume it.

If you register it as a bean, you can get it by name from the Application context.

user_12407630
Related