From bd53168ddb613aa5a5f8bc550f20a828ce3cf17d Mon Sep 17 00:00:00 2001 From: David Turanski Date: Mon, 15 Jul 2019 16:01:17 -0400 Subject: [PATCH] Implement AvroItemReader and AvroItemWriter Resolves BATCH-2833 --- build.gradle | 3 + .../asciidoc/readersAndWriters.adoc | 16 + .../batch/item/avro/AvroItemReader.java | 177 ++++++ .../batch/item/avro/AvroItemWriter.java | 200 +++++++ .../avro/builder/AvroItemReaderBuilder.java | 203 +++++++ .../avro/builder/AvroItemWriterBuilder.java | 118 ++++ .../batch/item/avro/AvroItemReaderTests.java | 81 +++ .../batch/item/avro/AvroItemWriterTests.java | 99 ++++ .../builder/AvroItemReaderBuilderTests.java | 85 +++ .../builder/AvroItemWriterBuilderTests.java | 124 +++++ .../item/avro/example/AvroTestUtils.java | 81 +++ .../batch/item/avro/example/User.java | 518 ++++++++++++++++++ .../support/AvroItemReaderTestSupport.java | 47 ++ .../support/AvroItemWriterTestSupport.java | 137 +++++ .../item/avro/support/AvroTestFixtures.java | 185 +++++++ .../avro/plain-old-user-data-no-schema.avro | 3 + .../batch/item/avro/user-data-no-schema.avro | Bin 0 -> 53 bytes .../batch/item/avro/user-data.avro | Bin 0 -> 339 bytes .../batch/item/avro/user-schema.json | 9 + 19 files changed, 2086 insertions(+) create mode 100755 spring-batch-infrastructure/src/main/java/org/springframework/batch/item/avro/AvroItemReader.java create mode 100644 spring-batch-infrastructure/src/main/java/org/springframework/batch/item/avro/AvroItemWriter.java create mode 100644 spring-batch-infrastructure/src/main/java/org/springframework/batch/item/avro/builder/AvroItemReaderBuilder.java create mode 100644 spring-batch-infrastructure/src/main/java/org/springframework/batch/item/avro/builder/AvroItemWriterBuilder.java create mode 100644 spring-batch-infrastructure/src/test/java/org/springframework/batch/item/avro/AvroItemReaderTests.java create mode 100644 spring-batch-infrastructure/src/test/java/org/springframework/batch/item/avro/AvroItemWriterTests.java create mode 100644 spring-batch-infrastructure/src/test/java/org/springframework/batch/item/avro/builder/AvroItemReaderBuilderTests.java create mode 100644 spring-batch-infrastructure/src/test/java/org/springframework/batch/item/avro/builder/AvroItemWriterBuilderTests.java create mode 100644 spring-batch-infrastructure/src/test/java/org/springframework/batch/item/avro/example/AvroTestUtils.java create mode 100644 spring-batch-infrastructure/src/test/java/org/springframework/batch/item/avro/example/User.java create mode 100644 spring-batch-infrastructure/src/test/java/org/springframework/batch/item/avro/support/AvroItemReaderTestSupport.java create mode 100644 spring-batch-infrastructure/src/test/java/org/springframework/batch/item/avro/support/AvroItemWriterTestSupport.java create mode 100644 spring-batch-infrastructure/src/test/java/org/springframework/batch/item/avro/support/AvroTestFixtures.java create mode 100644 spring-batch-infrastructure/src/test/resources/org/springframework/batch/item/avro/plain-old-user-data-no-schema.avro create mode 100644 spring-batch-infrastructure/src/test/resources/org/springframework/batch/item/avro/user-data-no-schema.avro create mode 100644 spring-batch-infrastructure/src/test/resources/org/springframework/batch/item/avro/user-data.avro create mode 100644 spring-batch-infrastructure/src/test/resources/org/springframework/batch/item/avro/user-schema.json diff --git a/build.gradle b/build.gradle index cf9a0843d..af6a9d972 100644 --- a/build.gradle +++ b/build.gradle @@ -60,6 +60,7 @@ allprojects { springLdapVersion = '2.3.2.RELEASE' activemqVersion = '5.15.9' + apacheAvroVersion ='1.9.0' aspectjVersion = '1.9.3' commonsDdbcpVersion = '2.6.0' commonsIoVersion = '2.6' @@ -385,6 +386,8 @@ project('spring-batch-infrastructure') { optional "org.springframework.ldap:spring-ldap-core-tiger:$springLdapVersion" optional "org.springframework.ldap:spring-ldap-ldif-core:$springLdapVersion" optional "org.springframework.kafka:spring-kafka:$springKafkaVersion" + optional "org.apache.avro:avro:$apacheAvroVersion" + // JSR-305 only used for non-required meta-annotations compileOnly("com.google.code.findbugs:jsr305:3.0.2") testCompileOnly("com.google.code.findbugs:jsr305:3.0.2") diff --git a/spring-batch-docs/asciidoc/readersAndWriters.adoc b/spring-batch-docs/asciidoc/readersAndWriters.adoc index 05c6cd3fd..f791663ee 100644 --- a/spring-batch-docs/asciidoc/readersAndWriters.adoc +++ b/spring-batch-docs/asciidoc/readersAndWriters.adoc @@ -3201,6 +3201,7 @@ Spring Batch offers the following specialized readers: * <> * <> +* <> [[ldifReader]] ===== `LdifReader` @@ -3216,11 +3217,19 @@ The `MappingLdifReader` reads LDIF (LDAP Data Interchange Format) records from a Each read returns a POJO. Spring Batch provides a `MappingLdifReaderBuilder` to construct an instance of the `MappingLdifReader`. +[[avroItemReader]] +===== `AvroItemReader` +The `AvroItemReader` reads serialized Avro data from a Resource. +Each read returns an instance of the type specified by a Java class or Avro Schema. +The reader may be optionnally configured for input that embeds an Avro schema or not. +Spring Batch provides an `AvroItemReaderBuilder` to construct an instance of the `AvroItemReader`. + [[specializedWriters]] ==== Specialized Writers Spring Batch offers the following specialized writers: * <> +* <> [[simpleMailMessageItemWriter]] ===== `SimpleMailMessageItemWriter` @@ -3229,6 +3238,13 @@ delegates the actual sending of messages to an instance of `MailSender`. Spring provides a `SimpleMailMessageItemWriterBuilder` to construct an instance of the `SimpleMailMessageItemWriter`. +[[avroItemWriter]] +===== `AvroItemWriter` +The `AvroItemWrite` serializes Java objects to a WriteableResource according to the given type or Schema. +The writer may be optionally configured to embed an Avro schema in the output or not. +Spring Batch provides an `AvroItemWriterBuilder` to construct an instance of the `AvroItemWriter`. + + [[specializedProcessors]] ==== Specialized Processors Spring Batch offers the following specialized processors: diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/avro/AvroItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/avro/AvroItemReader.java new file mode 100755 index 000000000..4b0fe0b9d --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/avro/AvroItemReader.java @@ -0,0 +1,177 @@ +/* + * Copyright 2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.batch.item.avro; + +import java.io.IOException; +import java.io.InputStream; + +import org.apache.avro.Schema; +import org.apache.avro.file.DataFileStream; +import org.apache.avro.generic.GenericDatumReader; +import org.apache.avro.generic.GenericRecord; +import org.apache.avro.io.BinaryDecoder; +import org.apache.avro.io.DatumReader; +import org.apache.avro.io.DecoderFactory; +import org.apache.avro.reflect.ReflectDatumReader; +import org.apache.avro.specific.SpecificDatumReader; +import org.apache.avro.specific.SpecificRecordBase; + +import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.ItemStreamException; +import org.springframework.batch.item.support.AbstractItemCountingItemStreamItemReader; +import org.springframework.core.io.Resource; +import org.springframework.util.Assert; + +/** + * An {@link ItemReader} that deserializes data from a {@link Resource} containing serialized Avro objects. + * + * @author David Turanski + * @since 4.2 + */ +public class AvroItemReader extends AbstractItemCountingItemStreamItemReader { + + private boolean embeddedSchema = true; + + private InputStreamReader inputStreamReader; + + private DataFileStream dataFileReader; + + private InputStream inputStream; + + private DatumReader datumReader; + + /** + * + * @param resource the {@link Resource} containing objects serialized with Avro. + * @param clazz the data type to be deserialized. + */ + public AvroItemReader(Resource resource, Class clazz) { + 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) { + 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; + } + + + @Override + protected T doRead() throws Exception { + if (inputStreamReader != null) { + return inputStreamReader.read(); + } + return dataFileReader.hasNext()? dataFileReader.next(): null; + } + + @Override + protected void doOpen() throws Exception { + initializeReader(); + } + + @Override + protected void doClose() throws Exception { + if (this.inputStreamReader != null) { + inputStreamReader.close(); + return; + } + dataFileReader.close(); + } + + private void initializeReader() throws IOException { + if (this.embeddedSchema) { + this.dataFileReader = new DataFileStream<>(this.inputStream, this.datumReader); + } else { + this.inputStreamReader = createInputStreamReader(this.inputStream, this.datumReader); + } + + } + + private InputStreamReader createInputStreamReader(InputStream inputStream, DatumReader datumReader) { + return new InputStreamReader<>(inputStream, datumReader); + } + + private static DatumReader datumReaderForClass(Class 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 { + private final DatumReader datumReader; + + private final BinaryDecoder binaryDecoder; + + private final InputStream inputStream; + + private InputStreamReader(InputStream inputStream, DatumReader datumReader) { + this.inputStream = inputStream; + this.datumReader = datumReader; + this.binaryDecoder = DecoderFactory.get().binaryDecoder(inputStream, null); + } + + private T read() throws Exception { + if (!binaryDecoder.isEnd()) { + return datumReader.read(null, binaryDecoder); + } + return null; + } + + private void close() { + try { + this.inputStream.close(); + } catch (IOException e) { + throw new ItemStreamException(e.getMessage(), e); + } + } + } +} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/avro/AvroItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/avro/AvroItemWriter.java new file mode 100644 index 000000000..5fef17ce9 --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/avro/AvroItemWriter.java @@ -0,0 +1,200 @@ +/* + * Copyright 2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.batch.item.avro; + +import java.io.IOException; +import java.io.OutputStream; +import java.util.List; + +import org.apache.avro.Schema; +import org.apache.avro.file.DataFileWriter; +import org.apache.avro.generic.GenericDatumWriter; +import org.apache.avro.generic.GenericRecord; +import org.apache.avro.io.BinaryEncoder; +import org.apache.avro.io.DatumWriter; +import org.apache.avro.io.EncoderFactory; +import org.apache.avro.reflect.ReflectDatumWriter; +import org.apache.avro.specific.SpecificDatumWriter; +import org.apache.avro.specific.SpecificRecordBase; + +import org.springframework.batch.item.ExecutionContext; +import org.springframework.batch.item.ItemStreamException; +import org.springframework.batch.item.ItemWriter; +import org.springframework.batch.item.support.AbstractItemStreamItemWriter; +import org.springframework.core.io.Resource; +import org.springframework.core.io.WritableResource; +import org.springframework.util.Assert; + +/** + * An {@link ItemWriter} that serializes data to an {@link WritableResource} using Avro. + * + * This does not support restart on failure. + * + * @since 4.2 + * @author David Turanski + */ +public class AvroItemWriter extends AbstractItemStreamItemWriter { + + private DataFileWriter dataFileWriter; + + private OutputStreamWriter outputStreamWriter; + + private WritableResource resource; + + private Schema schema; + + private Resource schemaResource; + + private Class 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 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 clazz) { + this(resource, null, clazz); + embedSchema = false; + } + + @Override + public void write(List 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) { + try { + initializeWriter(); + } catch (IOException e) { + throw new ItemStreamException(e.getMessage(), e); + } + } + + @Override + public void close() { + try { + if (this.dataFileWriter != null) { + dataFileWriter.close(); + } + else { + this.outputStreamWriter.close(); + } + } + catch (IOException e) { + throw new ItemStreamException(e.getMessage(), e); + } + } + + private void initializeWriter() throws IOException { + Assert.notNull(resource, "'resource' is required."); + Assert.notNull(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."); + try { + this.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(this.schema, this.resource.getOutputStream()); + } else { + this.outputStreamWriter = createOutputStreamWriter(resource.getOutputStream(), + datumWriterForClass(this.clazz)); + } + + } + + private static DatumWriter datumWriterForClass(Class clazz) { + if (GenericRecord.class.isAssignableFrom(clazz)) { + return new GenericDatumWriter<>(); + + } + if (SpecificRecordBase.class.isAssignableFrom(clazz)){ + return new SpecificDatumWriter<>(clazz); + } + + return new ReflectDatumWriter<>(clazz); + } + + private AvroItemWriter.OutputStreamWriter createOutputStreamWriter(OutputStream outputStream, + DatumWriter datumWriter) { + return new AvroItemWriter.OutputStreamWriter<>(outputStream, datumWriter); + } + + private static class OutputStreamWriter { + private final DatumWriter datumWriter; + + private final BinaryEncoder binaryEncoder; + + private final OutputStream outputStream; + + private OutputStreamWriter(OutputStream outputStream, DatumWriter datumWriter) { + this.outputStream = outputStream; + this.datumWriter = datumWriter; + this.binaryEncoder = EncoderFactory.get().binaryEncoder(outputStream, null); + } + + private void write(T datum) throws Exception { + datumWriter.write(datum, binaryEncoder); + binaryEncoder.flush(); + } + + private void close() { + try { + this.outputStream.close(); + } + catch (IOException e) { + throw new ItemStreamException(e.getMessage(), e); + } + } + } +} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/avro/builder/AvroItemReaderBuilder.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/avro/builder/AvroItemReaderBuilder.java new file mode 100644 index 000000000..449357d69 --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/avro/builder/AvroItemReaderBuilder.java @@ -0,0 +1,203 @@ +/* + * Copyright 2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.batch.item.avro.builder; + +import org.apache.avro.Schema; + +import org.springframework.batch.item.avro.AvroItemReader; +import org.springframework.core.io.ByteArrayResource; +import org.springframework.core.io.Resource; +import org.springframework.util.Assert; +import org.springframework.util.StringUtils; + +/** + * A builder implementation for the {@link AvroItemReader} + * + * @author David Turanski + * @since 4.2 + */ +public class AvroItemReaderBuilder { + + private boolean saveState = true; + + private String name = AvroItemReader.class.getSimpleName(); + + private int maxItemCount = Integer.MAX_VALUE; + + private int currentItemCount; + + private Resource schema; + + private Resource resource; + + private Class type; + + private boolean embeddedSchema =true; + + + /** + * Configure a {@link Resource} containing Avro serialized objects. + * @param resource an existing Resource. + * @return The current instance of the builder. + */ + public AvroItemReaderBuilder resource(Resource resource) { + Assert.notNull(resource, "A 'resource' is required."); + Assert.state(resource.exists(), "Resource " + resource.getFilename() + " does not exist."); + this.resource = resource; + return this; + } + + + /** + * Configure an Avro {@link Schema} from a {@link Resource}. + * @param schema an existing schema Resource. + * @return The current instance of the builder. + */ + public AvroItemReaderBuilder schema(Resource schema) { + Assert.notNull(schema, "A 'schema' Resource is required."); + Assert.state(schema.exists(), "Resource " + schema.getFilename() + " does not exist."); + this.schema = schema; + return this; + } + + /** + * Configure an Avro {@link Schema} from a String. + * @param schemaString the schema String. + * @return The current instance of the builder. + */ + public AvroItemReaderBuilder schema(String schemaString) { + Assert.hasText(schemaString, "A 'schema' is required."); + this.schema = new ByteArrayResource(schemaString.getBytes()); + return this; + } + + /** + * Configure a type to be deserialized. + * @param type the class to be deserialized. + * @return The current instance of the builder. + */ + public AvroItemReaderBuilder type(Class type) { + Assert.notNull(type, "A 'type' is required."); + this.type = type; + return this; + } + + /** + * Disable or enable reading an embedded Avro schema. True by default. + * @param embeddedSchema set to false to if the input does not contain an Avro schema. + */ + public AvroItemReaderBuilder embeddedSchema(boolean embeddedSchema) { + this.embeddedSchema = embeddedSchema; + return this; + } + + /** + * Configure if the state of the {@link org.springframework.batch.item.ItemStreamSupport} + * should be persisted within the {@link org.springframework.batch.item.ExecutionContext} + * for restart purposes. + * + * @param saveState defaults to true + * @return The current instance of the builder. + */ + public AvroItemReaderBuilder saveState(boolean saveState) { + this.saveState = saveState; + return this; + } + + /** + * The name used to calculate the key within the + * {@link org.springframework.batch.item.ExecutionContext}. Required if + * {@link #saveState(boolean)} is set to true. + * + * @param name name of the reader instance + * @return The current instance of the builder. + * @see org.springframework.batch.item.ItemStreamSupport#setName(String) + */ + public AvroItemReaderBuilder name(String name) { + this.name = name; + return this; + } + + /** + * Configure the max number of items to be read. + * + * @param maxItemCount the max items to be read + * @return The current instance of the builder. + * @see org.springframework.batch.item.support.AbstractItemCountingItemStreamItemReader#setMaxItemCount(int) + */ + public AvroItemReaderBuilder maxItemCount(int maxItemCount) { + this.maxItemCount = maxItemCount; + return this; + } + + /** + * Index for the current item. Used on restarts to indicate where to start from. + * + * @param currentItemCount current index + * @return this instance for method chaining + * @see org.springframework.batch.item.support.AbstractItemCountingItemStreamItemReader#setCurrentItemCount(int) + */ + public AvroItemReaderBuilder currentItemCount(int currentItemCount) { + this.currentItemCount = currentItemCount; + return this; + } + + + /** + * Build an instance of {@link AvroItemReader}. + * @return the instance; + */ + public AvroItemReader build() { + AvroItemReader avroItemReader; + + Assert.notNull(this.resource, "A 'resource' is required."); + + if (this.type != null) { + avroItemReader = buildForType(); + } + else { + avroItemReader = buildForSchema(); + } + + avroItemReader.setSaveState(this.saveState); + + if(this.saveState) { + Assert.state(StringUtils.hasText(this.name), + "A name is required when saveState is set to true."); + } + + avroItemReader.setName(this.name); + avroItemReader.setCurrentItemCount(this.currentItemCount); + avroItemReader.setMaxItemCount(this.maxItemCount); + avroItemReader.setEmbeddedSchema(this.embeddedSchema); + + return avroItemReader; + } + + + private AvroItemReader buildForType() { + Assert.isNull(this.schema, "You cannot specify a schema and 'type'."); + return new AvroItemReader<>(this.resource, this.type); + } + + private AvroItemReader buildForSchema() { + Assert.notNull(this.schema, "'schema' is required."); + return new AvroItemReader<>(resource, schema); + } + + +} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/avro/builder/AvroItemWriterBuilder.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/avro/builder/AvroItemWriterBuilder.java new file mode 100644 index 000000000..1c48cd287 --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/avro/builder/AvroItemWriterBuilder.java @@ -0,0 +1,118 @@ +/* + * Copyright 2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.batch.item.avro.builder; + +import org.springframework.batch.item.avro.AvroItemWriter; +import org.springframework.core.io.ByteArrayResource; +import org.springframework.core.io.Resource; +import org.springframework.core.io.WritableResource; +import org.springframework.util.Assert; + +/** + * @author David Turanski + * @since 4.2 + */ +public class AvroItemWriterBuilder { + private Class type; + + private WritableResource resource; + + private Resource schema; + + private String name = AvroItemWriter.class.getSimpleName(); + + /** + * + * @param resource the {@link WritableResource} used to write the serialized data. + * @return The current instance of the builder. + */ + public AvroItemWriterBuilder resource(WritableResource resource) { + Assert.notNull(resource, "A 'resource' is required."); + this.resource = resource; + return this; + } + + /** + * + * @param schema the Resource containing the schema JSON used to serialize the output. + * @return The current instance of the builder. + */ + public AvroItemWriterBuilder schema(Resource schema) { + Assert.notNull(schema, "A 'schema' is required."); + Assert.state(schema.exists(), "Resource " + schema.getFilename() + "does not exist."); + this.schema = schema; + return this; + } + + + /** + * + * @param schemaString the String containing the schema JSON used to serialize the output. + * @return The current instance of the builder. + */ + public AvroItemWriterBuilder schema(String schemaString) { + Assert.hasText(schemaString, "A 'schemaString' is required."); + this.schema = new ByteArrayResource(schemaString.getBytes()); + return this; + } + + + /** + * + * @param type the Class of objects to be serialized. + * @return The current instance of the builder. + */ + public AvroItemWriterBuilder type(Class type) { + Assert.notNull(type, "A 'type' is required."); + this.type = type; + return this; + } + + /** + * The name used to calculate the key within the + * {@link org.springframework.batch.item.ExecutionContext}. + * + * @param name name of the reader instance + * @return The current instance of the builder. + * @see org.springframework.batch.item.ItemStreamSupport#setName(String) + */ + public AvroItemWriterBuilder name(String name) { + Assert.hasText(name, "A 'name' is required."); + this.name = name; + return this; + } + + /** + * Build an instance of {@link AvroItemWriter}. + * + * @return the instance; + */ + public AvroItemWriter build() { + + Assert.notNull(this.resource, "A 'resource' is required."); + + Assert.notNull(this.type, "A 'type' is required."); + + + AvroItemWriter avroItemWriter = this.schema != null ? + new AvroItemWriter(this.resource, this.schema, this.type): + new AvroItemWriter<>(this.resource, this.type); + avroItemWriter.setName(name); + return avroItemWriter; + } + +} diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/avro/AvroItemReaderTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/avro/AvroItemReaderTests.java new file mode 100644 index 000000000..2166b3d74 --- /dev/null +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/avro/AvroItemReaderTests.java @@ -0,0 +1,81 @@ +/* + * Copyright 2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.batch.item.avro; + +import org.apache.avro.generic.GenericRecord; +import org.junit.Test; + +import org.springframework.batch.item.avro.example.User; +import org.springframework.batch.item.avro.support.AvroItemReaderTestSupport; +import org.springframework.core.io.ClassPathResource; + +/** + * @author David Turanski + */ +public class AvroItemReaderTests extends AvroItemReaderTestSupport { + + @Test + public void readGenericRecordsUsingResources() throws Exception { + + AvroItemReader itemReader = new AvroItemReader<>(dataResource, schemaResource); + itemReader.setName(itemReader.getClass().getSimpleName()); + itemReader.setEmbeddedSchema(false); + + verify(itemReader, genericAvroGeneratedUsers()); + } + + + @Test + public void readSpecificUsers() throws Exception { + + AvroItemReader itemReader = new AvroItemReader<>(dataResource, User.class); + itemReader.setEmbeddedSchema(false); + itemReader.setName(itemReader.getClass().getSimpleName()); + + verify(itemReader, avroGeneratedUsers()); + } + + @Test + public void readSpecificUsersWithEmbeddedSchema() throws Exception { + + AvroItemReader itemReader = new AvroItemReader<>(dataResourceWithSchema, User.class); + itemReader.setEmbeddedSchema(true); + itemReader.setName(itemReader.getClass().getSimpleName()); + + verify(itemReader, avroGeneratedUsers()); + } + + @Test + public void readPojosWithNoEmbeddedSchema() throws Exception { + + AvroItemReader itemReader = new AvroItemReader<>(plainOldUserDataResource, PlainOldUser.class); + itemReader.setEmbeddedSchema(false); + itemReader.setName(itemReader.getClass().getSimpleName()); + + verify(itemReader, plainOldUsers()); + } + + @Test(expected = IllegalStateException.class) + public void dataResourceDoesNotExist() { + new AvroItemReader(new ClassPathResource("doesnotexist"), schemaResource); + } + + @Test(expected = IllegalStateException.class) + public void schemaResourceDoesNotExist() { + new AvroItemReader(dataResource, new ClassPathResource("doesnotexist")); + } +} diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/avro/AvroItemWriterTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/avro/AvroItemWriterTests.java new file mode 100644 index 000000000..3eb934948 --- /dev/null +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/avro/AvroItemWriterTests.java @@ -0,0 +1,99 @@ +/* + * Copyright 2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.batch.item.avro; + +import java.io.ByteArrayOutputStream; + +import org.apache.avro.generic.GenericRecord; +import org.junit.Test; + +import org.springframework.batch.item.ExecutionContext; +import org.springframework.batch.item.avro.example.User; +import org.springframework.batch.item.avro.support.AvroItemWriterTestSupport; +import org.springframework.core.io.WritableResource; + +/** + * @author David Turanski + */ +public class AvroItemWriterTests extends AvroItemWriterTestSupport { + + private ByteArrayOutputStream outputStream = new ByteArrayOutputStream(2048); + + private WritableResource output = new OutputStreamResource(outputStream); + + @Test + public void itemWriterForAvroGeneratedClass() throws Exception { + + AvroItemWriter avroItemWriter = new AvroItemWriter<>(output,schemaResource, User.class); + avroItemWriter.open(new ExecutionContext()); + avroItemWriter.write(this.avroGeneratedUsers()); + avroItemWriter.close(); + + verifyRecordsWithEmbeddedHeader(outputStream.toByteArray(), this.avroGeneratedUsers(), User.class); + } + + @Test + public void itemWriterForGenericRecords() throws Exception { + + AvroItemWriter avroItemWriter = + new AvroItemWriter<>(output,plainOldUserSchemaResource,GenericRecord.class); + + avroItemWriter.open(new ExecutionContext()); + avroItemWriter.write(this.genericPlainOldUsers()); + avroItemWriter.close(); + + verifyRecordsWithEmbeddedHeader(outputStream.toByteArray(), this.genericPlainOldUsers(), GenericRecord.class); + + } + + @Test + public void itemWriterForPojos() throws Exception { + + AvroItemWriter avroItemWriter = new AvroItemWriter(output,plainOldUserSchemaResource, PlainOldUser.class); + avroItemWriter.open(new ExecutionContext()); + avroItemWriter.write(this.plainOldUsers()); + avroItemWriter.close(); + + verifyRecordsWithEmbeddedHeader(outputStream.toByteArray(), this.plainOldUsers(), PlainOldUser.class); + + } + + @Test + public void itemWriterWithNoEmbeddedHeaders() throws Exception { + + AvroItemWriter avroItemWriter = new AvroItemWriter(output, PlainOldUser.class); + avroItemWriter.open(new ExecutionContext()); + avroItemWriter.write(this.plainOldUsers()); + avroItemWriter.close(); + + verifyRecords(outputStream.toByteArray(), this.plainOldUsers(), PlainOldUser.class, false); + + } + + @Test(expected = IllegalArgumentException.class) + public void shouldFailWitNoOutput() { + + new AvroItemWriter(null, schemaResource, User.class).open(new ExecutionContext());; + + } + + @Test(expected = IllegalArgumentException.class) + public void shouldFailWitNoType() { + new AvroItemWriter(output, schemaResource, null).open(new ExecutionContext()); + + } +} diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/avro/builder/AvroItemReaderBuilderTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/avro/builder/AvroItemReaderBuilderTests.java new file mode 100644 index 000000000..034baa975 --- /dev/null +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/avro/builder/AvroItemReaderBuilderTests.java @@ -0,0 +1,85 @@ +/* + * Copyright 2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.batch.item.avro.builder; + +import org.apache.avro.generic.GenericRecord; +import org.junit.Test; + +import org.springframework.batch.item.avro.AvroItemReader; +import org.springframework.batch.item.avro.example.User; +import org.springframework.batch.item.avro.support.AvroItemReaderTestSupport; + +/** + * @author David Turanski + */ +public class AvroItemReaderBuilderTests extends AvroItemReaderTestSupport { + + @Test + public void itemReaderWithSchemaResource() throws Exception { + + AvroItemReader avroItemReader = new AvroItemReaderBuilder().resource(dataResource) + .embeddedSchema(false).schema(schemaResource).build(); + + verify(avroItemReader, genericAvroGeneratedUsers()); + } + + @Test + public void itemReaderWithGeneratedData() throws Exception { + AvroItemReader avroItemReader = new AvroItemReaderBuilder() + .resource(dataResourceWithSchema).schema(schemaResource).build(); + verify(avroItemReader, genericAvroGeneratedUsers()); + } + + @Test + public void itemReaderWithSchemaString() throws Exception { + AvroItemReader avroItemReader = new AvroItemReaderBuilder() + .schema(schemaString(schemaResource)).resource(dataResourceWithSchema).build(); + + verify(avroItemReader, genericAvroGeneratedUsers()); + } + + @Test + public void itemReaderWithEmbeddedHeader() throws Exception { + AvroItemReader avroItemReader = new AvroItemReaderBuilder().resource(dataResourceWithSchema) + .type(User.class).build(); + verify(avroItemReader, avroGeneratedUsers()); + } + + @Test + public void itemReaderForSpecificType() throws Exception { + AvroItemReader avroItemReader = new AvroItemReaderBuilder().type(User.class) + .resource(dataResourceWithSchema).build(); + verify(avroItemReader, avroGeneratedUsers()); + } + + @Test(expected = IllegalArgumentException.class) + public void itemReaderWithNoSchemaStringShouldFail() { + new AvroItemReaderBuilder().schema("").resource(dataResource).build(); + + } + + @Test(expected = IllegalArgumentException.class) + public void itemReaderWithPartialConfigurationShouldFail() { + new AvroItemReaderBuilder().resource(dataResource).build(); + } + + @Test(expected = IllegalArgumentException.class) + public void itemReaderWithNoInputsShouldFail() { + new AvroItemReaderBuilder().schema(schemaResource).build(); + } + +} diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/avro/builder/AvroItemWriterBuilderTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/avro/builder/AvroItemWriterBuilderTests.java new file mode 100644 index 000000000..f158a2b43 --- /dev/null +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/avro/builder/AvroItemWriterBuilderTests.java @@ -0,0 +1,124 @@ +/* + * Copyright 2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.batch.item.avro.builder; + +import java.io.ByteArrayOutputStream; + +import org.apache.avro.generic.GenericRecord; +import org.junit.Test; + +import org.springframework.batch.item.ExecutionContext; +import org.springframework.batch.item.avro.AvroItemWriter; +import org.springframework.batch.item.avro.example.User; +import org.springframework.batch.item.avro.support.AvroItemWriterTestSupport; +import org.springframework.core.io.WritableResource; + +/** + * @author David Turanski + */ +public class AvroItemWriterBuilderTests extends AvroItemWriterTestSupport { + + private ByteArrayOutputStream outputStream = new ByteArrayOutputStream(2048); + private WritableResource output = new OutputStreamResource(outputStream); + + @Test + public void itemWriterForAvroGeneratedClass() throws Exception { + + AvroItemWriter avroItemWriter = new AvroItemWriterBuilder() + .resource(output) + .schema(schemaResource) + .type(User.class) + .build(); + + avroItemWriter.open(new ExecutionContext()); + avroItemWriter.write(this.avroGeneratedUsers()); + avroItemWriter.close(); + + verifyRecordsWithEmbeddedHeader(outputStream.toByteArray(), this.avroGeneratedUsers(), User.class); + } + + + @Test + public void itemWriterForGenericRecords() throws Exception { + + AvroItemWriter avroItemWriter = new AvroItemWriterBuilder() + .type(GenericRecord.class) + .schema(plainOldUserSchemaResource) + .resource(output) + .build(); + + avroItemWriter.open(new ExecutionContext()); + avroItemWriter.write(this.genericPlainOldUsers()); + avroItemWriter.close(); + + verifyRecordsWithEmbeddedHeader(outputStream.toByteArray(), this.genericPlainOldUsers(), GenericRecord.class); + + } + + @Test + public void itemWriterForPojos() throws Exception { + + AvroItemWriter avroItemWriter = new AvroItemWriterBuilder() + .resource(output) + .schema(plainOldUserSchemaResource) + .type(PlainOldUser.class) + .build(); + + avroItemWriter.open(new ExecutionContext()); + avroItemWriter.write(this.plainOldUsers()); + avroItemWriter.close(); + + verifyRecordsWithEmbeddedHeader(outputStream.toByteArray(), this.plainOldUsers(), PlainOldUser.class); + + } + + @Test + public void itemWriterWithNoEmbeddedSchema() throws Exception { + + AvroItemWriter avroItemWriter = new AvroItemWriterBuilder() + .resource(output) + .type(PlainOldUser.class) + .build(); + avroItemWriter.open(new ExecutionContext()); + avroItemWriter.write(this.plainOldUsers()); + avroItemWriter.close(); + + verifyRecords(outputStream.toByteArray(), this.plainOldUsers(), PlainOldUser.class, false); + + } + + + @Test(expected = IllegalArgumentException.class) + public void shouldFailWitNoOutput() { + + new AvroItemWriterBuilder() + .type(GenericRecord.class) + .build(); + + } + + @Test(expected = IllegalArgumentException.class) + public void shouldFailWitNoType() { + + new AvroItemWriterBuilder() + .resource(output) + .schema(schemaResource) + .build(); + + } + +} diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/avro/example/AvroTestUtils.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/avro/example/AvroTestUtils.java new file mode 100644 index 000000000..c782f9a17 --- /dev/null +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/avro/example/AvroTestUtils.java @@ -0,0 +1,81 @@ +/* + * Copyright 2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.batch.item.avro.example; + +import java.io.File; +import java.io.FileOutputStream; +import org.apache.avro.Schema; +import org.apache.avro.file.DataFileWriter; +import org.apache.avro.io.DatumWriter; +import org.apache.avro.io.Encoder; +import org.apache.avro.io.EncoderFactory; +import org.apache.avro.reflect.ReflectDatumWriter; +import org.apache.avro.specific.SpecificDatumWriter; +import org.springframework.batch.item.avro.support.AvroTestFixtures; +import org.springframework.core.io.ClassPathResource; +import org.springframework.core.io.Resource; + +/** + * Used to create test data. See http://avro.apache.org/docs/1.9.0/gettingstartedjava.html + * + * @author David Turanski + */ +class AvroTestUtils { + + public static void main(String... args) { + try { + createTestDataWithNoEmbeddedSchema(); + createTestData(); + } catch (Exception e) { + e.printStackTrace(); + } + } + + static void createTestDataWithNoEmbeddedSchema() throws Exception { + + DatumWriter userDatumWriter = new SpecificDatumWriter<>(User.class); + + FileOutputStream fileOutputStream = new FileOutputStream("user-data-no-schema.avro"); + + Encoder encoder = EncoderFactory.get().binaryEncoder(fileOutputStream,null); + userDatumWriter.write(new User("David", 20, "blue"), encoder); + userDatumWriter.write(new User("Sue", 4, "red"), encoder); + userDatumWriter.write(new User("Alana", 13, "yellow"), encoder); + userDatumWriter.write(new User("Joe", 1, "pink"), encoder); + + encoder.flush(); + fileOutputStream.flush(); + fileOutputStream.close(); + } + + + static void createTestData() throws Exception { + + Resource schemaResource = new ClassPathResource("org/springframework/batch/item/avro/user-schema.json"); + + DatumWriter userDatumWriter = new SpecificDatumWriter<>(User.class); + DataFileWriter dataFileWriter = new DataFileWriter<>(userDatumWriter); + dataFileWriter.create(new Schema.Parser().parse(schemaResource.getInputStream()), new File("users.avro")); + dataFileWriter.append(new User("David", 20, "blue")); + dataFileWriter.append(new User("Sue", 4, "red")); + dataFileWriter.append(new User("Alana", 13, "yellow")); + dataFileWriter.append(new User("Joe", 1, "pink")); + dataFileWriter.close(); + } + + +} diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/avro/example/User.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/avro/example/User.java new file mode 100644 index 000000000..c4b13454c --- /dev/null +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/avro/example/User.java @@ -0,0 +1,518 @@ +/* + * Copyright 2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +/** + * Autogenerated by Avro + * + * DO NOT EDIT DIRECTLY + */ +package org.springframework.batch.item.avro.example; + +import org.apache.avro.generic.GenericArray; +import org.apache.avro.specific.SpecificData; +import org.apache.avro.util.Utf8; +import org.apache.avro.message.BinaryMessageEncoder; +import org.apache.avro.message.BinaryMessageDecoder; +import org.apache.avro.message.SchemaStore; + +@org.apache.avro.specific.AvroGenerated +public class User extends org.apache.avro.specific.SpecificRecordBase implements org.apache.avro.specific.SpecificRecord { + private static final long serialVersionUID = 1293362237195430714L; + public static final org.apache.avro.Schema SCHEMA$ = new org.apache.avro.Schema.Parser().parse("{\"type\":\"record\",\"name\":\"User\",\"namespace\":\"org.springframework.batch.item.avro.example\",\"fields\":[{\"name\":\"name\",\"type\":\"string\"},{\"name\":\"favorite_number\",\"type\":[\"int\",\"null\"]},{\"name\":\"favorite_color\",\"type\":[\"string\",\"null\"]}]}"); + public static org.apache.avro.Schema getClassSchema() { return SCHEMA$; } + + private static SpecificData MODEL$ = new SpecificData(); + + private static final BinaryMessageEncoder ENCODER = + new BinaryMessageEncoder(MODEL$, SCHEMA$); + + private static final BinaryMessageDecoder DECODER = + new BinaryMessageDecoder(MODEL$, SCHEMA$); + + /** + * Return the BinaryMessageEncoder instance used by this class. + * @return the message encoder used by this class + */ + public static BinaryMessageEncoder getEncoder() { + return ENCODER; + } + + /** + * Return the BinaryMessageDecoder instance used by this class. + * @return the message decoder used by this class + */ + public static BinaryMessageDecoder getDecoder() { + return DECODER; + } + + /** + * Create a new BinaryMessageDecoder instance for this class that uses the specified {@link SchemaStore}. + * @param resolver a {@link SchemaStore} used to find schemas by fingerprint + * @return a BinaryMessageDecoder instance for this class backed by the given SchemaStore + */ + public static BinaryMessageDecoder createDecoder(SchemaStore resolver) { + return new BinaryMessageDecoder(MODEL$, SCHEMA$, resolver); + } + + /** + * Serializes this User to a ByteBuffer. + * @return a buffer holding the serialized data for this instance + * @throws java.io.IOException if this instance could not be serialized + */ + public java.nio.ByteBuffer toByteBuffer() throws java.io.IOException { + return ENCODER.encode(this); + } + + /** + * Deserializes a User from a ByteBuffer. + * @param b a byte buffer holding serialized data for an instance of this class + * @return a User instance decoded from the given buffer + * @throws java.io.IOException if the given bytes could not be deserialized into an instance of this class + */ + public static User fromByteBuffer( + java.nio.ByteBuffer b) throws java.io.IOException { + return DECODER.decode(b); + } + + private CharSequence name; + private Integer favorite_number; + private CharSequence favorite_color; + + /** + * Default constructor. Note that this does not initialize fields + * to their default values from the SCHEMA. If that is desired then + * one should use newBuilder(). + */ + public User() {} + + /** + * All-args constructor. + * @param name The new value for name + * @param favorite_number The new value for favorite_number + * @param favorite_color The new value for favorite_color + */ + public User(CharSequence name, Integer favorite_number, CharSequence favorite_color) { + this.name = name; + this.favorite_number = favorite_number; + this.favorite_color = favorite_color; + } + + public SpecificData getSpecificData() { return MODEL$; } + public org.apache.avro.Schema getSchema() { return SCHEMA$; } + // Used by DatumWriter. Applications should not call. + public Object get(int field$) { + switch (field$) { + case 0: return name; + case 1: return favorite_number; + case 2: return favorite_color; + default: throw new org.apache.avro.AvroRuntimeException("Bad index"); + } + } + + // Used by DatumReader. Applications should not call. + @SuppressWarnings(value="unchecked") + public void put(int field$, Object value$) { + switch (field$) { + case 0: name = (CharSequence)value$; break; + case 1: favorite_number = (Integer)value$; break; + case 2: favorite_color = (CharSequence)value$; break; + default: throw new org.apache.avro.AvroRuntimeException("Bad index"); + } + } + + /** + * Gets the value of the 'name' field. + * @return The value of the 'name' field. + */ + public CharSequence getName() { + return name; + } + + + /** + * Sets the value of the 'name' field. + * @param value the value to set. + */ + public void setName(CharSequence value) { + this.name = value; + } + + /** + * Gets the value of the 'favorite_number' field. + * @return The value of the 'favorite_number' field. + */ + public Integer getFavoriteNumber() { + return favorite_number; + } + + + /** + * Sets the value of the 'favorite_number' field. + * @param value the value to set. + */ + public void setFavoriteNumber(Integer value) { + this.favorite_number = value; + } + + /** + * Gets the value of the 'favorite_color' field. + * @return The value of the 'favorite_color' field. + */ + public CharSequence getFavoriteColor() { + return favorite_color; + } + + + /** + * Sets the value of the 'favorite_color' field. + * @param value the value to set. + */ + public void setFavoriteColor(CharSequence value) { + this.favorite_color = value; + } + + /** + * Creates a new User RecordBuilder. + * @return A new User RecordBuilder + */ + public static Builder newBuilder() { + return new Builder(); + } + + /** + * Creates a new User RecordBuilder by copying an existing Builder. + * @param other The existing builder to copy. + * @return A new User RecordBuilder + */ + public static Builder newBuilder(Builder other) { + if (other == null) { + return new Builder(); + } else { + return new Builder(other); + } + } + + /** + * Creates a new User RecordBuilder by copying an existing User instance. + * @param other The existing instance to copy. + * @return A new User RecordBuilder + */ + public static Builder newBuilder(User other) { + if (other == null) { + return new Builder(); + } else { + return new Builder(other); + } + } + + /** + * RecordBuilder for User instances. + */ + public static class Builder extends org.apache.avro.specific.SpecificRecordBuilderBase + implements org.apache.avro.data.RecordBuilder { + + private CharSequence name; + private Integer favorite_number; + private CharSequence favorite_color; + + /** Creates a new Builder */ + private Builder() { + super(SCHEMA$); + } + + /** + * Creates a Builder by copying an existing Builder. + * @param other The existing Builder to copy. + */ + private Builder(Builder other) { + super(other); + if (isValidValue(fields()[0], other.name)) { + this.name = data().deepCopy(fields()[0].schema(), other.name); + fieldSetFlags()[0] = other.fieldSetFlags()[0]; + } + if (isValidValue(fields()[1], other.favorite_number)) { + this.favorite_number = data().deepCopy(fields()[1].schema(), other.favorite_number); + fieldSetFlags()[1] = other.fieldSetFlags()[1]; + } + if (isValidValue(fields()[2], other.favorite_color)) { + this.favorite_color = data().deepCopy(fields()[2].schema(), other.favorite_color); + fieldSetFlags()[2] = other.fieldSetFlags()[2]; + } + } + + /** + * Creates a Builder by copying an existing User instance + * @param other The existing instance to copy. + */ + private Builder(User other) { + super(SCHEMA$); + if (isValidValue(fields()[0], other.name)) { + this.name = data().deepCopy(fields()[0].schema(), other.name); + fieldSetFlags()[0] = true; + } + if (isValidValue(fields()[1], other.favorite_number)) { + this.favorite_number = data().deepCopy(fields()[1].schema(), other.favorite_number); + fieldSetFlags()[1] = true; + } + if (isValidValue(fields()[2], other.favorite_color)) { + this.favorite_color = data().deepCopy(fields()[2].schema(), other.favorite_color); + fieldSetFlags()[2] = true; + } + } + + /** + * Gets the value of the 'name' field. + * @return The value. + */ + public CharSequence getName() { + return name; + } + + + /** + * Sets the value of the 'name' field. + * @param value The value of 'name'. + * @return This builder. + */ + public Builder setName(CharSequence value) { + validate(fields()[0], value); + this.name = value; + fieldSetFlags()[0] = true; + return this; + } + + /** + * Checks whether the 'name' field has been set. + * @return True if the 'name' field has been set, false otherwise. + */ + public boolean hasName() { + return fieldSetFlags()[0]; + } + + + /** + * Clears the value of the 'name' field. + * @return This builder. + */ + public Builder clearName() { + name = null; + fieldSetFlags()[0] = false; + return this; + } + + /** + * Gets the value of the 'favorite_number' field. + * @return The value. + */ + public Integer getFavoriteNumber() { + return favorite_number; + } + + + /** + * Sets the value of the 'favorite_number' field. + * @param value The value of 'favorite_number'. + * @return This builder. + */ + public Builder setFavoriteNumber(Integer value) { + validate(fields()[1], value); + this.favorite_number = value; + fieldSetFlags()[1] = true; + return this; + } + + /** + * Checks whether the 'favorite_number' field has been set. + * @return True if the 'favorite_number' field has been set, false otherwise. + */ + public boolean hasFavoriteNumber() { + return fieldSetFlags()[1]; + } + + + /** + * Clears the value of the 'favorite_number' field. + * @return This builder. + */ + public Builder clearFavoriteNumber() { + favorite_number = null; + fieldSetFlags()[1] = false; + return this; + } + + /** + * Gets the value of the 'favorite_color' field. + * @return The value. + */ + public CharSequence getFavoriteColor() { + return favorite_color; + } + + + /** + * Sets the value of the 'favorite_color' field. + * @param value The value of 'favorite_color'. + * @return This builder. + */ + public Builder setFavoriteColor(CharSequence value) { + validate(fields()[2], value); + this.favorite_color = value; + fieldSetFlags()[2] = true; + return this; + } + + /** + * Checks whether the 'favorite_color' field has been set. + * @return True if the 'favorite_color' field has been set, false otherwise. + */ + public boolean hasFavoriteColor() { + return fieldSetFlags()[2]; + } + + + /** + * Clears the value of the 'favorite_color' field. + * @return This builder. + */ + public Builder clearFavoriteColor() { + favorite_color = null; + fieldSetFlags()[2] = false; + return this; + } + + @Override + @SuppressWarnings("unchecked") + public User build() { + try { + User record = new User(); + record.name = fieldSetFlags()[0] ? this.name : (CharSequence) defaultValue(fields()[0]); + record.favorite_number = fieldSetFlags()[1] ? this.favorite_number : (Integer) defaultValue(fields()[1]); + record.favorite_color = fieldSetFlags()[2] ? this.favorite_color : (CharSequence) defaultValue(fields()[2]); + return record; + } catch (org.apache.avro.AvroMissingFieldException e) { + throw e; + } catch (Exception e) { + throw new org.apache.avro.AvroRuntimeException(e); + } + } + } + + @SuppressWarnings("unchecked") + private static final org.apache.avro.io.DatumWriter + WRITER$ = (org.apache.avro.io.DatumWriter)MODEL$.createDatumWriter(SCHEMA$); + + @Override public void writeExternal(java.io.ObjectOutput out) + throws java.io.IOException { + WRITER$.write(this, SpecificData.getEncoder(out)); + } + + @SuppressWarnings("unchecked") + private static final org.apache.avro.io.DatumReader + READER$ = (org.apache.avro.io.DatumReader)MODEL$.createDatumReader(SCHEMA$); + + @Override public void readExternal(java.io.ObjectInput in) + throws java.io.IOException { + READER$.read(this, SpecificData.getDecoder(in)); + } + + @Override protected boolean hasCustomCoders() { return true; } + + @Override public void customEncode(org.apache.avro.io.Encoder out) + throws java.io.IOException + { + out.writeString(this.name); + + if (this.favorite_number == null) { + out.writeIndex(1); + out.writeNull(); + } else { + out.writeIndex(0); + out.writeInt(this.favorite_number); + } + + if (this.favorite_color == null) { + out.writeIndex(1); + out.writeNull(); + } else { + out.writeIndex(0); + out.writeString(this.favorite_color); + } + + } + + @Override public void customDecode(org.apache.avro.io.ResolvingDecoder in) + throws java.io.IOException + { + org.apache.avro.Schema.Field[] fieldOrder = in.readFieldOrderIfDiff(); + if (fieldOrder == null) { + this.name = in.readString(this.name instanceof Utf8 ? (Utf8)this.name : null); + + if (in.readIndex() != 0) { + in.readNull(); + this.favorite_number = null; + } else { + this.favorite_number = in.readInt(); + } + + if (in.readIndex() != 0) { + in.readNull(); + this.favorite_color = null; + } else { + this.favorite_color = in.readString(this.favorite_color instanceof Utf8 ? (Utf8)this.favorite_color : null); + } + + } else { + for (int i = 0; i < 3; i++) { + switch (fieldOrder[i].pos()) { + case 0: + this.name = in.readString(this.name instanceof Utf8 ? (Utf8)this.name : null); + break; + + case 1: + if (in.readIndex() != 0) { + in.readNull(); + this.favorite_number = null; + } else { + this.favorite_number = in.readInt(); + } + break; + + case 2: + if (in.readIndex() != 0) { + in.readNull(); + this.favorite_color = null; + } else { + this.favorite_color = in.readString(this.favorite_color instanceof Utf8 ? (Utf8)this.favorite_color : null); + } + break; + + default: + throw new java.io.IOException("Corrupt ResolvingDecoder."); + } + } + } + } +} + + + + + + + + + + diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/avro/support/AvroItemReaderTestSupport.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/avro/support/AvroItemReaderTestSupport.java new file mode 100644 index 000000000..62bb321ad --- /dev/null +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/avro/support/AvroItemReaderTestSupport.java @@ -0,0 +1,47 @@ +/* + * Copyright 2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.batch.item.avro.support; + +import java.util.ArrayList; +import java.util.List; + +import org.springframework.batch.item.ExecutionContext; +import org.springframework.batch.item.avro.AvroItemReader; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * @author David Turanski + */ +public abstract class AvroItemReaderTestSupport extends AvroTestFixtures { + + protected void verify(AvroItemReader avroItemReader, List actual) throws Exception { + + avroItemReader.open(new ExecutionContext()); + List users = new ArrayList<>(); + + T user; + while ((user = avroItemReader.read()) != null) { + users.add(user); + } + + assertThat(users).hasSize(4); + assertThat(users).containsExactlyInAnyOrder(actual.get(0), actual.get(1), actual.get(2), actual.get(3)); + + avroItemReader.close(); + } +} diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/avro/support/AvroItemWriterTestSupport.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/avro/support/AvroItemWriterTestSupport.java new file mode 100644 index 000000000..fb9cf2ffc --- /dev/null +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/avro/support/AvroItemWriterTestSupport.java @@ -0,0 +1,137 @@ +/* + * Copyright 2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.batch.item.avro.support; + +import java.io.File; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.net.URI; +import java.net.URL; +import java.util.ArrayList; +import java.util.List; + +import org.springframework.batch.item.ExecutionContext; +import org.springframework.batch.item.avro.AvroItemReader; +import org.springframework.batch.item.avro.builder.AvroItemReaderBuilder; +import org.springframework.core.io.ByteArrayResource; +import org.springframework.core.io.Resource; +import org.springframework.core.io.WritableResource; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * @author David Turanski + */ +public abstract class AvroItemWriterTestSupport extends AvroTestFixtures { + + /* + * This item reader configured for Specific Avro types. + */ + protected void verifyRecords(byte[] bytes, List actual, Class clazz, boolean embeddedSchema) throws Exception { + doVerify(bytes, clazz, actual, embeddedSchema); + } + + protected void verifyRecordsWithEmbeddedHeader(byte[] bytes, List actual, Class clazz) throws Exception { + doVerify(bytes, clazz, actual, true); + } + + + private void doVerify(byte[] bytes, Class clazz, List actual, boolean embeddedSchema) throws Exception { + AvroItemReader avroItemReader = new AvroItemReaderBuilder() + .type(clazz) + .resource(new ByteArrayResource(bytes)) + .embeddedSchema(embeddedSchema) + .build(); + + avroItemReader.open(new ExecutionContext()); + + List records = new ArrayList<>(); + T record; + while ((record = avroItemReader.read()) != null) { + records.add(record); + } + assertThat(records).hasSize(4); + assertThat(records).containsExactlyInAnyOrder(actual.get(0), actual.get(1), actual.get(2), actual.get(3)); + } + + + protected static class OutputStreamResource implements WritableResource { + + final private OutputStream outputStream; + + public OutputStreamResource(OutputStream outputStream) { + this.outputStream = outputStream; + } + + @Override + public OutputStream getOutputStream() throws IOException { + return this.outputStream; + } + + @Override + public boolean exists() { + return true; + } + + @Override + public URL getURL() throws IOException { + return null; + } + + @Override + public URI getURI() throws IOException { + return null; + } + + @Override + public File getFile() throws IOException { + return null; + } + + @Override + public long contentLength() throws IOException { + return 0; + } + + @Override + public long lastModified() throws IOException { + return 0; + } + + @Override + public Resource createRelative(String relativePath) throws IOException { + return null; + } + + @Override + public String getFilename() { + return null; + } + + @Override + public String getDescription() { + return "Output stream resource"; + } + + @Override + public InputStream getInputStream() throws IOException { + return null; + } + } + +} diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/avro/support/AvroTestFixtures.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/avro/support/AvroTestFixtures.java new file mode 100644 index 000000000..11af7ac80 --- /dev/null +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/avro/support/AvroTestFixtures.java @@ -0,0 +1,185 @@ +/* + * Copyright 2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.batch.item.avro.support; + +import java.io.FileOutputStream; +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Paths; +import java.util.Arrays; +import java.util.List; +import java.util.Objects; +import java.util.stream.Collectors; + +import org.apache.avro.Schema; +import org.apache.avro.generic.GenericData; +import org.apache.avro.generic.GenericRecord; +import org.apache.avro.io.DatumWriter; +import org.apache.avro.io.Encoder; +import org.apache.avro.io.EncoderFactory; +import org.apache.avro.reflect.ReflectData; +import org.apache.avro.reflect.ReflectDatumWriter; +import org.springframework.batch.item.avro.example.User; +import org.springframework.core.io.ByteArrayResource; +import org.springframework.core.io.ClassPathResource; +import org.springframework.core.io.Resource; + +/** + * @author David Turanski + */ +public abstract class AvroTestFixtures { + + + //@formatter:off + private final List avroGeneratedUsers = Arrays.asList( + new User("David", 20, "blue"), + new User("Sue", 4, "red"), + new User("Alana", 13, "yellow"), + new User("Joe", 1, "pink")); + + List plainOldUsers = Arrays.asList( + new PlainOldUser("David", 20, "blue"), + new PlainOldUser("Sue", 4, "red"), + new PlainOldUser("Alana", 13, "yellow"), + new PlainOldUser("Joe", 1, "pink")); + //@formatter:on + + + protected Resource schemaResource = new ClassPathResource("org/springframework/batch/item/avro/user-schema.json"); + + + protected Resource plainOldUserSchemaResource = new ByteArrayResource(PlainOldUser.SCHEMA.toString().getBytes()); + + // Serialized data only + protected Resource dataResource = new ClassPathResource( + "org/springframework/batch/item/avro/user-data-no-schema.avro"); + + // Data written with DataFileWriter, includes embedded SCHEMA (more common) + protected Resource dataResourceWithSchema = new ClassPathResource( + "org/springframework/batch/item/avro/user-data.avro"); + + protected Resource plainOldUserDataResource + = new ClassPathResource("org/springframework/batch/item/avro/plain-old-user-data-no-schema.avro"); + + protected String schemaString(Resource resource) { + { + String content; + try { + content = new String(Files.readAllBytes(Paths.get(resource.getFile().getAbsolutePath()))); + } + catch (IOException e) { + throw new IllegalArgumentException(e.getMessage(), e); + } + return content; + } + } + + protected List avroGeneratedUsers() { + return this.avroGeneratedUsers; + } + + protected List genericAvroGeneratedUsers() { + return this.avroGeneratedUsers.stream().map(u-> { + GenericData.Record avroRecord; + avroRecord = new GenericData.Record(u.getSchema()); + avroRecord.put("name", u.getName()); + avroRecord.put("favorite_number", u.getFavoriteNumber()); + avroRecord.put("favorite_color",u.getFavoriteColor()); + return avroRecord; + } + ).collect(Collectors.toList()); + } + + protected List plainOldUsers() { + return this.plainOldUsers; + } + + protected List genericPlainOldUsers() { + return this.plainOldUsers.stream().map(PlainOldUser::toGenericRecord).collect(Collectors.toList()); + } + + protected static class PlainOldUser { + public static final Schema SCHEMA = ReflectData.get().getSchema(PlainOldUser.class); + private CharSequence name; + + private int favoriteNumber; + + private CharSequence favoriteColor; + + public PlainOldUser(){ + + } + + public PlainOldUser(CharSequence name, int favoriteNumber, CharSequence favoriteColor) { + this.name = name; + this.favoriteNumber = favoriteNumber; + this.favoriteColor = favoriteColor; + } + + public String getName() { + return name.toString(); + } + + public int getFavoriteNumber() { + return favoriteNumber; + } + + public String getFavoriteColor() { + return favoriteColor.toString(); + } + + public GenericRecord toGenericRecord() { + GenericData.Record avroRecord = new GenericData.Record(SCHEMA); + avroRecord.put("name", this.name); + avroRecord.put("favoriteNumber", this.favoriteNumber); + avroRecord.put("favoriteColor",this.favoriteColor); + return avroRecord; + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + PlainOldUser that = (PlainOldUser) o; + return favoriteNumber == that.favoriteNumber && + Objects.equals(name, that.name) && + Objects.equals(favoriteColor, that.favoriteColor); + } + + @Override + public int hashCode() { + return Objects.hash(name, favoriteNumber, favoriteColor); + } + } + + public static void createPlainOldUsersWithNoEmbeddedSchema() throws Exception { + + DatumWriter userDatumWriter = new ReflectDatumWriter<>(AvroTestFixtures.PlainOldUser.class); + + FileOutputStream fileOutputStream = new FileOutputStream("plain-old-user-data-no-schema.avro"); + + Encoder encoder = EncoderFactory.get().binaryEncoder(fileOutputStream,null); + userDatumWriter.write(new PlainOldUser("David", 20, "blue"), encoder); + userDatumWriter.write(new PlainOldUser("Sue", 4, "red"), encoder); + userDatumWriter.write(new PlainOldUser("Alana", 13, "yellow"), encoder); + userDatumWriter.write(new PlainOldUser("Joe", 1, "pink"), encoder); + + encoder.flush(); + fileOutputStream.flush(); + fileOutputStream.close(); + } +} diff --git a/spring-batch-infrastructure/src/test/resources/org/springframework/batch/item/avro/plain-old-user-data-no-schema.avro b/spring-batch-infrastructure/src/test/resources/org/springframework/batch/item/avro/plain-old-user-data-no-schema.avro new file mode 100644 index 000000000..104e8581c --- /dev/null +++ b/spring-batch-infrastructure/src/test/resources/org/springframework/batch/item/avro/plain-old-user-data-no-schema.avro @@ -0,0 +1,3 @@ + +David(blueSuered +Alana yellowJoepink \ No newline at end of file diff --git a/spring-batch-infrastructure/src/test/resources/org/springframework/batch/item/avro/user-data-no-schema.avro b/spring-batch-infrastructure/src/test/resources/org/springframework/batch/item/avro/user-data-no-schema.avro new file mode 100644 index 0000000000000000000000000000000000000000..1b51bdec07afc797abddddd4c5080760ddf72e1d GIT binary patch literal 53 zcmd;bNi54uVbEaUNXjWqWeYA%W#C|7D@sk_a?DB0OJtB@;HgZ_$;mHg^U631C1)`u_t#>-eM6? z#?835PUum6g;cD?zgKFq&+7WJk{WLLVXtw%PGtVPjm2 zdeF7XyY}C06A(JYEwz*|{Vz}$Y2w=6e_$UY(=Ge?y!w9Geb6)PzQ%8_Pw8TKgN-QJ nB};QzQ*vJuOIhNm99~P*$S&B~nxr(V