[Bug] [connector-kafka] Unable to parse Avro-formatted Kafka data

[moody1117]

Search before asking

• I had searched in the issues and found no similar issues.

What happened

The Flink program can successfully parse Avro-formatted Kafka data, but when parsing via Seatunnel, serialization fails. Is there an issue with the schema configuration?

SeaTunnel Version

2.3.12

SeaTunnel Config

env {
  parallelism = 1
  job.mode = "BATCH"
}

source {
  Kafka {
    topic = "t_event"
    bootstrap.servers = "xxx:9092"
    consumer.group = "test_avro_kafka"
    start_mode = "timestamp"
    start_mode.timestamp = "1769504419000"
    start_mode.end_timestamp = "1769504479000"
    format = "avro"
    format_error_handle_way = "skip"  
    plugin_output = "kafka_table"

    kafka.config = {
      enable.auto.commit = "true"
      max.poll.records = "10000"
    }

    schema = {
      fields {
        dev_id        = string
        dev_ip        = string
        event_id      = string
        timestamp     = bigint
        s_ipv4        = string
        s_ipv6        = string
        s_port        = int
        d_ipv4        = string
        d_ipv6        = string
        d_port        = int
        sid           = int
        severity      = int
        alarm_name    = string
        alarm_type    = string
        classification = string
        protocol      = string
        domain_name   = string
        request_method = string
        protocol_version = string
        request_uri   = string
        request_url   = string
        referer       = string
        user_agent    = string
        request_header = string
        request_body  = string
        response_code = string
        response_header = string
        response_body = string
        XFF           = string
        extra_info    = string
        s_country     = string
        s_province    = string
        s_city        = string
        s_operator    = string
        d_country     = string
        d_province    = string
        d_city        = string
        d_operator    = string
        payload       = string
        attack_result = int
        attack_chain  = int
        description   = string
        solution      = string
        original_info = string
        pcap_info     = string
      }
    }
  }
}

sink {
  Console {
    plugin_input = "kafka_table"
  }
  LocalFile {
    plugin_input = "kafka_table"
    path = "/data/seatunnel/file"
    file_format_type = "json"
    custom_filename = true
    filename_time_format = "yyyy.MM.dd"
    is_enable_transaction = true
    file_name_expression = "${transactionId}_${now}"
   }
}

Running Command

seatunnel.sh --config $SEATUNNEL_HOME/job/saslkafka2file.conf  -m local  -DJvmOption="-Xms2G -Xmx2G"

Error Exception

2026-01-28 10:14:32,418 WARN  [a.s.c.s.k.s.KafkaRecordEmitter] [BlockingWorker-TaskGroupLocation{jobId=1068715656217624577, pipelineId=1, taskGroupId=3}] - Deserialize m

Zeta or Flink or Spark Version

zeta

Java or Scala Version

1.8

Screenshots

avro schema:

