private void startPollPartitionsTask() {
        vertx.setPeriodic(pollDelay, timerId -> {
            timerIds.put(timerId, true);
            kafkaConsumer.poll(Duration.ofMillis(pollDelay), ar1 -> {
                if (ar1.succeeded()) {
                    KafkaConsumerRecords<String, T> records = ar1.result();

                    // we need to wait for processing all records before commit
                    CountDownLatch batchLatch = new CountDownLatch(records.size());
 
                    logger.info("Consumer={}. Processing {} records from the topic={}", consumerId, records.size(), topic);
                    for (int i = 0; i < records.size(); i++) {
                        KafkaConsumerRecord<String, T> record = records.recordAt(i);
                        vertx.executeBlocking(promise -> executeOnRecordReceivedAction(record, promise, recordReceivedAction), res -> {

                            // signal that one record has been processed
                            batchLatch.countDown();

                            if (res.result() instanceof Exception) {
                                logger.error("Consumer: {}. Fail to ingest entity. key: {}. Stopping process", consumerId, record.key());
                                ((Exception) res.result()).printStackTrace();
                                killConsumer();
                                logger.error("Consumer: {}. Polling process is stopped", consumerId);
                            }
                        });
                    }

                    // wait all records to be processed
                    batchLatch.await();

                    kafkaConsumer.commit();
                }
                if (ar1.failed()) {
                    logger.error("Consumer={}. Polling process failed", consumerId);
                    ar1.cause().printStackTrace();
                }
            });
        });
    }