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_r277386164
##########
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:
We can't change to variable registry because then you can't route different
flow files to different topics using a single PublishKafka processor, plus it
would be a behavioral change to existing flows that are relying on attributes,
so even if we wanted to change it, it could only be done on a major release
(2.x).
The only real option is to handle it in onTrigger. Typically you wouldn't
really do validation at that point, you would just check that after calling
context.getProperty(topic).evaluateAttributeExpressions(flowFile).getValue()
you have a non-blank value. If so then you try to publish to kafka, if an
exception happens route to failure.
----------------------------------------------------------------
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
