I am trying to read data from a kafka topic that consists of json messages with about 4 fields. The ExecutionProcess has the following script and it returns the following error: Failed to insert into influxDB due to {"error":"unable to parse 'ops_inbound, messageType=[object Object], if (flowFile != null) {
var StreamCallback = Java.type("org.apache.nifi.processor.io.StreamCallback");
var IOUtils = Java.type("org.apache.commons.io.IOUtils");
var StandardCharsets = Java.type("java.nio.charset.StandardCharsets");
var transformed_message = {};
var error = false;
var line = "ops_inbound";
// Get attributes
flowFile = session.write(flowFile, new StreamCallback(function (inputStream, outputStream) {
// Read input FlowFile content
var content = IOUtils.toString(inputStream, StandardCharsets.UTF_8); // message or content
var message_content = {};
try {
message_content = JSON.parse(content);
transformed_message.type = (((message_content || {}).message_type || {}));
transformed_message.post_state = (((message_content || {}).message_body || {}).type || "NULL");
line = line + ",messageType=json" +
", postID=NULL" + ",contentType=NULL" + ",published=NULL" + ",crawled=NULL" + ",custID=NULL" +
" value=" + "1"
+ " " + Date.now() * 1000000;
// Write output content
if (transformed_message) {
outputStream.write(line.getBytes(StandardCharsets.UTF_8));
}
} catch (e) {
error = true;
outputStream.write(content.getBytes(StandardCharsets.UTF_8));
}
}));
if (transformed_message.post_state) {
flowFile = session.putAttribute(flowFile, "type", transformed_message.type);
}
if (error) {
session.transfer(flowFile, REL_FAILURE)
} else {
session.transfer(flowFile, REL_SUCCESS)
}
}
... View more
