CODEWITHSUNDEEP
apache-kafkakafka-consumer-apispring-kafka
Althaf Khan
Althaf Khan

Reputation: 35

Prioritizing Kafka topic

I need to read message from topic1 completely and then read message from topic2. I will be receiving messages in these topic everyday once. I managed to stop reading messages from topic2 before reading all the messages in topic1, but this is happening for me only once when the server is started. Can someone help me with this scenario.

ListenerConfig code

@EnableKafka
@Configuration
public class ListenerConfig {

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;


    @Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "batch");
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "5");
        return props;
    }

    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setBatchListener(true);
        return factory;
    }

    @Bean("kafkaListenerContainerTopic1Factory")
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerTopic1Factory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.getContainerProperties().setIdleEventInterval(60000L);
        factory.setBatchListener(true);
        return factory;
    }

    @Bean("kafkaListenerContainerTopic2Factory")
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerTopic2Factory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setBatchListener(true);
        return factory;
    }

}

Listner code

@Service
public class Listener {

    private static final Logger LOG = LoggerFactory.getLogger(Listener.class);

    @Autowired
    private KafkaListenerEndpointRegistry registry;

    @KafkaListener(id = "first-listener", topics = "topic1", containerFactory = "kafkaListenerContainerTopic1Factory")
    public void receive(@Payload List<String> messages,
                        @Header(KafkaHeaders.RECEIVED_PARTITION_ID) List<Integer> partitions,
                        @Header(KafkaHeaders.OFFSET) List<Long> offsets)  {
        for (int i = 0; i < messages.size(); i++) {
            LOG.info("received first='{}' with partition-offset='{}'",
                    messages.get(i), partitions.get(i) + "-" + offsets.get(i));
        }
    }

    @KafkaListener(id = "second-listener", topics = "topic2", containerFactory = "kafkaListenerContaierTopic2Factory" , autoStartup="false" )
    public void receiveRel(@Payload List<String> messages,
                        @Header(KafkaHeaders.RECEIVED_PARTITION_ID) List<Integer> partitions,
                        @Header(KafkaHeaders.OFFSET) List<Long> offsets) {
        for (int i = 0; i < messages.size(); i++) {
            LOG.info("received second='{}' with partition-offset='{}'",
                    messages.get(i), partitions.get(i) + "-" + offsets.get(i));
        }
    }

    @EventListener()
    public void eventHandler(ListenerContainerIdleEvent event) {
        LOG.info("Inside event");
        this.registry.getListenerContainer("second-listener").start();
    }

Kindly help me in resolving , as this cycle should happen everyday. Reading topic1 message completely and then reading message from topic2.

Upvotes: 1

Views: 161

Answers (2)

Gary Russell
Gary Russell

Reputation: 174584

You are already using an idle event listener to start the second listener - it should also stop the first listener.

When the second listener goes idle; stop it.

You should be checking which container the event is for to decide which container to stop and/or start.

Then, using a TaskScheduler, schedule a start() of the first listener at the next time you want it to start.

Upvotes: 1

Mert Ozer
Mert Ozer

Reputation: 23

Topic in Kafka is an abstraction where stream of records are published. Streams are naturally unbounded, so they have a start but they do not have a defined end. For your case, first you need to clearly define what is the end of your topic1 and your topic2 so that you can stop/presume your consumers when needed. Maybe you know how many messages you will process for each topic, so you can use: position or commmited to stop one consumer and presume the other one in that moment. Or if you are using a streaming framework they usually have a session window where the framework detects a groups elements by sessions of activity. You can also prefer to put that logic into the application side so that you don't need to stop/start any consumer threads.

Upvotes: 0

Related Questions