{
  "type": "record",
  "name": "AvroEvent",
  "namespace": "safe.serialize",
  "fields": [
    {
      "name": "dev_id",
      "type": [
        {
          "type": "string",
          "avro.java.string": "String"
        },
        "null"
      ]
    },
    {
      "name": "dev_ip",
      "type": [
        {
          "type": "string",
          "avro.java.string": "String"
        },
        "null"
      ]
    },
    {
      "name": "event_id",
      "type": [
        {
          "type": "string",
          "avro.java.string": "String"
        },
        "null"
      ]
    },
    {
      "name": "timestamp",
      "type": [
        "long",
        "null"
      ]
    },
    {
      "name": "s_ipv4",
      "type": [
        {
          "type": "string",
          "avro.java.string": "String"
        },
        "null"
      ]
    },
    {
      "name": "s_ipv6",
      "type": [
        {
          "type": "string",
          "avro.java.string": "String"
        },
        "null"
      ]
    },
    {
      "name": "s_port",
      "type": [
        "int",
        "null"
      ]
    },
    {
      "name": "d_ipv4",
      "type": [
        {
          "type": "string",
          "avro.java.string": "String"
        },
        "null"
      ]
    },
    {
      "name": "d_ipv6",
      "type": [
        {
          "type": "string",
          "avro.java.string": "String"
        },
        "null"
      ]
    },
    {
      "name": "d_port",
      "type": [
        "int",
        "null"
      ]
    },
    {
      "name": "sid",
      "type": [
        "int",
        "null"
      ]
    },
    {
      "name": "severity",
      "type": [
        "int",
        "null"
      ]
    },
    {
      "name": "alarm_name",
      "type": [
        {
          "type": "string",
          "avro.java.string": "String"
        },
        "null"
      ]
    },
    {
      "name": "alarm_type",
      "type": [
        {
          "type": "string",
          "avro.java.string": "String"
        },
        "null"
      ]
    },
    {
      "name": "classification",
      "type": [
        {
          "type": "string",
          "avro.java.string": "String"
        },
        "null"
      ]
    },
    {
      "name": "protocol",
      "type": [
        {
          "type": "string",
          "avro.java.string": "String"
        },
        "null"
      ]
    },
    {
      "name": "domain_name",
      "type": [
        {
          "type": "string",
          "avro.java.string": "String"
        },
        "null"
      ]
    },
    {
      "name": "request_method",
      "type": [
        {
          "type": "string",
          "avro.java.string": "String"
        },
        "null"
      ]
    },
    {
      "name": "protocol_version",
      "type": [
        {
          "type": "string",
          "avro.java.string": "String"
        },
        "null"
      ]
    },
    {
      "name": "request_uri",
      "type": [
        {
          "type": "string",
          "avro.java.string": "String"
        },
        "null"
      ]
    },
    {
      "name": "request_url",
      "type": [
        {
          "type": "string",
          "avro.java.string": "String"
        },
        "null"
      ]
    },
    {
      "name": "referer",
      "type": [
        {
          "type": "string",
          "avro.java.string": "String"
        },
        "null"
      ]
    },
    {
      "name": "user_agent",
      "type": [
        {
          "type": "string",
          "avro.java.string": "String"
        },
        "null"
      ]
    },
    {
      "name": "request_header",
      "type": [
        {
          "type": "string",
          "avro.java.string": "String"
        },
        "null"
      ]
    },
    {
      "name": "request_body",
      "type": [
        {
          "type": "string",
          "avro.java.string": "String"
        },
        "null"
      ]
    },
    {
      "name": "response_code",
      "type": [
        {
          "type": "string",
          "avro.java.string": "String"
        },
        "null"
      ]
    },
    {
      "name": "response_header",
      "type": [
        {
          "type": "string",
          "avro.java.string": "String"
        },
        "null"
      ]
    },
    {
      "name": "response_body",
      "type": [
        {
          "type": "string",
          "avro.java.string": "String"
        },
        "null"
      ]
    },
    {
      "name": "XFF",
      "type": [
        {
          "type": "string",
          "avro.java.string": "String"
        },
        "null"
      ]
    },
    {
      "name": "extra_info",
      "type": [
        {
          "type": "string",
          "avro.java.string": "String"
        },
        "null"
      ]
    },
    {
      "name": "s_country",
      "type": [
        {
          "type": "string",
          "avro.java.string": "String"
        },
        "null"
      ]
    },
    {
      "name": "s_province",
      "type": [
        {
          "type": "string",
          "avro.java.string": "String"
        },
        "null"
      ]
    },
    {
      "name": "s_city",
      "type": [
        {
          "type": "string",
          "avro.java.string": "String"
        },
        "null"
      ]
    },
    {
      "name": "s_operator",
      "type": [
        {
          "type": "string",
          "avro.java.string": "String"
        },
        "null"
      ]
    },
    {
      "name": "d_country",
      "type": [
        {
          "type": "string",
          "avro.java.string": "String"
        },
        "null"
      ]
    },
    {
      "name": "d_province",
      "type": [
        {
          "type": "string",
          "avro.java.string": "String"
        },
        "null"
      ]
    },
    {
      "name": "d_city",
      "type": [
        {
          "type": "string",
          "avro.java.string": "String"
        },
        "null"
      ]
    },
    {
      "name": "d_operator",
      "type": [
        {
          "type": "string",
          "avro.java.string": "String"
        },
        "null"
      ]
    },
    {
      "name": "payload",
      "type": [
        {
          "type": "string",
          "avro.java.string": "String"
        },
        "null"
      ]
    },
    {
      "name": "attack_result",
      "type": [
        "int",
        "null"
      ]
    },
    {
      "name": "attack_chain",
      "type": [
        "int",
        "null"
      ]
    },
    {
      "name": "description",
      "type": [
        {
          "type": "string",
          "avro.java.string": "String"
        },
        "null"
      ]
    },
    {
      "name": "solution",
      "type": [
        {
          "type": "string",
          "avro.java.string": "String"
        },
        "null"
      ]
    },
    {
      "name": "original_info",
      "type": [
        {
          "type": "string",
          "avro.java.string": "String"
        },
        "null"
      ]
    },
    {
      "name": "pcap_info",
      "type": [
        {
          "type": "string",
          "avro.java.string": "String"
        },
        "null"
      ]
    }
  ]
}

