I have Kafka many topics. And there is a Consumer Kafka for reading entries from these topics:

public Set<ConsumerRecord> consumeKafka() { consumer.subscribe(topics); Set<ConsumerRecord> resultRecords = new HashSet<>(); int i = 0; while (i++ < topicIteration) { ConsumerRecords<Object, Object> records = consumer.poll(100); System.out.println(records.partitions()); for (ConsumerRecord consumerRecord : records){ resultRecords.add(consumerRecord); } } return resultRecords; } private Consumer consumerInit(String consumerId){ props = new Properties(); props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId); props.put(ConsumerConfig.CLIENT_ID_CONFIG, consumerId); props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class.getName()); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class.getName()); props.put(KafkaAvroDeserializerConfig.SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryUrl); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); if (isActualizationTopics) { props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1); } else { props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, topicLimit); } return new KafkaConsumer<>(props); }

In order not to read all the topics in a row, I need to sort the list of topics, and receive data from those from which the data has not been read for the longest. I understand that I need to get the date of the last offset for each topic and sort it? How can this be done? Or maybe there are other approaches to the solution?

    0