Kafka message field is being nested when type has null or default is present

user_12342700 0
Viewed 472

I am using Avro serialiser to push messages to kafka topic. I generated the Java class out of the below avro schema

 {
    "type": "record",
    "name": "CommonHeader",
    "fields": [
      {
        "name": "domain",
        "type": "string"
      },
      {
        "name": "eventSource",
        "type": "string"
      },
      {
        "name": "eventEntity",
        "type": "string"
      },
      {
        "name": "eventType",
        "type": "string"
      },
      {
        "name": "tsEventPublished",
        "type": "string"
      },
      {
        "name": "tsEntityCreated",
        "type": [
          "null",
          "string"
        ],
        "default": null
      },
      {
        "name": "tsEntityUpdated",
        "type": [
          "null",
          "string"
        ],
        "default": null
      },
      {
        "name": "isSynthetic",
        "type": [
          "boolean"
        ],
        "default": false
      },
      {
        "name": "correlationId",
        "type": "string"
      }
    ]
  }

The generated class is attached here.CommonHeader.java

When I write the message to kafka, the message looks like this

    "domain": "product",
    "eventSource": "source",
    "eventEntity": "entity",
    "eventType": "Create",
    "tsEventPublished": "2021-11-04T02:39:42.261Z",
    "tsEntityCreated": {
      "string": "2021-11-21T21:25:04.000Z"
    },
    "tsEntityUpdated": {
      "string": "2011-11-04T00:11:52.000Z"
    },
    "isSynthetic": {
      "boolean": false
    },
    "correlationId": "id"
  }

why are the values of "isSynthetic" and "tsEntityUpdated" are nested? How can I avoid this?

Spring boot properties for kafka:

spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=io.confluent.kafka.serializers.KafkaAvroSerializer
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=io.confluent.kafka.serializers.KafkaAvroDeserializer
spring.kafka.consumer.auto-offset-reset=latest
spring.kafka.properties.auto.register.schemas=false
spring.kafka.properties.specific.avro.reader=true

Plugin info:

            <plugin>
                <groupId>org.apache.avro</groupId>
                <artifactId>avro-maven-plugin</artifactId>
                <version>${avro.version}</version>
                <executions>
                    <execution>
                        <id>schemas</id>
                        <phase>generate-sources</phase>
                        <goals>
                            <goal>schema</goal>
                            <goal>protocol</goal>
                            <goal>idl-protocol</goal>
                        </goals>
                        <configuration>
                            <sourceDirectory>${project.basedir}/src/main/resources/avro</sourceDirectory>
                            <outputDirectory>${project.basedir}/src/main/java/</outputDirectory>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
apache-kafka avro spring java
1 Answers

This is how Avro works based on official documentation. The fields tsEntityCreated, tsEntityUpdated

"type": [
   "null",
   "string"
]

and isSynthetic

"type": [
   "boolean"
]

defined as Unions. There are special rules for unions for JSON encoding:

The value of a union is encoded in JSON as follows:

  • if its type is null, then it is encoded as a JSON null;
  • otherwise it is encoded as a JSON object with one name/value pair whose name is the type's name and whose value is the recursively encoded value. For Avro's named types (record, fixed or enum) the user-specified name is used, for other types the type name is used.

For example, the union schema ["null","string"] would encode:

  • null as null;
  • the string "a" as {"string": "a"};
user_44510100
Related