class KafkaDatabaseHistory extends AbstractDatabaseHistory {

    @Override
    public boolean exists() {
        boolean exists = false;
        if (storageExists()) {
            try (KafkaConsumer<String, String> historyConsumer = new KafkaConsumer<>(consumerConfig.asProperties());) {
                checkTopicSettings(topicName);
                // check if the topic is empty
                Set<TopicPartition> historyTopic = Collections.singleton(new TopicPartition(topicName, PARTITION));

                Map<TopicPartition, Long> beginningOffsets = historyConsumer.beginningOffsets(historyTopic);
                Map<TopicPartition, Long> endOffsets = historyConsumer.endOffsets(historyTopic);

                Long beginOffset = beginningOffsets.entrySet().iterator().next().getValue();
                Long endOffset = endOffsets.entrySet().iterator().next().getValue();

                exists = endOffset > beginOffset;
            }
        }
        return exists;
    }

}