I am trying to build a custom processor and have an avro record set incoming from a ConvertRecord processor which transforms my CSV to Avro. The processor says that conversion was done correctly , but when I try to unpack the DataFileStream of GenericRecord type into individual records the processor gives me an IOException that simply says 'Not a data file.' My scala code is as follows: override def onTrigger(context: ProcessContext, session: ProcessSession): Unit = {
val sourceFlowFile = Option(session.get)
sourceFlowFile match {
case Some(f) =>
//some property loading, processing and logging..
val transferList = new ListBuffer[FlowFile]
session.read(f, new InputStreamCallback {
override def process(in: InputStream) {
val stream: Try[DataFileStream[GenericRecord]] = Try(new DataFileStream(in, new GenericDatumReader[GenericRecord]()))
stream match {
case Failure(e) =>
getLogger.error(s"${e.getMessage}: error reading from data file stream, passing to relationship:failure", e)
failed = true
case Success(dataStream) =>
getLogger.info("Reading data from stream was a success, will proceed to writing")
while (dataStream.hasNext) {
//some processing
}
if (failed) {
session.transfer(f, RelFailure)
return
}
for (recordFlowFile <- transferList) {
writeToFlowFile(session, recordFlowFile)
}
getLogger.info("Writing was a success, passing list of generic records to success relationship")
session.transfer(transferList, RelSuccess)
case None =>
//some error handling
}
} not sure why this is happening because the way I am trying to read the stream is the way they test the writer in the nifi codebase. This happens when I try to create the DataFileStream from the InputStream. The match always results in Failure, so what Success does is irrelevant at present. Am I misunderstanding the way the stream callbacks work? I have been checking the codebase for people doing similar things in their processors, but couldn't find much except for the tests. Please pardon my formatting I'm not really sure how to fix it.
... View more
