Hello...I am trying to convert my CSV file to get an AvroSchema by using my own custom processor which will get the file from the disk and converts it into flowfile and then passes it to the method named "inferAvroSchemaFromCSV" of the processor inferAvroSchema.After that i am getting my required schema successfully, which i am trying to transfer it into the success relationship so as to get on the disk, but i am getting following error -
"DBColumnMapper[id=3836e9a0-015b-1000-9d9b-3d14d4066516] Failed to infer Avro schema for StandardFlowFileRecord[uuid=785128f9-acea-4d1c-a3dc-ec1313c900a0,claim=StandardContentClaim [resourceClaim=StandardResourceClaim[id=1491380551546-1, container=default, section=1], offset=15934, length=797],offset=0,name=vikas,size=797] due to org.apache.nifi.processor.exception.FlowFileHandlingException: StandardFlowFileRecord[uuid=785128f9-acea-4d1c-a3dc-ec1313c900a0,claim=StandardContentClaim [resourceClaim=StandardResourceClaim[id=1491380551546-1, container=default, section=1], offset=15934, length=797],offset=0,name=vikas,size=797] is not the most recent version of this FlowFile within this session (StandardProcessSession[id=12]): org.apache.nifi.processor.exception.FlowFileHandlingException: StandardFlowFileRecord[uuid=785128f9-acea-4d1c-a3dc-ec1313c900a0,claim=StandardContentClaim [resourceClaim=StandardResourceClaim[id=1491380551546-1, container=default, section=1], offset=15934, length=797],offset=0,name=vikas,size=797] is not the most recent version of this FlowFile within this session (StandardProcessSession[id=12])
14:20:53 ISTERROR3836e9a0-015b-1000-9d9b-3d14d4066516"
Following is my workflow template.png and my configuration of custom processor dbcolumnmapper-configuration.png
Here is the Java Code--
@Override
public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException {
URI uri=null;
String url=context.getProperty(INPUT_SCHEMA).getValue();
StringBuilder out = new StringBuilder();
try {
uri = new URI(url);
} catch (URISyntaxException e) {
e.printStackTrace();
}
Path schemaPath = new Path(uri);
FileSystem fs;
InputStream in=null;
try {
fs = schemaPath.getFileSystem(DefaultConfiguration.get());
in= fs.open(schemaPath);
BufferedReader reader = new BufferedReader(new InputStreamReader(in));
String line;
while ((line = reader.readLine()) != null) {
out.append(line);
}
System.out.println("csv file data"+out.toString()); //Prints the string content read from input stream
reader.close();
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
FlowFile flowFile=null;
try{
//Writing csv data inside flow file for infer avro schema converter
final byte[] dataByte=String.valueOf(out).getBytes(StandardCharsets.UTF_8);
flowFile = session.create();
flowFile = session.write(flowFile, new OutputStreamCallback() {
@Override
public void process(final OutputStream output) throws IOException {
output.write(dataByte);
}
});
//flowFile = session.putAttribute(flowFile, CoreAttributes.MIME_TYPE.key(), "text/plain");
//flowFile = session.putAttribute(flowFile, CoreAttributes.FILENAME.key(), "vikas");
final AtomicReference<String> avroSchema = new AtomicReference<>();
avroSchema.set(inferAvroSchemaFromCSV(flowFile, context, session));
FlowFile avroSchemaFF = null;
avroSchemaFF = session.write(session.create(), new OutputStreamCallback() {
@Override
public void process(OutputStream out) throws IOException {
out.write(avroSchema.get().getBytes());
}
});
avroSchemaFF = session.putAttribute(avroSchemaFF, CoreAttributes.MIME_TYPE.key(), AVRO_MIME_TYPE);
avroSchemaFF = session.putAttribute(avroSchemaFF, CoreAttributes.FILENAME.key(), "newFlowFile");
session.transfer(avroSchemaFF, REL_ORIGINAL);
session.transfer(avroSchemaFF, SUCCESS);
//session.transfer(avroSchemaFF, FAILURE);
session.commit();
}catch (Exception ex) {
getLogger().error("Failed to infer Avro schema for {} due to {}", new Object[] {flowFile, ex});
}
}
