Hi team, I implemented a flink job to read CustomerSession avro files from s3 path, and without modifying them, directly output to a CustomerSession kafka topic, which has been registered the CustomerSession schema. I implemented the reading from s3 part like this using

AvroInputFormat

:

String s3Path =
        "s3://...";
    AvroInputFormat<CustomerSession> avroInputFormat =
        new AvroInputFormat<>(new Path(s3Path), CustomerSession.class);
    TypeInformation<CustomerSession> typeInfo = TypeInformation.of(CustomerSession.class);
    DataStream<CustomerSession> datastream = env.createInput(avroInputFormat, typeInfo);

But encountered this error in the job:

org.apache.avro.AvroTypeException: Found KafkaRecord, expecting object.analytical.CustomerSession, missing required field channel

Is it related to the way implementing reading from s3 path part using

AvroInputFormat

? Shall I switch to

FileSource

instead? Thank you!