[FLINK-32273] What is org.apache.flink.avro.generated.record_json?

• Type: Bug
• Status: Closed
• Priority: Major
• Resolution: Invalid
• Component/s: Formats (JSON, Avro, Parquet, ORC, SequenceFile), Table SQL / API
• Assignee: Unassigned
• Reporter: Tan Kim
• Created: 07/Jun/23 01:18
• Updated: 28/Jun/23 12:48
• Resolved: 28/Jun/23 12:48

Description

I'm trying to relay a topic from kafka to another kafka.

This is the original record in source topic.

"json": {
        "eventName": "event-ABC",
        ...
    } 

The source is json format and sink is avro format with confluent-schema registry.

Here is my code.

 

tableEnv.executeSql("CREATE TABLE source_table (..) WITH (
'connector'='kafka', 
'format'='json',
)")

tableEnv.executeSql("CREATE TABLE sink_table WITH (
'connector'='kafka',
'format'='avro-confluent',
..
) AS SELECT * FROM source_table") 

If I run this code without 'value.avro-confluent.subject' configuration, the record is something like this.

{
    "json": {
        "org.apache.flink.avro.generated.record_json": {
            "eventName": {
                "string": "event-ABC"
            },
           ..
         }
}      

I don't understand why flink-avro inserts "org.apache.flink.avro.generated.record_json" between `json` and `eventName`.

Also `eventName` is not just 'event-ABC' but `string: event-ABC`.

 

Is this bug? or something I missed?