If you want to avoid duplicates you could hash the content of the files and leverage the DetectDuplicate processor to only insert the unique files into your DB. ... View more

Agree with @SAMSAL's approach and if you can provide a parameter or something in the header or request so your API returns a JSON response each time it'll make things a lot easier for you to parse and build the request for the next step in your flow. ... View more

...a 3rd option because I like scripted processors 😂...using ExcecuteGroovyScript   import groovy.json.JsonOutput import groovy.json.JsonSlurper import java.nio.charset.StandardCharsets JsonSlurper jsonSlurper = new JsonSlurper() JsonOutput jsonOutput = new JsonOutput() FlowFile flowFile = session.get() if(!flowFile) return flowFile = session.write(flowFile, { inputStream, outputStream -> List<Map> data = jsonSlurper.parse(inputStream) data.each { it.order_item = jsonSlurper.parseText(it.order_item) } outputStream.write(jsonOutput.toJson(data).getBytes(StandardCharsets.UTF_8)) } as StreamCallback) session.transfer(flowFile, REL_SUCCESS)   Looks like a lot but this is what takes the string JSON and converts it to JSON: it.order_item = jsonSlurper.parseText(it.order_item) ... View more

PublishKafka must have an active connection to Kafka before it even attempts to send a FlowFile which mean it never even gets into the block of code that sends it and routes it to "success" or "failure" accordingly. Making sure your Kafka cluster is up and running should be the focus if this is what you're experiencing. My guess if this "error" topic you have is on the same Kafka cluster then, even if PublishKafka was able to route a FlowFile to "failure" when it's unable to connect to Kafka, it wouldn't work anyways. ... View more

How about SplitJson $[*] followed by EvaluateJson $.SEARCH_RESULT ... View more

I think you could probably use EvaluateJsonPath to parse the JSON value for "SEARCH_RESULT" but I like scripted processors so I would use a Groovy based InvokeScriptedProcessor with this code   import groovy.json.JsonOutput import groovy.json.JsonSlurper import java.nio.charset.StandardCharsets class GroovyProcessor implements Processor { PropertyDescriptor BATCH_SIZE = new PropertyDescriptor.Builder() .name("BATCH_SIZE") .displayName("Batch Size") .description("The number of incoming FlowFiles to process in a single execution of this processor.") .required(true) .defaultValue("100") .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR) .build() Relationship REL_SUCCESS = new Relationship.Builder() .name("success") .description('FlowFiles that were successfully processed are routed here') .build() Relationship REL_FAILURE = new Relationship.Builder() .name("failure") .description('FlowFiles that were not successfully processed are routed here') .build() ComponentLog log JsonSlurper jsonSlurper = new JsonSlurper() JsonOutput jsonOutput = new JsonOutput() void initialize(ProcessorInitializationContext context) { log = context.logger } Set<Relationship> getRelationships() { Set<Relationship> relationships = new HashSet<>() relationships.add(REL_FAILURE) relationships.add(REL_SUCCESS) return relationships } Collection<ValidationResult> validate(ValidationContext context) { } PropertyDescriptor getPropertyDescriptor(String name) { } void onPropertyModified(PropertyDescriptor descriptor, String oldValue, String newValue) { } List<PropertyDescriptor> getPropertyDescriptors() { List<PropertyDescriptor> descriptors = new ArrayList<>() descriptors.add(BATCH_SIZE) return Collections.unmodifiableList(descriptors) } String getIdentifier() { } void onScheduled(ProcessContext context) throws ProcessException { } void onUnscheduled(ProcessContext context) throws ProcessException { } void onStopped(ProcessContext context) throws ProcessException { } void setLogger(ComponentLog logger) { } void onTrigger(ProcessContext context, ProcessSessionFactory sessionFactory) throws ProcessException { ProcessSession session = sessionFactory.createSession() try { List<FlowFile> flowFiles = session.get(context.getProperty(BATCH_SIZE).asInteger()) if (!flowFiles) return flowFiles.each { FlowFile flowFile -> Map customAttributes = [ "mime.type": "application/json" ] flowFile = session.write(flowFile, { inputStream, outputStream -> List<Map> searchResults = jsonSlurper.parse(inputStream) searchResults = searchResults.collect { jsonSlurper.parseText(it.SEARCH_RESULT) } outputStream.write(JsonOutput.toJson(searchResults).getBytes(StandardCharsets.UTF_8)) } as StreamCallback) session.putAllAttributes(flowFile, customAttributes) session.transfer(flowFile, REL_SUCCESS) } session.commit() } catch (final Throwable t) { log.error('{} failed to process due to {}; rolling back session', [this, t] as Object[]) session.rollback(true) throw t } } } processor = new GroovyProcessor() It looks like a lot but most of it is just boilerplate with the actual work being done here: ...and the output   ... View more

I don't see EvaluateJsonPath give options so it might be something you'd have to handle on your own...personally I'd do it via a Groovy scripted processor for greater control and performance. ... View more