PublishKafkaRecord_0_10 1.2.0.3.0.1.1-5 Error: Unexpected Charter

[#1]

Hi

I am getting the below error when trying to pass a Json record into a Publish Kafka Record Process.

Unexpected character ('A'(code 65)): expected a vaild value (number, String, array, object, 'true','false' or 'null') at [source: java.io.stringReader@7070ee3e; line: 1, column: 2

See attached image error.png

Explanation:

I am trying to pass the below Json File into the PublishKafkaRecord.

Json File

{"storeId":"psi119745","dob":"20180905","printedCheckId":"90001","typeId":"212919","label":"Product A","id":"9437243","createdOn":"2018-09-05T04:20:49+12:00","quantity":1.0,"UniqueItemKey":"psi119745_20180905_90001_9437243"} 

I have created a schema in the schema registry called AllItems with the below definition.

AllItems Schema

{
 "type": "record",
 "namespace": "AllItems.trans",
 "name": "AllItems",
 "fields": [
  {
   "name": "storeId",
   "type": "string"
  },
  {
   "name": "dob",
   "type": "string"
  },
  {
   "name": "printedCheckId",
   "type": "string"
  },
  {
   "name": "typeId",
   "type": "string"
  },
  {
   "name": "label",
   "type": "string"
  },
  {
   "name": "id",
   "type": "string"
  },
  {
   "name": "createdOn",
   "type": "string"
  },
   {
   "name": "quantity",
   "type": "double"
  },
    {
   "name": "UniqueItemKey",
   "type": "string"
  }
 ]
}

Nifi data flow

1. Update Attributes

In this step I set the parameters

avro.schema = AllItems, kafka.topic = All_Items_Transactions and schema.name = AllItems

See attached image for properties process-update-attributes-properties.png

2. PublishKafkaRecord

Topic Name = ${kafka.topic}, Record Reader = JsonTreeReader and Record Writer = AvroRecordSetWriter

see attached image for properites process-publishkafkarecord-properties.png

JsonTreeReader, AvroRecordSetWriter and HortonworksSchemaRegistry properties serviceproperties.png

Thanks for any help

Tim

[#2 · Accepted solution]

The configuration of the reader and writer is not totally correct... you have selected the strategy as "Schema Text" which means the schema will come from the value of the "Schema Text" property which you then have set to the default value of ${avro.schema}, and then in UpdateAttribute you set avro.schema to "AllItems" so it is trying to parse the string "AllItems" into an Avro schema and failing because it is not a JSON schema.

If you want to use the "Schema Text" strategy then in UpdateAttribute the value of avro.schema needs to be the full text of your schema that you showed in your post.

If you want to use the schema from HWX schema registry, then the access strategy needs to be "Schema Name" and the "Schema Name" property needs to reference the name of the schema in the HWX schema registry (this part you already have setup correctly, so just changing the strategy should work).

[#3]

Hi Bryan

Thank you for you help. I was getting confused between the different Schema Strategies and your explanation helped a lot. I tried altering the strategy to "Schema Name", but sadly this resulted in a different error

"Failed to determine schema for writing: connection refused". 

I tried altering the HWX Schema registry and Kafka URLs, but without success.

I then decided to alter the UpdateAttribute process and set the avro.schema to the schema definition and used the strategy "Schema Text" and this worked. This will allow me to continue development and I shall look at implementing the "Schema Name" strategy at a later date.

Many thanks

[#4]

I think in the controller service for Hortonworks Schema Registry the URL needs to be http://localhost:7788/api/v1 where as in your screenshot it is missing the http://

Also if you can upgrade to HDF 3.2 there is an option in the writers to inherit the schema from the reader, so the writer doesn't even need to be configured with the schema registry, only the reader would need it.