Uwe,
Since ConsumeKafkaRecord is a "source" processor, you won't be able to
set schema.name as a FlowFile attribute. However you could set it as a
property in a Variable Registry file and use that. Are your schemas
dynamic based on the topic? If not, you likely don't need to use the
schema.name attribute, you could hard-code the name in the reader
(${schema.name} is just the default), or even the schema itself (if
you use the Schema Text strategy and Schema Text set to your schema).
If the schemas depend on the topic, I'm not sure a single
ConsumeKafkaRecord would work in this case. Having said that, it might
be cool as an improvement to make a "transient attribute" available,
exposing "topic.name" as what might look like a flow file attribute to
the EL evaluator, and whose value would be the topic name for the
current incoming message.
Regards,
Matt
On Thu, Jun 22, 2017 at 3:40 PM, Uwe Geercken <[email protected]> wrote:
> Hello,
>
> besides my other problem with the ConsumeKafkaRecord_0_10 processor, I have
> another question.
>
> Using the AvroSchemaRegistry 1.3.0, I can define a schema and reference it
> e.g. in the CSVReader controller using the 'Schema Name' property and by
> setting this property to ${schema.name}.
>
> But how can I set this attribute ${schema.name}? Usually I would do so using
> the UpdateAttribute Processor but the ConsumeKafkaRecord_0_10 processor is
> the first one in the flow and it does not allow incomming connections.
>
> Thanks for any comments or help.
>
> Rgds,
>
> Uwe
