class AvroItemWriter<T> extends AbstractItemStreamItemWriter<T> {

	private DataFileWriter<T> dataFileWriter;

	private OutputStreamWriter<T> outputStreamWriter;

	private final WritableResource resource;

	private final Resource schemaResource;

	private final Class<T> clazz;

	private boolean embedSchema = true;

	/**
	 * @param resource a {@link WritableResource} to which the objects will be serialized.
	 * @param schema a {@link Resource} containing the Avro schema.
	 * @param clazz the data type to be serialized.
	 */
	public AvroItemWriter(WritableResource resource, Resource schema, Class<T> clazz) {
		this.schemaResource = schema;
		this.resource = resource;
		this.clazz = clazz;
	}

	/**
	 * This constructor will create an ItemWriter that does not embedded Avro schema.
	 * @param resource a {@link WritableResource} to which the objects will be serialized.
	 * @param clazz the data type to be serialized.
	 */
	public AvroItemWriter(WritableResource resource, Class<T> clazz) {
		this(resource, null, clazz);
		embedSchema = false;
	}

	@Override
	public void write(Chunk<? extends T> items) throws Exception {
		items.forEach(item -> {
			try {
				if (this.dataFileWriter != null) {
					this.dataFileWriter.append(item);
				}
				else {
					this.outputStreamWriter.write(item);
				}
			}
			catch (Exception e) {
				throw new ItemStreamException(e.getMessage(), e);
			}
		});
	}

	/**
	 * @see org.springframework.batch.item.ItemStream#open(ExecutionContext)
	 */
	@Override
	public void open(ExecutionContext executionContext) {<FILL_FUNCTION_BODY>}

	@Override
	public void close() {
		try {
			if (this.dataFileWriter != null) {
				this.dataFileWriter.close();
			}
			else {
				this.outputStreamWriter.close();
			}
		}
		catch (IOException e) {
			throw new ItemStreamException(e.getMessage(), e);
		}
	}

	private void initializeWriter() throws IOException {
		Assert.notNull(this.resource, "'resource' is required.");
		Assert.notNull(this.clazz, "'class' is required.");

		if (this.embedSchema) {
			Assert.notNull(this.schemaResource, "'schema' is required.");
			Assert.state(this.schemaResource.exists(),
					"'schema' " + this.schemaResource.getFilename() + " does not exist.");
			Schema schema;
			try {
				schema = new Schema.Parser().parse(this.schemaResource.getInputStream());
			}
			catch (IOException e) {
				throw new IllegalArgumentException(e.getMessage(), e);
			}
			this.dataFileWriter = new DataFileWriter<>(datumWriterForClass(this.clazz));
			this.dataFileWriter.create(schema, this.resource.getOutputStream());
		}
		else {
			this.outputStreamWriter = createOutputStreamWriter(this.resource.getOutputStream(),
					datumWriterForClass(this.clazz));
		}

	}

	private static <T> DatumWriter<T> datumWriterForClass(Class<T> clazz) {
		if (SpecificRecordBase.class.isAssignableFrom(clazz)) {
			return new SpecificDatumWriter<>(clazz);
		}
		if (GenericRecord.class.isAssignableFrom(clazz)) {
			return new GenericDatumWriter<>();
		}
		return new ReflectDatumWriter<>(clazz);
	}

	private AvroItemWriter.OutputStreamWriter<T> createOutputStreamWriter(OutputStream outputStream,
			DatumWriter<T> datumWriter) {
		return new AvroItemWriter.OutputStreamWriter<>(outputStream, datumWriter);
	}

	private static class OutputStreamWriter<T> {

		private final DatumWriter<T> datumWriter;

		private final BinaryEncoder binaryEncoder;

		private final OutputStream outputStream;

		private OutputStreamWriter(OutputStream outputStream, DatumWriter<T> datumWriter) {
			this.outputStream = outputStream;
			this.datumWriter = datumWriter;
			this.binaryEncoder = EncoderFactory.get().binaryEncoder(outputStream, null);
		}

		private void write(T datum) throws Exception {
			this.datumWriter.write(datum, this.binaryEncoder);
			this.binaryEncoder.flush();
		}

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

	}

}

		super.open(executionContext);
		try {
			initializeWriter();
		}
		catch (IOException e) {
			throw new ItemStreamException(e.getMessage(), e);
		}
	
<methods>public non-sealed void <init>() <variables>