How to use JSON schema in Pulsar function?

[labuladong]

I use the go client to send messages to a Pulsar topic, with a JSON schema:

const eventJsonSchemaDef = `
{
  "type": "record",
  "name": "EventMessage",
  "namespace": "game",
  "fields": [
    {
      "name": "Type",
      "type": "string"
    },
    {
      "name": "Name",
      "type": "string"
    },
    {
      "name": "Avatar",
      "type": "string"
    },
    ......
`

producer, err := client.CreateProducer(pulsar.ProducerOptions{
	Topic:           topicName,
	DisableBatching: true,
	// use schema to confirm the structure of message
	Schema: pulsar.NewJSONSchema(eventJsonSchemaDef, nil),
})

I want to extract some files from the JSON message, so I wrote a Pulsar function using Java. But I don't know how to use function to handle JSON schema. The document https://pulsar.apache.org/docs/next/functions-develop-schema-registry and https://github.com/apache/pulsar/tree/master/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples don't have an example.

I try to use GenericRecord to handle JSON schema, but seems it doesn't work:

For now, I can only cast messages to JSON string in this way:

new String((byte[]) input.getNativeObject())

Do you have any suggestions for using schema in Pulsar functions? Thanks!

[labuladong]

I compile source code on master branch of Pulsar, then I can read JSON data with GenericJsonRecord. So this question is resolved by the new version.