This message was deleted.

Not yet that is a feature that has to be added to the NiFi processor itself.

is there something in works or on the road map for anytime soon, was thinking of extending AvroReader to talk to Pulsar’s inbuilt schema registry to fetch the schemas otherwise?

@David K - who would be a good contact to bounce ideas regarding this and maybe review a possible PR? (if we get there)

Well, since I am the original author and maintainer of that NiFi processor, I guess it would be me.

sounds good, hope you don't mind if we reach out to you with ideas to pick your brain when we get there

Sounds great. I am all about improving the processor and appreciate the input and PRs.

so in theory, within

ConsumePulsarRecord

line 266-7 instead of reading the schema from

RecordReaderFactory

as in

schema = getSchema(flowFile, readerFactory, data);

we do something like

if (lastMessage == null) {
                    flowFile = session.create();
                    flowFile = session.putAllAttributes(flowFile, currentAttributes);
                    schema = new SimpleRecordSchema(
                            new String(msg.getReaderSchema().get().getSchemaInfo().getSchema()),
                            "avro",
                            SchemaIdentifier.EMPTY
                    );
                    rawOut = session.write(flowFile);
                    writer = getRecordWriter(writerFactory, schema, rawOut, flowFile);

By making the reader property optional in the processor, the

AVROSetWriter

can continue without interruption, allowing the processor to operate based on embedded schemas. I have created a prototype that correctly generates a flow file with this modification. Do you anticipate any potential problems with this approach?

That seems reasonable for the Consumer processors, assuming that

msg.getReaderSchema()

is not null, etc. Test cases will have to be added for the different schema types in pulsar, e.g. KeyValue, GenericObject, primitives, etc.

@David K @Zsolt Kovacs please have a look at this PR pls https://github.com/david-streamlio/pulsar-nifi-bundle/pull/72

hi @David K, hopefully your holidays are relaxing. I have a quick question, seems in the nifi pulsar processor, you don't explicitly commit the flowfiles, just transfer them (https://github.com/david-streamlio/pulsar-nifi-bundle/blob/main/nifi-pulsar-proces[…]g/apache/nifi/processors/pulsar/pubsub/ConsumePulsarRecord.java) and then you rely on the nifi `AbstractProcessor`to commit it after the onTrigger finished (in an async way) but that will happen after you have committed to Pulsar already. So if the flowfile gets lost in the meantime, Pulsar will not re-send the messages -> we would experience data loss. Note: kafka consumer is doing the other way (first commits the session (writes flowfiles), then kafka offsets): https://github.com/apache/nifi/blob/main/nifi-nar-bundles/nifi-kafka-bundle/nifi-k[…]java/org/apache/nifi/processors/kafka/pubsub/ConsumerLease.java if you agree we can raise a PR for this cc @Umar Cheema

Let’s open a PR for this.

I will do it next week when I am back to work, thanks!

hi @David K hny, hope u had a good break! Thanks for accepting the above PR, now that we have had a chance to work on this a bit more was hoping to submit a few improvements. It seems we overlooked the verification of whether the schema was actually an Avro schema. Furthermore, we’ve come to realize that when there’s a schema update on a topic, it necessitates the creation of a new flow file. This is because we cannot mix multiple schemas within a single Parquet or Avro file. To address these concerns, we have prepared a new pull request. This update ensures that the avro.schema attribute is only set when the record schema is of Avro types. Additionally, we have refactored the code and added multiple tests to validate the correct handling of these scenarios. Furthermore, we have conducted manual testing of both JSON and Avro flows in NiFi with these changes to ensure they work effectively in real-world scenarios. We appreciate your assistance in reviewing and addressing these issues. Any who please check out the PR https://github.com/david-streamlio/pulsar-nifi-bundle/pull/74 fyi: @Zsolt Kovacs

Approved

Thank you sir