@Michal R Below is the code for CSVtoAvroConverter
You need to add relationship to your custom processor private static final Relationship INVALID = new Relationship.Builder()
.name("invalid")
.description("failed records with error appended").build();
final Set<Relationship> relationship = new HashSet<>();
relationship.add(INVALID);
//The current code already captures the error count as
long errors = failures.count();
//Here is a sample code snippet(I have not implemented it - Just to give you an idea)
if (written.get() > 0L && errors == 0L) {
session.remove(badRecords);
session.transfer(outgoingAvro, SUCCESS);
} else {
session.remove(outgoingAvro);
/*This is the code which adds an attribute to the flow file. You can have this as a string and append it to your records instead of having it as an attribute*/
if (errors > 0L) {
getLogger()
.warn("Failed to convert {}/{} records from CSV to Avro",
new Object[] { errors, errors });
badRecords = session.putAttribute(badRecords, "errors",
failures.summary());
} else {
badRecords = session.putAttribute(badRecords, "errors",
"No incoming records");
}
//Here, you are writing the contents into the flow file
WriteContentsToFlowFile.writeContentsToFlowFile(session,
badRecords, failurereason, INVALID); //You can add something like failure reason
Hope this helps
... View more
