title: Spring Cloud Stream Manual Poller Kafka
body: <p>I don't want to use @KafkaListener or @StreamListener but I want to manually poll kafka. I am using spring-cloud-starter-stream-kafka library and I have the following Kafka Producer</p>
  
  <pre><code>  @Autowired
    private KafkaTemplate&lt;byte[], byte[]&gt; template;
  
    public void sendMessages() {
      IntStream.range(2)
               .forEach(val -&gt; {
                 template.send("kafka-topic", "hello".getBytes());
               });
    }
  </code></pre>
  
  <p>I would like to manually poll the same kafka topic using spring-kafka. I tried the following consumer</p>
  
  <pre><code> @Autowired
    private ConsumerFactory consumerFactory;
  
    public void processKafkaRecords() throws InterruptedException {
      Consumer&lt;byte[], byte[]&gt; consumer = consumerFactory.createConsumer("0", "consumer-1");
      consumer.subscribe(Arrays.asList("kafka-topic"));
      ConsumerRecords&lt;byte[], byte[]&gt; poll = consumer.poll(Duration.ofMillis(1000));
      poll.forEach(record -&gt; {
        log.info("record {}", record);
      });
    }
  </code></pre>
  
  <p>application.properties</p>
  
  <pre><code>spring.cloud.stream.bindings.pollableInput.destination=kafka-topic
  spring.cloud.stream.bindings.pollableInput.group=kafka-topic
  spring.cloud.stream.bindings.pollableInput.consumer.batch-mode=true
  spring.cloud.stream.bindings.pollableInput.consumer.header-mode=none
  spring.cloud.stream.bindings.pollableInput.consumer.use-native-decoding=true
  
  spring.cloud.stream.kafka.bindings.pollableInput.consumer.autoCommitOffset=false
  </code></pre>
  
  <p>However, the consumer never gets any records which the producer had sent. Any ideas how to manually poll a kafka topic?</p>