bbende commented on a change in pull request #3450: NIFI-1642 : Kafka
Processors Topic Name Validations
URL: https://github.com/apache/nifi/pull/3450#discussion_r277371399
##########
File path:
nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-10-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/KafkaProcessorUtils.java
##########
@@ -257,6 +288,54 @@
return results;
}
+
+ static final class KafkaConsumerTopicNameValidator implements Validator {
+
+ final Class<?> classType;
+
+ public KafkaConsumerTopicNameValidator(final Class<?> classType) {
+ this.classType = classType;
+ }
+
+ @Override
+ public ValidationResult validate(final String subject, final String
value, final ValidationContext context) {
+ final String topicType =
context.getProperty(CONSUMER_TOPIC_TYPE).evaluateAttributeExpressions().getValue();
+ final String topicListing =
context.getProperty(CONSUMER_TOPICS).evaluateAttributeExpressions().getValue();
+ boolean isValidTopicName = true;
+ String topicName = "";
+ if (topicListing != null &&
topicType.equals(TOPIC_NAME.getValue())) {
+ for (final String topic : topicListing.split(",", 100)) {
+ if(topic.isEmpty() ||
TOPIC_NAME_PATTERN.matcher(topic).replaceAll("").length() != 0 ) {
+ isValidTopicName = false;
+ topicName = topic;
+ }
+ }
+ }
+ return new
ValidationResult.Builder().subject(subject).explanation(topicName + " is not a
valid topic name.").valid(isValidTopicName).build();
+ }
+ };
+
+ static final class KafkaProducerTopicNameValidator implements Validator {
+
+ final Class<?> classType;
+
+ public KafkaProducerTopicNameValidator(final Class<?> classType) {
+ this.classType = classType;
+ }
+
+ @Override
+ public ValidationResult validate(final String subject, final String
value, final ValidationContext context) {
+ final String topicName =
context.getProperty(PRODUCER_TOPIC).evaluateAttributeExpressions().getValue();
Review comment:
I'm not sure if we can validate the producer topic like this due to the fact
that it supports EL from flow files. When EL from flow files is supported, flow
files attributes are checked before variable registry. So there could be a
variable name "kafka.topic" that is invalid, but at runtime the flow files have
an attribute named kafka.topic that is valid, but you can't know that until the
processor is running.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
