in src/main/java/com/epam/digital/data/platform/starter/restapi/service/StartupKafkaTopicsCreator.java [69:81]
private Set<String> getMissingTopicNames(AdminClient kafkaAdminClient) {
Set<String> existingTopics;
try {
existingTopics = kafkaAdminClient.listTopics().names()
.get(TOPIC_CREATION_TIMEOUT, TimeUnit.SECONDS);
} catch (Exception e) {
throw new CreateKafkaTopicException(String.format(
"Failed to retrieve existing kafka topics in %d sec", TOPIC_CREATION_TIMEOUT), e);
}
Set<String> requiredTopics = getRequiredTopics(kafkaProperties.getTopics());
requiredTopics.removeAll(existingTopics);
return requiredTopics;
}