|
15 | 15 | */ |
16 | 16 | package io.lenses.streamreactor.connect.azure.eventhubs.util; |
17 | 17 |
|
18 | | -import java.util.ArrayList; |
19 | | -import java.util.HashMap; |
| 18 | +import static io.lenses.streamreactor.common.util.StringUtils.getSystemsNewLineChar; |
| 19 | + |
| 20 | +import cyclops.control.Either; |
| 21 | +import io.lenses.kcql.Kcql; |
| 22 | +import io.lenses.streamreactor.common.exception.ConnectorStartupException; |
| 23 | +import java.util.Collection; |
20 | 24 | import java.util.List; |
21 | 25 | import java.util.Map; |
| 26 | +import java.util.Set; |
| 27 | +import java.util.function.Function; |
22 | 28 | import java.util.regex.Matcher; |
23 | 29 | import java.util.regex.Pattern; |
24 | | - |
25 | | -import org.apache.kafka.common.config.ConfigException; |
26 | | - |
27 | | -import io.lenses.kcql.Kcql; |
| 30 | +import java.util.stream.Collectors; |
| 31 | +import java.util.stream.Stream; |
| 32 | +import lombok.AccessLevel; |
| 33 | +import lombok.NoArgsConstructor; |
28 | 34 |
|
29 | 35 | /** |
30 | 36 | * Class that represents methods around KCQL topic handling. |
31 | 37 | */ |
| 38 | +@NoArgsConstructor(access = AccessLevel.PRIVATE) |
32 | 39 | public class KcqlConfigTopicMapper { |
33 | 40 |
|
34 | 41 | private static final String TOPIC_NAME_REGEX = "^[\\w][\\w\\-\\_\\.]*$"; |
35 | 42 | private static final Pattern TOPIC_NAME_PATTERN = Pattern.compile(TOPIC_NAME_REGEX); |
| 43 | + |
| 44 | + private static final String ERROR_DELIMITER = ";" + getSystemsNewLineChar(); |
36 | 45 | public static final String TOPIC_NAME_ERROR_MESSAGE = |
37 | 46 | "%s topic %s, name is not correctly specified (It can contain only letters, numbers and hyphens," |
38 | | - + " underscores and dots and has to start with number or letter"; |
| 47 | + + " underscores and dots and has to start with number or letter)"; |
39 | 48 |
|
40 | 49 | /** |
41 | | - * This method parses KCQL statements and fetches input and output topics checking against |
42 | | - * regex for invalid topic names in input and output. |
43 | | - * |
| 50 | + * This method parses KCQL statements and fetches input and output topics checking against regex for invalid topic |
| 51 | + * names in input and output. |
| 52 | + * |
44 | 53 | * @param kcqlString string to parse |
45 | 54 | * @return map of input to output topic names |
46 | 55 | */ |
47 | | - public static Map<String, String> mapInputToOutputsFromConfig(String kcqlString) { |
| 56 | + public static Either<ConnectorStartupException, List<Kcql>> mapInputToOutputsFromConfig(String kcqlString) { |
48 | 57 | List<Kcql> kcqls = Kcql.parseMultiple(kcqlString); |
49 | | - Map<String, String> inputToOutputTopics = new HashMap<>(kcqls.size()); |
50 | | - List<String> outputTopics = new ArrayList<>(kcqls.size()); |
51 | 58 |
|
52 | | - for (Kcql kcql : kcqls) { |
53 | | - String inputTopic = kcql.getSource(); |
54 | | - String outputTopic = kcql.getTarget(); |
| 59 | + List<String> inputTopics = kcqls.stream().map(Kcql::getSource).collect(Collectors.toUnmodifiableList()); |
| 60 | + List<String> outputTopics = kcqls.stream().map(Kcql::getTarget).collect(Collectors.toUnmodifiableList()); |
55 | 61 |
|
56 | | - if (!topicNameMatchesAgainstRegex(inputTopic)) { |
57 | | - throw new ConfigException(String.format(TOPIC_NAME_ERROR_MESSAGE, "Input", inputTopic)); |
58 | | - } |
59 | | - if (!topicNameMatchesAgainstRegex(outputTopic)) { |
60 | | - throw new ConfigException(String.format(TOPIC_NAME_ERROR_MESSAGE, "Output", outputTopic)); |
61 | | - } |
62 | | - if (inputToOutputTopics.containsKey(inputTopic)) { |
63 | | - throw new ConfigException(String.format("Input %s cannot be mapped twice.", inputTopic)); |
64 | | - } |
65 | | - if (outputTopics.contains(outputTopic)) { |
66 | | - throw new ConfigException(String.format("Output %s cannot be mapped twice.", outputTopic)); |
67 | | - } |
| 62 | + Set<String> allErrors = |
| 63 | + Stream.of( |
| 64 | + validateTopicMappings(inputTopics, "Input"), |
| 65 | + validateTopicMappings(outputTopics, "Output"), |
| 66 | + validateTopicName(inputTopics, "Input"), |
| 67 | + validateTopicName(outputTopics, "Output") |
| 68 | + ).flatMap(Collection::stream).collect(Collectors.toUnmodifiableSet()); |
68 | 69 |
|
69 | | - inputToOutputTopics.put(inputTopic, outputTopic); |
70 | | - outputTopics.add(outputTopic); |
| 70 | + if (!allErrors.isEmpty()) { |
| 71 | + return Either.left(new ConnectorStartupException( |
| 72 | + String.format("The following errors occurred during validation: %s", String.join(ERROR_DELIMITER, |
| 73 | + allErrors)))); |
71 | 74 | } |
72 | 75 |
|
73 | | - return inputToOutputTopics; |
| 76 | + return Either.right(List.copyOf(kcqls)); |
| 77 | + } |
| 78 | + |
| 79 | + private static List<String> validateTopicName(List<String> topicNames, String description) { |
| 80 | + return topicNames.stream() |
| 81 | + .filter(topicName -> !topicNameMatchesAgainstRegex(topicName)) |
| 82 | + .map(topicName -> String.format(TOPIC_NAME_ERROR_MESSAGE, description, topicName)) |
| 83 | + .collect(Collectors.toUnmodifiableList()); |
| 84 | + } |
| 85 | + |
| 86 | + private static List<String> detectDuplicateTopicMappings(List<String> topics) { |
| 87 | + return topics.stream() |
| 88 | + .collect(Collectors.groupingBy(Function.identity(), Collectors.counting())) |
| 89 | + .entrySet().stream() |
| 90 | + .filter(entry -> entry.getValue() > 1) |
| 91 | + .map(Map.Entry::getKey) |
| 92 | + .collect(Collectors.toUnmodifiableList()); |
| 93 | + } |
| 94 | + |
| 95 | + private static List<String> validateTopicMappings(List<String> topics, String description) { |
| 96 | + return detectDuplicateTopicMappings(topics).stream() |
| 97 | + .map( |
| 98 | + dupe -> String.format("%s '%s' cannot be mapped twice.", description, dupe) |
| 99 | + ).collect(Collectors.toUnmodifiableList()); |
74 | 100 | } |
75 | 101 |
|
76 | 102 | private static boolean topicNameMatchesAgainstRegex(String topicName) { |
|
0 commit comments