class AvroItemReader<T> extends AbstractItemCountingItemStreamItemReader<T> {

	private boolean embeddedSchema = true;

	private InputStreamReader<T> inputStreamReader;

	private DataFileStream<T> dataFileReader;

	private final InputStream inputStream;

	private final DatumReader<T> datumReader;

	/**
	 * @param resource the {@link Resource} containing objects serialized with Avro.
	 * @param clazz the data type to be deserialized.
	 */
	public AvroItemReader(Resource resource, Class<T> clazz) {
		setName(ClassUtils.getShortName(AvroItemReader.class));

		Assert.notNull(resource, "'resource' is required.");
		Assert.notNull(clazz, "'class' is required.");

		try {
			this.inputStream = resource.getInputStream();
			this.datumReader = datumReaderForClass(clazz);
		}
		catch (IOException e) {
			throw new IllegalArgumentException(e.getMessage(), e);
		}
	}

	/**
	 * @param data the {@link Resource} containing the data to be read.
	 * @param schema the {@link Resource} containing the Avro schema.
	 */
	public AvroItemReader(Resource data, Resource schema) {
		setName(ClassUtils.getShortName(AvroItemReader.class));

		Assert.notNull(data, "'data' is required.");
		Assert.state(data.exists(), "'data' " + data.getFilename() + " does not exist.");
		Assert.notNull(schema, "'schema' is required");
		Assert.state(schema.exists(), "'schema' " + schema.getFilename() + " does not exist.");
		try {
			this.inputStream = data.getInputStream();
			Schema avroSchema = new Schema.Parser().parse(schema.getInputStream());
			this.datumReader = new GenericDatumReader<>(avroSchema);
		}
		catch (IOException e) {
			throw new IllegalArgumentException(e.getMessage(), e);
		}
	}

	/**
	 * Disable or enable reading an embedded Avro schema. True by default.
	 * @param embeddedSchema set to false to if the input does not embed an Avro schema.
	 */
	public void setEmbeddedSchema(boolean embeddedSchema) {
		this.embeddedSchema = embeddedSchema;
	}

	@Nullable
	@Override
	protected T doRead() throws Exception {
		if (this.inputStreamReader != null) {
			return this.inputStreamReader.read();
		}
		return this.dataFileReader.hasNext() ? this.dataFileReader.next() : null;
	}

	@Override
	protected void doOpen() throws Exception {
		initializeReader();
	}

	@Override
	protected void doClose() throws Exception {
		if (this.inputStreamReader != null) {
			this.inputStreamReader.close();
			return;
		}
		this.dataFileReader.close();
	}

	private void initializeReader() throws IOException {<FILL_FUNCTION_BODY>}

	private InputStreamReader<T> createInputStreamReader(InputStream inputStream, DatumReader<T> datumReader) {
		return new InputStreamReader<>(inputStream, datumReader);
	}

	private static <T> DatumReader<T> datumReaderForClass(Class<T> clazz) {
		if (SpecificRecordBase.class.isAssignableFrom(clazz)) {
			return new SpecificDatumReader<>(clazz);
		}
		if (GenericRecord.class.isAssignableFrom(clazz)) {
			return new GenericDatumReader<>();
		}
		return new ReflectDatumReader<>(clazz);
	}

	private static class InputStreamReader<T> {

		private final DatumReader<T> datumReader;

		private final BinaryDecoder binaryDecoder;

		private final InputStream inputStream;

		private InputStreamReader(InputStream inputStream, DatumReader<T> datumReader) {
			this.inputStream = inputStream;
			this.datumReader = datumReader;
			this.binaryDecoder = DecoderFactory.get().binaryDecoder(inputStream, null);
		}

		private T read() throws Exception {
			if (!this.binaryDecoder.isEnd()) {
				return this.datumReader.read(null, this.binaryDecoder);
			}
			return null;
		}

		private void close() {
			try {
				this.inputStream.close();
			}
			catch (IOException e) {
				throw new ItemStreamException(e.getMessage(), e);
			}
		}

	}

}

		if (this.embeddedSchema) {
			this.dataFileReader = new DataFileStream<>(this.inputStream, this.datumReader);
		}
		else {
			this.inputStreamReader = createInputStreamReader(this.inputStream, this.datumReader);
		}

	
<methods>public non-sealed void <init>() ,public void close() throws org.springframework.batch.item.ItemStreamException,public int getCurrentItemCount() ,public boolean isSaveState() ,public void open(org.springframework.batch.item.ExecutionContext) throws org.springframework.batch.item.ItemStreamException,public T read() throws java.lang.Exception,public void setCurrentItemCount(int) ,public void setMaxItemCount(int) ,public void setSaveState(boolean) ,public void update(org.springframework.batch.item.ExecutionContext) throws org.springframework.batch.item.ItemStreamException<variables>private static final java.lang.String READ_COUNT,private static final java.lang.String READ_COUNT_MAX,private int currentItemCount,private int maxItemCount,private boolean saveState