java.lang.IllegalArgumentException: A consumerFactory is required

This configuration existed before my changes:

@Configuration
public class KafkaConfiguration {

    @Value(value = "${kafka.bootstrapServers:XXX}")
    private String bootstrapServers;

    @Value(value = "${kafka.connect.url:XXX}")
    public String kafkaConnectUrl;

    @Bean
    public AdminClient adminClient() {
        Map<String, Object> configs = new HashMap<>();
        configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);

        return AdminClient.create(configs);
    }

    @Bean
    public WebProperties.Resources resources() {
        return new WebProperties.Resources();
    }

     @Bean
    public KafkaConnectClient connectClient() {
        return new KafkaConnectClient(new org.sourcelab.kafka.connect.apiclient.Configuration(kafkaConnectUrl));
    }
}

I've setup the app so that I can use KafkaTemaplate:

application.yaml

    spring:
      profiles:
        active: dev
    
      kafka:
        bootstrap-servers: XXX
        properties:
          schema.registry.url: XXX
        producer:
          key-serializer: org.apache.kafka.common.serialization.StringSerializer
          value-serializer: io.confluent.kafka.serializers.protobuf.KafkaProtobufSerializer
        consumer:
          group-id: kafka-management-service-group-id
          auto-offset-reset: earliest
          key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
          value-deserializer: io.confluent.kafka.serializers.protobuf.KafkaProtobufDeserializer

I can correctly write and read messages using KafkaTemplate from a topic with 1 partition that I created manually:

@GetMapping("/loadData")
public void loadData() {
    IntStream.range(0, 1000000).forEach(key ->
        {
            final General general = General.newBuilder()
                .setUuid(StringValue.newBuilder().setValue(String.valueOf(key))
                    .build()).build();
            template.send("test",
                String.valueOf(key),
                ScheduledJob.newBuilder().setGeneral(general).build()
            );
        }
    );
}

@GetMapping("/readData")
public void readData() {
    Collection<TopicPartitionOffset> topicPartitionOffsets = new ArrayList<>();
    final TopicPartitionOffset topicPartitionOffset = new TopicPartitionOffset("test", 0, SeekPosition.BEGINNING);
    topicPartitionOffsets.add(topicPartitionOffset);
    final Iterator<ConsumerRecord<String, ScheduledJob>> iterator = template.receive(topicPartitionOffsets).iterator();
    while (iterator.hasNext()) {
        final ConsumerRecord<String, ScheduledJob> next = iterator.next();
        System.out.println("Key = "+next.key());
        System.out.println("Value = "+next.value());
        System.out.println();
    }
}

Current task: I'd like to read all messages of a topic from a specific partition X (therefore I need the number of partitions), from a specific offet Y to a specific offset Z; and ignore the rest.

I've read that I can do something along these lines to get the partition number for a specific topic:

@Autowired
private final ConsumerFactory<Object, Object> consumerFactory;

public String[] partitions(String topic) {
    try (Consumer<Object, Object> consumer = consumerFactory.createConsumer()) {
        return consumer.partitionsFor(topic).stream()
            .map(pi -> "" + pi.partition())
            .toArray(String[]::new);
    }
}

Problem: Despite the fact that I can read and write messages, a ConsumerFactory isn't created automatically.

Context: Our frontend needs to support paging, multi-column order and filtering. The source are Kafka messages from a generic topic with any number of partitions. Maybe this isn't the best way to do it, therefore I'm also open for suggestions