private void collectMaxPartitionsAndOffsets() {
        kafkaConsumer.partitionsFor(topic, pts -> {
            if (pts.succeeded()) {
                kafkaConsumer.endOffsets(pts.result().stream().map(p -> new TopicPartition().setTopic(topic).setPartition(p.getPartition())).collect(Collectors.toSet()),
                        res -> {
                            if (res.succeeded()) {
                                maxTopicOffsetsRegistry = res.result().entrySet().stream()
                                        .collect(Collectors.toConcurrentMap(
                                                e -> e.getKey().getPartition(),
                                                Map.Entry::getValue));
                                logger.info("Consumer={}, Added endOffset data for partitions registry={}", consumerId, maxTopicOffsetsRegistry);
                            }
                        }
                );
            }
            if (pts.failed()) {
                logger.error("Consumer={}. Failed to collect available partitions", consumerId);
                pts.cause().printStackTrace();
            }
        });
    }

  consumer.partitionsFor(topic, pts -> {
            if (pts.succeeded()) {
                Set<TopicPartition> topicPartitions = pts.result().stream().map(p -> new TopicPartition().setTopic(topic).setPartition(p.getPartition())).collect(Collectors.toSet());
                topicPartitions.forEach(tp -> {
                    consumer.committed(tp, cmt -> {
                        if (cmt.succeeded()) {
                            Long committedOffset = cmt.result().getOffset();
                            Integer topicPartition = tp.getPartition();
                            committedPartitionOffsets.put(topicPartition, committedOffset);
                        }
                        if (cmt.failed()) {
                            logger.error("Failed to fetch commited partitions list");
                            cmt.cause().printStackTrace();
                        }
                    });
                });