Hello everyone,
I am currently facing an issue with integrating NiFi and Kafka (sending data to Kafka).
I am using the latest version of the NiFi docker image (apache/nifi, version 1.5) with Kafka confluent docker image (confluentinc/cp-kafka:4.0.0).
I am trying to send specific data to Kafka using the publishKafkaRecord_1_0 processor. However I was yet not able to succeed due to an IllegalTypeConversion error that is described further down this post.
Here is the NiFi template with all the processors and controller services configuration:
GenerateFlowFile config :
The custom generated text looks like this:
col_1;col_2;col_3;col_4; etc. 36531253;4787ea68-4276-4b3b-b154-d70419d23113;https://www.dummyexample.com;My Dummy website, description;?;365311753;2018-01-02T07:08:40Z;https://www.dummyexample.com/fr/openspace/loggin?axes4=priv;1;1;1;_15148769143240.5030172901622478_;?
The header line contains the column names.The "?" fields are unknown fields.
UpdateAttribute config :
I put the schema name in the schema.name attribute.
PublishKafkaRecord_1_0 config :
As for the controller services, I am using CSVReader for the record reader and JSONRecordSetWriter for the record writer.
Both use the Schema Name property for the access strategy (specified in the attribute thanks to the updateAttribute processor).
Both refer to a local Avro Schema Registry set up with the following schema :
{
"type": "record",
"name": "page_record",
"fields" : [
{"name": "session_number", "type": "bytes", "logicalType": "decimal", "precision": 20},
{"name": "tracking_uuid", "type": "bytes"},
{"name": "page_location_domain", "type": "string"},
{"name": "page_title", "type": "string"},
{"name": "referring_page_instance_id", "type": "bytes", "logicalType": "decimal", "precision": 20},
{"name": "page_instance_id", "type": "bytes", "logicalType": "decimal", "precision": 20},
{"name": "event_timestamp", "type": "long", "logicalType": "timestamp-millis"},
{"name": "page_location", "type": "string"},
{"name": "attribution_sequence_in_session", "type": "long"},
{"name": "page_sequence_in_session", "type": "long"},
{"name": "page_sequence_in_attribution", "type": "long"},
{"name": "top_level_window_id", "type": "string"},
{"name": "profile_uuid", "type": "string"}
]
}N.B : (col_1, col_2 etc... are not the real names of the columns. One column name example is "session_number").
When going through the publishKafkaRecord_1_0 processor, the flowfiles cannot be sent to Kafka. Here is the error displayed by NiFi :
PublishKafkaRecord_1_0[id=22b8f3a8-32cc-39f2-1151-6eb4291aa590] Failed to send StandardFlowFileRecord[uuid=d7ab2519-d8b8-419b-8e0c-6d0704f36a97,claim=StandardContentClaim [resourceClaim=StandardResourceClaim[id=1522227238377-1, container=default, section=1], offset=1338, length=525],offset=0,name=12260221790638,size=525] to Kafka: org.apache.nifi.serialization.record.util.IllegalTypeConversionException: Cannot convert value [session_number] of type class java.lang.String to Object Array for field session_number
The problem is the same when I use the Confluent schema-registry instead of a local Avro schema registry.
I have also tried to use an AvroRecordSetWriter for the record writer but the result is the same : I still get an IllegalTypeConversion error from String to ByteBuffer.
Do you have any idea what can be the issue ?
Thank you in advance.
