Apache Flink

Hi Flink team, if I want to implement a flink Job to read avro files from S3 path, how can I do that? Which class shall I use to do that?

This should help you <https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/formats/avro/|https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/formats/avro/>

Hi <@U03PVF21Q9E> , thanks for the link. I implemented like this using `AvroInputFormat` :
```String s3Path =
        "s3://...";
    AvroInputFormat&lt;CustomerSession&gt; avroInputFormat =
        new AvroInputFormat&lt;&gt;(new Path(s3Path), CustomerSession.class);
    TypeInformation&lt;CustomerSession&gt; typeInfo = TypeInformation.of(CustomerSession.class);
    DataStream&lt;CustomerSession&gt; datastream = env.createInput(avroInputFormat, typeInfo); ```

But encountered this error in the job:

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

But I am currently reading CustomerSession avro files from our s3 bucket, and without any modifications, output to a customersession kafka topic, which has already been registered to the schema of CustomerSession.