Hi @xav webmaster, i think this answer will help you for sure,
Sample Flow :-
GenerateFlowFile--> ExtractText-->UpdateAttribute-->PublishKafka
GenerateFlowFile:- As for testing purpose i'm using this but in your case you are having some other processors
ExtractText Processor:- in this processor i'm extracting the contents of flowfile as attribute.
Ex:- adult,dog,bulldog,9,23,male,brown,4,etc
The above content of flowfile by adding new property to ExtractText i'm going to extract the content and keeping that as attribute of the flowfile
cnt_attr as
(.*) //capture everything and add to ff as cnt_attr
Configs:-
Output of this processor:-
Every flowfile will associated with the attribute called cnt_attr to it, we can use this attribute in
UpdateAttribute Processor:-
To dynamically change the topic names based on the cnt_attr attribute, for this case we need to use Advanced Usage of Update Attribute processor.
Right Click on UpdateAttribute processor and click on Advanced Button in the lower right corner.
Steps:-
open above screenshot in new tab to see 1,2,3,4 steps and refer them with below steps
1. As mentioned in the above screenshot click on FlowFile Policy change to
UseOriginal
2. Click on + sign at Rules and give name as adult_dog
3. Click on + sign at Conditions and give our check condition in it
${cnt_attrt:matches('.*adult.*dog.*')}4. Click on + sign at Actions and give the attribute name as
kafka_topic
Value for the kafka_topic attribute as
adult_dog
New Rule:-
for cat_dog
conditions check is ${cnt_attr:matches('.*cat.*dog.*')} and Actions add attribute name as kafka_topic and value is cat_dog same as 2,3,4 steps above.
summarize all the steps:-
step 1 we are using original flowfile and step2 we are creating a rule and step3 adding conditions to check if the cnt_attr attribute satisfies or not step4 if it satisfies then adding kafka_topic attribute with desired name to it.
like this way we can add as many rules as we want in same UpdateAttribute Processor as you can see in my screenshot i have added 2 Rules(adult_dog,cat_dog).
This processor checks which Rule has satisfied and updates kafka_topic attribute with the mentioned name in it.
PublishKafka:-
use the kafka_topic attribute in Topic Name property of processor
${kafka_topic}
Flow Screenshot:-
In this way we can use only one UpdateAttribute to dynamically change the value of kafka_topic based on update attribute processor and use same kafka_topic attribute to publish messages to respective topics.
We can do that by using RouteonContent processor followed by UpdateAttribute processor.
Routeoncontent:-
change Match Requirement strategy to content must contain match
add the properties
adult_dog as (.*adult.*dog.*) cat_dog as (.*cat.*dog.*)
Config:-
Right now we are checking the contents of flowfile and add new properties for all the cases that you need to publish the data to kafka, then use
UpdateAttribute processor:
Add your deserved name to kafka_topic attribute of the flowfile.
add new property
kafka_topic as cat_dog
Configs:-
PublishKafka:-
every flowfile will have only kafka_topic as the attribute having different topic names in it, make use of kafka_topic in our publish kafka processor.
Change Topic Name property to
${kafka_topic}Configs:-
Flow screenshot:-
Flow Explanation:-
Routeoncontent //check the content and transfers flowfile to matching relationships. UpdataAttribute //adding kafka_topic attribute with some name. PublishKafka //using kafka_topic attribute and publishing the contents to respective topics dynamically.
Like this way we don't have to use any scripts and dynamically publish into kafka topics.
Hi @xav webmaster, i think this answer will help you for sure,
Sample Flow :-
GenerateFlowFile--> ExtractText-->UpdateAttribute-->PublishKafka
GenerateFlowFile:- As for testing purpose i'm using this but in your case you are having some other processors
ExtractText Processor:- in this processor i'm extracting the contents of flowfile as attribute.
Ex:- adult,dog,bulldog,9,23,male,brown,4,etc
The above content of flowfile by adding new property to ExtractText i'm going to extract the content and keeping that as attribute of the flowfile
cnt_attr as
(.*) //capture everything and add to ff as cnt_attr
Configs:-
Output of this processor:-
Every flowfile will associated with the attribute called cnt_attr to it, we can use this attribute in
UpdateAttribute Processor:-
To dynamically change the topic names based on the cnt_attr attribute, for this case we need to use Advanced Usage of Update Attribute processor.
Right Click on UpdateAttribute processor and click on Advanced Button in the lower right corner.
Steps:-
open above screenshot in new tab to see 1,2,3,4 steps and refer them with below steps
1. As mentioned in the above screenshot click on FlowFile Policy change to
UseOriginal
2. Click on + sign at Rules and give name as adult_dog
3. Click on + sign at Conditions and give our check condition in it
${cnt_attrt:matches('.*adult.*dog.*')}4. Click on + sign at Actions and give the attribute name as
kafka_topic
Value for the kafka_topic attribute as
adult_dog
New Rule:-
for cat_dog
conditions check is ${cnt_attr:matches('.*cat.*dog.*')} and Actions add attribute name as kafka_topic and value is cat_dog same as 2,3,4 steps above.
summarize all the steps:-
step 1 we are using original flowfile and step2 we are creating a rule and step3 adding conditions to check if the cnt_attr attribute satisfies or not step4 if it satisfies then adding kafka_topic attribute with desired name to it.
like this way we can add as many rules as we want in same UpdateAttribute Processor as you can see in my screenshot i have added 2 Rules(adult_dog,cat_dog).
This processor checks which Rule has satisfied and updates kafka_topic attribute with the mentioned name in it.
PublishKafka:-
use the kafka_topic attribute in Topic Name property of processor
${kafka_topic}Flow Screenshot:-
In this way we can use only one UpdateAttribute to dynamically change the value of kafka_topic based on update attribute processor and use same kafka_topic attribute to publish messages to respective topics.