Flink code:

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[DanielCarter-stack]

@zoucx

Thanks for the report. Based on your Avro schema, all fields are union types (e.g., ["string", "null"]), but your SeaTunnel config defines them as non-nullable types (string, int, bigint). This mismatch likely causes the deserialization failure when Kafka messages contain null values.

Current implementation builds the Avro schema from your SeaTunnel config rather than reading the actual schema from Kafka messages. See:

• seatunnel-formats/seatunnel-format-avro/src/main/java/org/apache/seatunnel/format/avro/AvroToRowConverter.java:50-51
• seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceConfig.java:407-408

To resolve:

1. Confirm field nullability: Does your data actually contain null values in any fields?
2. Adjust schema definition: Check the docs for declaring nullable fields in SeaTunnel schema syntax (see docs/zh/connectors/formats/avro.md).
3. Provide full stack trace: The WARN log alone is insufficient—please share the complete exception stack trace.

Could you also share how Flink is configured to handle the same Avro schema successfully?

[moody1117]

Thank you for your reply.
1.Data may contain null values.
2.I modified the configuration based on the recommended approach in the schema documentation and the Avro format documentation, but the data still fails to parse correctly. When no schema is defined and data from this Avro-formatted topic is read into another Avro-formatted topic, the data becomes unparsable by the Flink program as well, indicating that the content has changed.
3.Currently, only this WARN log exists. If you comment out format_error_handle_way = "skip", an ArrayIndexOutOfBoundsException error will occur.

env {
  parallelism = 1
  job.mode = "BATCH"
}
source {
  Kafka {
    topic = "t_event"
    bootstrap.servers = "xxx:9092"
    consumer.group = "test_avro_kafka"
    start_mode = "timestamp"
    start_mode.timestamp = "1769504419000"
    start_mode.end_timestamp = "1769504479000"
    format = "avro"
    format_error_handle_way = "skip"
    
    plugin_output = "kafka_table"
    kafka.config = {
      enable.auto.commit = "true"
      max.poll.records = "10000"
      
    }
    schema {
      table = "simple_example"
      columns = [
        { name = "dev_id",        type = "string",  nullable = true },
        { name = "dev_ip",        type = "string",  nullable = true },
        { name = "event_id",      type = "string",  nullable = true },
        { name = "timestamp",     type = "bigint",  nullable = true },
        { name = "s_ipv4",        type = "string",  nullable = true },
        { name = "s_ipv6",        type = "string",  nullable = true },
        { name = "s_port",        type = "int",     nullable = true },
        { name = "d_ipv4",        type = "string",  nullable = true },
        { name = "d_ipv6",        type = "string",  nullable = true },
        { name = "d_port",        type = "int",     nullable = true },
        { name = "sid",           type = "int",     nullable = true },
        { name = "severity",      type = "int",     nullable = true },
        { name = "alarm_name",    type = "string",  nullable = true },
        { name = "alarm_type",    type = "string",  nullable = true },
        { name = "classification",type = "string",  nullable = true },
        { name = "protocol",      type = "string",  nullable = true },
        { name = "domain_name",   type = "string",  nullable = true },
        { name = "request_method",type = "string",  nullable = true },
        { name = "protocol_version", type = "string", nullable = true },
        { name = "request_uri",   type = "string",  nullable = true },
        { name = "request_url",   type = "string",  nullable = true },
        { name = "referer",       type = "string",  nullable = true },
        { name = "user_agent",    type = "string",  nullable = true },
        { name = "request_header",type = "string",  nullable = true },
        { name = "request_body",  type = "string",  nullable = true },
        { name = "response_code", type = "string",  nullable = true },
        { name = "response_header",type = "string", nullable = true },
        { name = "response_body", type = "string",  nullable = true },
        { name = "XFF",           type = "string",  nullable = true },
        { name = "extra_info",    type = "string",  nullable = true },
        { name = "s_country",     type = "string",  nullable = true },
        { name = "s_province",    type = "string",  nullable = true },
        { name = "s_city",        type = "string",  nullable = true },
        { name = "s_operator",    type = "string",  nullable = true },
        { name = "d_country",     type = "string",  nullable = true },
        { name = "d_province",    type = "string",  nullable = true },
        { name = "d_city",        type = "string",  nullable = true },
        { name = "d_operator",    type = "string",  nullable = true },
        { name = "payload",       type = "string",  nullable = true },
        { name = "attack_result", type = "int",     nullable = true },
        { name = "attack_chain",  type = "int",     nullable = true },
        { name = "description",   type = "string",  nullable = true },
        { name = "solution",      type = "string",  nullable = true },
        { name = "original_info", type = "string",  nullable = true },
        { name = "pcap_info",     type = "string",  nullable = true }
      ]
    }
  }
}
sink {
  Console {
    plugin_input = "kafka_table"
  }
  LocalFile {
    plugin_input = "kafka_table"
    path = "/data/seatunnel/file"
    file_format_type = "json"
    custom_filename = true
    filename_time_format = "yyyy.MM.dd"
    is_enable_transaction = true
    file_name_expression = "${transactionId}_${now}"
   }
}

2026-01-27 18:01:20,957 INFO  [o.a.s.e.s.d.p.PhysicalVertex  ] [hz.main.seaTunnel.task.thread-5] - Job (1068470671115288577), Pipeline: [(1/1)], task: [pipeline-1 [Source[0]-Kafka]-SourceTask (1/1)], taskGroupLocation: [TaskGroupLocation{jobId=1068470671115288577, pipelineId=1, taskGroupId=3}] state process is stopped
2026-01-27 18:01:20,957 ERROR [o.a.s.e.s.d.p.PhysicalVertex  ] [hz.main.seaTunnel.task.thread-5] - Job (1068470671115288577), Pipeline: [(1/1)], task: [pipeline-1 [Source[0]-Kafka]-SourceTask (1/1)], taskGroupLocation: [TaskGroupLocation{jobId=1068470671115288577, pipelineId=1, taskGroupId=3}] end with state FAILED and Exception: java.lang.ArrayIndexOutOfBoundsException: 12
        at org.apache.avro.io.parsing.Symbol$Alternative.getSymbol(Symbol.java:460)
        at org.apache.avro.io.ResolvingDecoder.readIndex(ResolvingDecoder.java:283)
        at org.apache.avro.generic.GenericDatumReader.readWithoutConversion(GenericDatumReader.java:188)
        at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:161)
        at org.apache.avro.generic.GenericDatumReader.readField(GenericDatumReader.java:260)
        at org.apache.avro.generic.GenericDatumReader.readRecord(GenericDatumReader.java:248)
        at org.apache.avro.generic.GenericDatumReader.readWithoutConversion(GenericDatumReader.java:180)
        at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:161)
        at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:154)
        at org.apache.seatunnel.format.avro.AvroDeserializationSchema.deserialize(AvroDeserializationSchema.java:51)
        at org.apache.seatunnel.format.avro.AvroDeserializationSchema.deserialize(AvroDeserializationSchema.java:34)
        at org.apache.seatunnel.api.serialization.DeserializationSchema.deserialize(DeserializationSchema.java:38)
        at org.apache.seatunnel.connectors.seatunnel.kafka.source.KafkaRecordEmitter.emitRecord(KafkaRecordEmitter.java:72)
        at org.apache.seatunnel.connectors.seatunnel.kafka.source.KafkaRecordEmitter.emitRecord(KafkaRecordEmitter.java:37)
        at org.apache.seatunnel.connectors.seatunnel.common.source.reader.SourceReaderBase.pollNext(SourceReaderBase.java:110)
        at org.apache.seatunnel.engine.server.task.flow.SourceFlowLifeCycle.collect(SourceFlowLifeCycle.java:159)
        at org.apache.seatunnel.engine.server.task.SourceSeaTunnelTask.collect(SourceSeaTunnelTask.java:127)
        at org.apache.seatunnel.engine.server.task.SeaTunnelTask.stateProcess(SeaTunnelTask.java:165)
        at org.apache.seatunnel.engine.server.task.SourceSeaTunnelTask.call(SourceSeaTunnelTask.java:132)
        at org.apache.seatunnel.engine.server.TaskExecutionService$BlockingWorker.run(TaskExecutionService.java:679)
        at org.apache.seatunnel.engine.server.TaskExecutionService$NamedTaskWrapper.run(TaskExecutionService.java:1008)
        at org.apache.seatunnel.api.tracing.MDCRunnable.run(MDCRunnable.java:43)
        at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
        at java.util.concurrent.FutureTask.run(FutureTask.java:266)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:750)

Flink tasks：

1. compile the schema from event.avsc files into Java classes.
2. then parse the data using the AvroDeserializationSchema serializer.