Repository: nifi Updated Branches: refs/heads/master 57ccf97c5 -> 50ea1083e
http://git-wip-us.apache.org/repos/asf/nifi/blob/50ea1083/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/avro/TestAvroReaderWithEmbeddedSchema.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/avro/TestAvroReaderWithEmbeddedSchema.java b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/avro/TestAvroReaderWithEmbeddedSchema.java new file mode 100644 index 0000000..da9f70b --- /dev/null +++ b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/avro/TestAvroReaderWithEmbeddedSchema.java @@ -0,0 +1,290 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 + * + * http://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.apache.nifi.avro; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.File; +import java.io.IOException; +import java.io.InputStream; +import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; +import java.text.DateFormat; +import java.text.ParseException; +import java.text.SimpleDateFormat; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.TimeZone; + +import org.apache.avro.Schema; +import org.apache.avro.Schema.Field; +import org.apache.avro.Schema.Type; +import org.apache.avro.file.DataFileWriter; +import org.apache.avro.generic.GenericData; +import org.apache.avro.generic.GenericDatumWriter; +import org.apache.avro.generic.GenericRecord; +import org.apache.avro.io.DatumWriter; +import org.apache.nifi.schema.access.SchemaNotFoundException; +import org.apache.nifi.serialization.MalformedRecordException; +import org.apache.nifi.serialization.SimpleRecordSchema; +import org.apache.nifi.serialization.record.MapRecord; +import org.apache.nifi.serialization.record.Record; +import org.apache.nifi.serialization.record.RecordField; +import org.apache.nifi.serialization.record.RecordFieldType; +import org.apache.nifi.serialization.record.RecordSchema; +import org.junit.Test; + +public class TestAvroReaderWithEmbeddedSchema { + + + @Test + public void testLogicalTypes() throws IOException, ParseException, MalformedRecordException, SchemaNotFoundException { + final Schema schema = new Schema.Parser().parse(new File("src/test/resources/avro/logical-types.avsc")); + + final ByteArrayOutputStream baos = new ByteArrayOutputStream(); + + final String expectedTime = "2017-04-04 14:20:33.000"; + final DateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS"); + df.setTimeZone(TimeZone.getTimeZone("gmt")); + final long timeLong = df.parse(expectedTime).getTime(); + + final long secondsSinceMidnight = 33 + (20 * 60) + (14 * 60 * 60); + final long millisSinceMidnight = secondsSinceMidnight * 1000L; + + + final byte[] serialized; + final DatumWriter<GenericRecord> datumWriter = new GenericDatumWriter<>(schema); + try (final DataFileWriter<GenericRecord> dataFileWriter = new DataFileWriter<>(datumWriter); + final DataFileWriter<GenericRecord> writer = dataFileWriter.create(schema, baos)) { + + final GenericRecord record = new GenericData.Record(schema); + record.put("timeMillis", millisSinceMidnight); + record.put("timeMicros", millisSinceMidnight * 1000L); + record.put("timestampMillis", timeLong); + record.put("timestampMicros", timeLong * 1000L); + record.put("date", 17260); + + writer.append(record); + writer.flush(); + + serialized = baos.toByteArray(); + } + + try (final InputStream in = new ByteArrayInputStream(serialized)) { + final AvroRecordReader reader = new AvroReaderWithEmbeddedSchema(in); + final RecordSchema recordSchema = reader.getSchema(); + + assertEquals(RecordFieldType.TIME, recordSchema.getDataType("timeMillis").get().getFieldType()); + assertEquals(RecordFieldType.TIME, recordSchema.getDataType("timeMicros").get().getFieldType()); + assertEquals(RecordFieldType.TIMESTAMP, recordSchema.getDataType("timestampMillis").get().getFieldType()); + assertEquals(RecordFieldType.TIMESTAMP, recordSchema.getDataType("timestampMicros").get().getFieldType()); + assertEquals(RecordFieldType.DATE, recordSchema.getDataType("date").get().getFieldType()); + + final Record record = reader.nextRecord(); + assertEquals(new java.sql.Time(millisSinceMidnight), record.getValue("timeMillis")); + assertEquals(new java.sql.Time(millisSinceMidnight), record.getValue("timeMicros")); + assertEquals(new java.sql.Timestamp(timeLong), record.getValue("timestampMillis")); + assertEquals(new java.sql.Timestamp(timeLong), record.getValue("timestampMicros")); + final DateFormat noTimeOfDayDateFormat = new SimpleDateFormat("yyyy-MM-dd"); + noTimeOfDayDateFormat.setTimeZone(TimeZone.getTimeZone("gmt")); + assertEquals(new java.sql.Date(timeLong).toString(), noTimeOfDayDateFormat.format(record.getValue("date"))); + } + } + + @Test + public void testDataTypes() throws IOException, MalformedRecordException, SchemaNotFoundException { + final List<Field> accountFields = new ArrayList<>(); + accountFields.add(new Field("accountId", Schema.create(Type.LONG), null, (Object) null)); + accountFields.add(new Field("accountName", Schema.create(Type.STRING), null, (Object) null)); + final Schema accountSchema = Schema.createRecord("account", null, null, false); + accountSchema.setFields(accountFields); + + final List<Field> catFields = new ArrayList<>(); + catFields.add(new Field("catTailLength", Schema.create(Type.INT), null, (Object) null)); + catFields.add(new Field("catName", Schema.create(Type.STRING), null, (Object) null)); + final Schema catSchema = Schema.createRecord("cat", null, null, false); + catSchema.setFields(catFields); + + final List<Field> dogFields = new ArrayList<>(); + dogFields.add(new Field("dogTailLength", Schema.create(Type.INT), null, (Object) null)); + dogFields.add(new Field("dogName", Schema.create(Type.STRING), null, (Object) null)); + final Schema dogSchema = Schema.createRecord("dog", null, null, false); + dogSchema.setFields(dogFields); + + final List<Field> fields = new ArrayList<>(); + fields.add(new Field("name", Schema.create(Type.STRING), null, (Object) null)); + fields.add(new Field("age", Schema.create(Type.INT), null, (Object) null)); + fields.add(new Field("balance", Schema.create(Type.DOUBLE), null, (Object) null)); + fields.add(new Field("rate", Schema.create(Type.FLOAT), null, (Object) null)); + fields.add(new Field("debt", Schema.create(Type.BOOLEAN), null, (Object) null)); + fields.add(new Field("nickname", Schema.create(Type.NULL), null, (Object) null)); + fields.add(new Field("binary", Schema.create(Type.BYTES), null, (Object) null)); + fields.add(new Field("fixed", Schema.createFixed("fixed", null, null, 5), null, (Object) null)); + fields.add(new Field("map", Schema.createMap(Schema.create(Type.STRING)), null, (Object) null)); + fields.add(new Field("array", Schema.createArray(Schema.create(Type.LONG)), null, (Object) null)); + fields.add(new Field("account", accountSchema, null, (Object) null)); + fields.add(new Field("desiredbalance", Schema.createUnion( // test union of NULL and other type with no value + Arrays.asList(Schema.create(Type.NULL), Schema.create(Type.DOUBLE))), + null, (Object) null)); + fields.add(new Field("dreambalance", Schema.createUnion( // test union of NULL and other type with a value + Arrays.asList(Schema.create(Type.NULL), Schema.create(Type.DOUBLE))), + null, (Object) null)); + fields.add(new Field("favAnimal", Schema.createUnion(Arrays.asList(catSchema, dogSchema)), null, (Object) null)); + fields.add(new Field("otherFavAnimal", Schema.createUnion(Arrays.asList(catSchema, dogSchema)), null, (Object) null)); + + final Schema schema = Schema.createRecord("record", null, null, false); + schema.setFields(fields); + + final byte[] source; + final ByteArrayOutputStream baos = new ByteArrayOutputStream(); + + final Map<String, String> map = new HashMap<>(); + map.put("greeting", "hello"); + map.put("salutation", "good-bye"); + + final DatumWriter<GenericRecord> datumWriter = new GenericDatumWriter<>(schema); + try (final DataFileWriter<GenericRecord> dataFileWriter = new DataFileWriter<>(datumWriter); + final DataFileWriter<GenericRecord> writer = dataFileWriter.create(schema, baos)) { + + final GenericRecord record = new GenericData.Record(schema); + record.put("name", "John"); + record.put("age", 33); + record.put("balance", 1234.56D); + record.put("rate", 0.045F); + record.put("debt", false); + record.put("binary", ByteBuffer.wrap("binary".getBytes(StandardCharsets.UTF_8))); + record.put("fixed", new GenericData.Fixed(Schema.create(Type.BYTES), "fixed".getBytes(StandardCharsets.UTF_8))); + record.put("map", map); + record.put("array", Arrays.asList(1L, 2L)); + record.put("dreambalance", 10_000_000.00D); + + final GenericRecord accountRecord = new GenericData.Record(accountSchema); + accountRecord.put("accountId", 83L); + accountRecord.put("accountName", "Checking"); + record.put("account", accountRecord); + + final GenericRecord catRecord = new GenericData.Record(catSchema); + catRecord.put("catTailLength", 1); + catRecord.put("catName", "Meow"); + record.put("otherFavAnimal", catRecord); + + final GenericRecord dogRecord = new GenericData.Record(dogSchema); + dogRecord.put("dogTailLength", 14); + dogRecord.put("dogName", "Fido"); + record.put("favAnimal", dogRecord); + + writer.append(record); + } + + source = baos.toByteArray(); + + try (final InputStream in = new ByteArrayInputStream(source)) { + final AvroRecordReader reader = new AvroReaderWithEmbeddedSchema(in); + final RecordSchema recordSchema = reader.getSchema(); + assertEquals(15, recordSchema.getFieldCount()); + + assertEquals(RecordFieldType.STRING, recordSchema.getDataType("name").get().getFieldType()); + assertEquals(RecordFieldType.INT, recordSchema.getDataType("age").get().getFieldType()); + assertEquals(RecordFieldType.DOUBLE, recordSchema.getDataType("balance").get().getFieldType()); + assertEquals(RecordFieldType.FLOAT, recordSchema.getDataType("rate").get().getFieldType()); + assertEquals(RecordFieldType.BOOLEAN, recordSchema.getDataType("debt").get().getFieldType()); + assertEquals(RecordFieldType.STRING, recordSchema.getDataType("nickname").get().getFieldType()); + assertEquals(RecordFieldType.ARRAY, recordSchema.getDataType("binary").get().getFieldType()); + assertEquals(RecordFieldType.ARRAY, recordSchema.getDataType("fixed").get().getFieldType()); + assertEquals(RecordFieldType.MAP, recordSchema.getDataType("map").get().getFieldType()); + assertEquals(RecordFieldType.ARRAY, recordSchema.getDataType("array").get().getFieldType()); + assertEquals(RecordFieldType.RECORD, recordSchema.getDataType("account").get().getFieldType()); + assertEquals(RecordFieldType.DOUBLE, recordSchema.getDataType("desiredbalance").get().getFieldType()); + assertEquals(RecordFieldType.DOUBLE, recordSchema.getDataType("dreambalance").get().getFieldType()); + assertEquals(RecordFieldType.CHOICE, recordSchema.getDataType("favAnimal").get().getFieldType()); + assertEquals(RecordFieldType.CHOICE, recordSchema.getDataType("otherFavAnimal").get().getFieldType()); + + final Object[] values = reader.nextRecord().getValues(); + assertEquals(15, values.length); + assertEquals("John", values[0]); + assertEquals(33, values[1]); + assertEquals(1234.56D, values[2]); + assertEquals(0.045F, values[3]); + assertEquals(false, values[4]); + assertEquals(null, values[5]); + assertArrayEquals(toObjectArray("binary".getBytes(StandardCharsets.UTF_8)), (Object[]) values[6]); + assertArrayEquals(toObjectArray("fixed".getBytes(StandardCharsets.UTF_8)), (Object[]) values[7]); + assertEquals(map, values[8]); + assertArrayEquals(new Object[] {1L, 2L}, (Object[]) values[9]); + + final Map<String, Object> accountValues = new HashMap<>(); + accountValues.put("accountName", "Checking"); + accountValues.put("accountId", 83L); + + final List<RecordField> accountRecordFields = new ArrayList<>(); + accountRecordFields.add(new RecordField("accountId", RecordFieldType.LONG.getDataType())); + accountRecordFields.add(new RecordField("accountName", RecordFieldType.STRING.getDataType())); + + final RecordSchema accountRecordSchema = new SimpleRecordSchema(accountRecordFields); + final Record mapRecord = new MapRecord(accountRecordSchema, accountValues); + + assertEquals(mapRecord, values[10]); + + assertNull(values[11]); + assertEquals(10_000_000.0D, values[12]); + + final Map<String, Object> dogMap = new HashMap<>(); + dogMap.put("dogName", "Fido"); + dogMap.put("dogTailLength", 14); + + final List<RecordField> dogRecordFields = new ArrayList<>(); + dogRecordFields.add(new RecordField("dogTailLength", RecordFieldType.INT.getDataType())); + dogRecordFields.add(new RecordField("dogName", RecordFieldType.STRING.getDataType())); + final RecordSchema dogRecordSchema = new SimpleRecordSchema(dogRecordFields); + final Record dogRecord = new MapRecord(dogRecordSchema, dogMap); + + assertEquals(dogRecord, values[13]); + + final Map<String, Object> catMap = new HashMap<>(); + catMap.put("catName", "Meow"); + catMap.put("catTailLength", 1); + + final List<RecordField> catRecordFields = new ArrayList<>(); + catRecordFields.add(new RecordField("catTailLength", RecordFieldType.INT.getDataType())); + catRecordFields.add(new RecordField("catName", RecordFieldType.STRING.getDataType())); + final RecordSchema catRecordSchema = new SimpleRecordSchema(catRecordFields); + final Record catRecord = new MapRecord(catRecordSchema, catMap); + + assertEquals(catRecord, values[14]); + } + } + + private Object[] toObjectArray(final byte[] bytes) { + final Object[] array = new Object[bytes.length]; + for (int i = 0; i < bytes.length; i++) { + array[i] = Byte.valueOf(bytes[i]); + } + return array; + } + + public static enum Status { + GOOD, BAD; + } +} http://git-wip-us.apache.org/repos/asf/nifi/blob/50ea1083/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/avro/TestAvroRecordReader.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/avro/TestAvroRecordReader.java b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/avro/TestAvroRecordReader.java deleted file mode 100644 index 56e2e3d..0000000 --- a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/avro/TestAvroRecordReader.java +++ /dev/null @@ -1,296 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You 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 - * - * http://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.apache.nifi.avro; - -import static org.junit.Assert.assertArrayEquals; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNull; - -import java.io.ByteArrayInputStream; -import java.io.ByteArrayOutputStream; -import java.io.File; -import java.io.IOException; -import java.io.InputStream; -import java.nio.ByteBuffer; -import java.nio.charset.StandardCharsets; -import java.text.DateFormat; -import java.text.ParseException; -import java.text.SimpleDateFormat; -import java.util.ArrayList; -import java.util.Arrays; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.TimeZone; - -import org.apache.avro.Schema; -import org.apache.avro.Schema.Field; -import org.apache.avro.Schema.Type; -import org.apache.avro.file.DataFileWriter; -import org.apache.avro.generic.GenericData; -import org.apache.avro.generic.GenericDatumWriter; -import org.apache.avro.generic.GenericRecord; -import org.apache.avro.io.DatumWriter; -import org.apache.nifi.serialization.MalformedRecordException; -import org.apache.nifi.serialization.SimpleRecordSchema; -import org.apache.nifi.serialization.record.MapRecord; -import org.apache.nifi.serialization.record.Record; -import org.apache.nifi.serialization.record.RecordField; -import org.apache.nifi.serialization.record.RecordFieldType; -import org.apache.nifi.serialization.record.RecordSchema; -import org.junit.Test; - -public class TestAvroRecordReader { - - - @Test - public void testLogicalTypes() throws IOException, ParseException, MalformedRecordException { - final Schema schema = new Schema.Parser().parse(new File("src/test/resources/avro/logical-types.avsc")); - - final ByteArrayOutputStream baos = new ByteArrayOutputStream(); - - final String expectedTime = "2017-04-04 14:20:33.000"; - final DateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS"); - df.setTimeZone(TimeZone.getTimeZone("gmt")); - final long timeLong = df.parse(expectedTime).getTime(); - - final long secondsSinceMidnight = 33 + (20 * 60) + (14 * 60 * 60); - final long millisSinceMidnight = secondsSinceMidnight * 1000L; - - - final byte[] serialized; - final DatumWriter<GenericRecord> datumWriter = new GenericDatumWriter<>(schema); - try (final DataFileWriter<GenericRecord> dataFileWriter = new DataFileWriter<>(datumWriter); - final DataFileWriter<GenericRecord> writer = dataFileWriter.create(schema, baos)) { - - final GenericRecord record = new GenericData.Record(schema); - record.put("timeMillis", millisSinceMidnight); - record.put("timeMicros", millisSinceMidnight * 1000L); - record.put("timestampMillis", timeLong); - record.put("timestampMicros", timeLong * 1000L); - record.put("date", 17260); - - writer.append(record); - writer.flush(); - - serialized = baos.toByteArray(); - } - - try (final InputStream in = new ByteArrayInputStream(serialized)) { - final AvroRecordReader reader = new AvroRecordReader(in); - final RecordSchema recordSchema = reader.getSchema(); - - assertEquals(RecordFieldType.TIME, recordSchema.getDataType("timeMillis").get().getFieldType()); - assertEquals(RecordFieldType.TIME, recordSchema.getDataType("timeMicros").get().getFieldType()); - assertEquals(RecordFieldType.TIMESTAMP, recordSchema.getDataType("timestampMillis").get().getFieldType()); - assertEquals(RecordFieldType.TIMESTAMP, recordSchema.getDataType("timestampMicros").get().getFieldType()); - assertEquals(RecordFieldType.DATE, recordSchema.getDataType("date").get().getFieldType()); - - final Record record = reader.nextRecord(); - assertEquals(new java.sql.Time(millisSinceMidnight), record.getValue("timeMillis")); - assertEquals(new java.sql.Time(millisSinceMidnight), record.getValue("timeMicros")); - assertEquals(new java.sql.Timestamp(timeLong), record.getValue("timestampMillis")); - assertEquals(new java.sql.Timestamp(timeLong), record.getValue("timestampMicros")); - final DateFormat noTimeOfDayDateFormat = new SimpleDateFormat("yyyy-MM-dd"); - noTimeOfDayDateFormat.setTimeZone(TimeZone.getTimeZone("gmt")); - assertEquals(new java.sql.Date(timeLong).toString(), noTimeOfDayDateFormat.format(record.getValue("date"))); - } - } - - @Test - @SuppressWarnings({"unchecked", "rawtypes"}) - public void testDataTypes() throws IOException, MalformedRecordException { - final List<Field> accountFields = new ArrayList<>(); - accountFields.add(new Field("accountId", Schema.create(Type.LONG), null, (Object) null)); - accountFields.add(new Field("accountName", Schema.create(Type.STRING), null, (Object) null)); - final Schema accountSchema = Schema.createRecord("account", null, null, false); - accountSchema.setFields(accountFields); - - final List<Field> catFields = new ArrayList<>(); - catFields.add(new Field("catTailLength", Schema.create(Type.INT), null, (Object) null)); - catFields.add(new Field("catName", Schema.create(Type.STRING), null, (Object) null)); - final Schema catSchema = Schema.createRecord("cat", null, null, false); - catSchema.setFields(catFields); - - final List<Field> dogFields = new ArrayList<>(); - dogFields.add(new Field("dogTailLength", Schema.create(Type.INT), null, (Object) null)); - dogFields.add(new Field("dogName", Schema.create(Type.STRING), null, (Object) null)); - final Schema dogSchema = Schema.createRecord("dog", null, null, false); - dogSchema.setFields(dogFields); - - final List<Field> fields = new ArrayList<>(); - fields.add(new Field("name", Schema.create(Type.STRING), null, (Object) null)); - fields.add(new Field("age", Schema.create(Type.INT), null, (Object) null)); - fields.add(new Field("balance", Schema.create(Type.DOUBLE), null, (Object) null)); - fields.add(new Field("rate", Schema.create(Type.FLOAT), null, (Object) null)); - fields.add(new Field("debt", Schema.create(Type.BOOLEAN), null, (Object) null)); - fields.add(new Field("nickname", Schema.create(Type.NULL), null, (Object) null)); - fields.add(new Field("binary", Schema.create(Type.BYTES), null, (Object) null)); - fields.add(new Field("fixed", Schema.createFixed("fixed", null, null, 5), null, (Object) null)); - fields.add(new Field("map", Schema.createMap(Schema.create(Type.STRING)), null, (Object) null)); - fields.add(new Field("array", Schema.createArray(Schema.create(Type.LONG)), null, (Object) null)); - fields.add(new Field("account", accountSchema, null, (Object) null)); - fields.add(new Field("desiredbalance", Schema.createUnion( // test union of NULL and other type with no value - Arrays.asList(Schema.create(Type.NULL), Schema.create(Type.DOUBLE))), - null, (Object) null)); - fields.add(new Field("dreambalance", Schema.createUnion( // test union of NULL and other type with a value - Arrays.asList(Schema.create(Type.NULL), Schema.create(Type.DOUBLE))), - null, (Object) null)); - fields.add(new Field("favAnimal", Schema.createUnion(Arrays.asList(catSchema, dogSchema)), null, (Object) null)); - fields.add(new Field("otherFavAnimal", Schema.createUnion(Arrays.asList(catSchema, dogSchema)), null, (Object) null)); - - final Schema schema = Schema.createRecord("record", null, null, false); - schema.setFields(fields); - - final byte[] source; - final ByteArrayOutputStream baos = new ByteArrayOutputStream(); - - final Map<String, String> map = new HashMap<>(); - map.put("greeting", "hello"); - map.put("salutation", "good-bye"); - - final List<RecordField> mapFields = new ArrayList<>(); - mapFields.add(new RecordField("greeting", RecordFieldType.STRING.getDataType())); - mapFields.add(new RecordField("salutation", RecordFieldType.STRING.getDataType())); - final RecordSchema mapSchema = new SimpleRecordSchema(mapFields); - final Record expectedRecord = new MapRecord(mapSchema, (Map) map); - - final DatumWriter<GenericRecord> datumWriter = new GenericDatumWriter<>(schema); - try (final DataFileWriter<GenericRecord> dataFileWriter = new DataFileWriter<>(datumWriter); - final DataFileWriter<GenericRecord> writer = dataFileWriter.create(schema, baos)) { - - final GenericRecord record = new GenericData.Record(schema); - record.put("name", "John"); - record.put("age", 33); - record.put("balance", 1234.56D); - record.put("rate", 0.045F); - record.put("debt", false); - record.put("binary", ByteBuffer.wrap("binary".getBytes(StandardCharsets.UTF_8))); - record.put("fixed", new GenericData.Fixed(Schema.create(Type.BYTES), "fixed".getBytes(StandardCharsets.UTF_8))); - record.put("map", map); - record.put("array", Arrays.asList(1L, 2L)); - record.put("dreambalance", 10_000_000.00D); - - final GenericRecord accountRecord = new GenericData.Record(accountSchema); - accountRecord.put("accountId", 83L); - accountRecord.put("accountName", "Checking"); - record.put("account", accountRecord); - - final GenericRecord catRecord = new GenericData.Record(catSchema); - catRecord.put("catTailLength", 1); - catRecord.put("catName", "Meow"); - record.put("otherFavAnimal", catRecord); - - final GenericRecord dogRecord = new GenericData.Record(dogSchema); - dogRecord.put("dogTailLength", 14); - dogRecord.put("dogName", "Fido"); - record.put("favAnimal", dogRecord); - - writer.append(record); - } - - source = baos.toByteArray(); - - try (final InputStream in = new ByteArrayInputStream(source)) { - final AvroRecordReader reader = new AvroRecordReader(in); - final RecordSchema recordSchema = reader.getSchema(); - assertEquals(15, recordSchema.getFieldCount()); - - assertEquals(RecordFieldType.STRING, recordSchema.getDataType("name").get().getFieldType()); - assertEquals(RecordFieldType.INT, recordSchema.getDataType("age").get().getFieldType()); - assertEquals(RecordFieldType.DOUBLE, recordSchema.getDataType("balance").get().getFieldType()); - assertEquals(RecordFieldType.FLOAT, recordSchema.getDataType("rate").get().getFieldType()); - assertEquals(RecordFieldType.BOOLEAN, recordSchema.getDataType("debt").get().getFieldType()); - assertEquals(RecordFieldType.RECORD, recordSchema.getDataType("nickname").get().getFieldType()); - assertEquals(RecordFieldType.ARRAY, recordSchema.getDataType("binary").get().getFieldType()); - assertEquals(RecordFieldType.ARRAY, recordSchema.getDataType("fixed").get().getFieldType()); - assertEquals(RecordFieldType.RECORD, recordSchema.getDataType("map").get().getFieldType()); - assertEquals(RecordFieldType.ARRAY, recordSchema.getDataType("array").get().getFieldType()); - assertEquals(RecordFieldType.RECORD, recordSchema.getDataType("account").get().getFieldType()); - assertEquals(RecordFieldType.DOUBLE, recordSchema.getDataType("desiredbalance").get().getFieldType()); - assertEquals(RecordFieldType.DOUBLE, recordSchema.getDataType("dreambalance").get().getFieldType()); - assertEquals(RecordFieldType.CHOICE, recordSchema.getDataType("favAnimal").get().getFieldType()); - assertEquals(RecordFieldType.CHOICE, recordSchema.getDataType("otherFavAnimal").get().getFieldType()); - - final Object[] values = reader.nextRecord().getValues(); - assertEquals(15, values.length); - assertEquals("John", values[0]); - assertEquals(33, values[1]); - assertEquals(1234.56D, values[2]); - assertEquals(0.045F, values[3]); - assertEquals(false, values[4]); - assertEquals(null, values[5]); - assertArrayEquals(toObjectArray("binary".getBytes(StandardCharsets.UTF_8)), (Object[]) values[6]); - assertArrayEquals(toObjectArray("fixed".getBytes(StandardCharsets.UTF_8)), (Object[]) values[7]); - assertEquals(expectedRecord, values[8]); - assertArrayEquals(new Object[] {1L, 2L}, (Object[]) values[9]); - - final Map<String, Object> accountValues = new HashMap<>(); - accountValues.put("accountName", "Checking"); - accountValues.put("accountId", 83L); - - final List<RecordField> accountRecordFields = new ArrayList<>(); - accountRecordFields.add(new RecordField("accountId", RecordFieldType.LONG.getDataType())); - accountRecordFields.add(new RecordField("accountName", RecordFieldType.STRING.getDataType())); - - final RecordSchema accountRecordSchema = new SimpleRecordSchema(accountRecordFields); - final Record mapRecord = new MapRecord(accountRecordSchema, accountValues); - - assertEquals(mapRecord, values[10]); - - assertNull(values[11]); - assertEquals(10_000_000.0D, values[12]); - - final Map<String, Object> dogMap = new HashMap<>(); - dogMap.put("dogName", "Fido"); - dogMap.put("dogTailLength", 14); - - final List<RecordField> dogRecordFields = new ArrayList<>(); - dogRecordFields.add(new RecordField("dogTailLength", RecordFieldType.INT.getDataType())); - dogRecordFields.add(new RecordField("dogName", RecordFieldType.STRING.getDataType())); - final RecordSchema dogRecordSchema = new SimpleRecordSchema(dogRecordFields); - final Record dogRecord = new MapRecord(dogRecordSchema, dogMap); - - assertEquals(dogRecord, values[13]); - - final Map<String, Object> catMap = new HashMap<>(); - catMap.put("catName", "Meow"); - catMap.put("catTailLength", 1); - - final List<RecordField> catRecordFields = new ArrayList<>(); - catRecordFields.add(new RecordField("catTailLength", RecordFieldType.INT.getDataType())); - catRecordFields.add(new RecordField("catName", RecordFieldType.STRING.getDataType())); - final RecordSchema catRecordSchema = new SimpleRecordSchema(catRecordFields); - final Record catRecord = new MapRecord(catRecordSchema, catMap); - - assertEquals(catRecord, values[14]); - } - } - - private Object[] toObjectArray(final byte[] bytes) { - final Object[] array = new Object[bytes.length]; - for (int i = 0; i < bytes.length; i++) { - array[i] = Byte.valueOf(bytes[i]); - } - return array; - } - - public static enum Status { - GOOD, BAD; - } -} http://git-wip-us.apache.org/repos/asf/nifi/blob/50ea1083/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/avro/TestWriteAvroResult.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/avro/TestWriteAvroResult.java b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/avro/TestWriteAvroResult.java index 2102813..409ede2 100644 --- a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/avro/TestWriteAvroResult.java +++ b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/avro/TestWriteAvroResult.java @@ -41,13 +41,10 @@ import java.util.Map; import java.util.TimeZone; import org.apache.avro.Schema; -import org.apache.avro.file.DataFileStream; -import org.apache.avro.generic.GenericData; -import org.apache.avro.generic.GenericDatumReader; -import org.apache.avro.generic.GenericRecord; import org.apache.avro.generic.GenericData.Array; -import org.apache.avro.generic.GenericData.StringType; +import org.apache.avro.generic.GenericRecord; import org.apache.nifi.serialization.SimpleRecordSchema; +import org.apache.nifi.serialization.WriteResult; import org.apache.nifi.serialization.record.MapRecord; import org.apache.nifi.serialization.record.Record; import org.apache.nifi.serialization.record.RecordField; @@ -56,12 +53,19 @@ import org.apache.nifi.serialization.record.RecordSchema; import org.apache.nifi.serialization.record.RecordSet; import org.junit.Test; -public class TestWriteAvroResult { +public abstract class TestWriteAvroResult { + + protected abstract WriteAvroResult createWriter(Schema schema); + + protected abstract GenericRecord readRecord(InputStream in, Schema schema) throws IOException; + + protected void verify(final WriteResult writeResult) { + } @Test public void testLogicalTypes() throws IOException, ParseException { final Schema schema = new Schema.Parser().parse(new File("src/test/resources/avro/logical-types.avsc")); - final WriteAvroResult writer = new WriteAvroResult(schema); + final WriteAvroResult writer = createWriter(schema); final List<RecordField> fields = new ArrayList<>(); fields.add(new RecordField("timeMillis", RecordFieldType.TIME.getDataType())); @@ -91,11 +95,7 @@ public class TestWriteAvroResult { } try (final InputStream in = new ByteArrayInputStream(data)) { - final DataFileStream<GenericRecord> dataFileStream = new DataFileStream<>(in, new GenericDatumReader<GenericRecord>()); - final Schema avroSchema = dataFileStream.getSchema(); - GenericData.setStringType(avroSchema, StringType.String); - - final GenericRecord avroRecord = dataFileStream.next(); + final GenericRecord avroRecord = readRecord(in, schema); final long secondsSinceMidnight = 33 + (20 * 60) + (14 * 60 * 60); final long millisSinceMidnight = secondsSinceMidnight * 1000L; @@ -110,9 +110,8 @@ public class TestWriteAvroResult { @Test public void testDataTypes() throws IOException { - // TODO: Test Enums final Schema schema = new Schema.Parser().parse(new File("src/test/resources/avro/datatypes.avsc")); - final WriteAvroResult writer = new WriteAvroResult(schema); + final WriteAvroResult writer = createWriter(schema); final List<RecordField> subRecordFields = Collections.singletonList(new RecordField("field1", RecordFieldType.STRING.getDataType())); final RecordSchema subRecordSchema = new SimpleRecordSchema(subRecordFields); @@ -124,7 +123,7 @@ public class TestWriteAvroResult { fields.add(new RecordField("double", RecordFieldType.DOUBLE.getDataType())); fields.add(new RecordField("float", RecordFieldType.FLOAT.getDataType())); fields.add(new RecordField("boolean", RecordFieldType.BOOLEAN.getDataType())); - fields.add(new RecordField("bytes", RecordFieldType.ARRAY.getChoiceDataType(Collections.singletonList(RecordFieldType.BYTE.getDataType())))); + fields.add(new RecordField("bytes", RecordFieldType.ARRAY.getArrayDataType(RecordFieldType.BYTE.getDataType()))); fields.add(new RecordField("nullOrLong", RecordFieldType.LONG.getDataType())); fields.add(new RecordField("array", RecordFieldType.ARRAY.getArrayDataType(RecordFieldType.INT.getDataType()))); fields.add(new RecordField("record", RecordFieldType.RECORD.getRecordDataType(subRecordSchema))); @@ -148,21 +147,18 @@ public class TestWriteAvroResult { final byte[] data; try (final ByteArrayOutputStream baos = new ByteArrayOutputStream()) { - writer.write(RecordSet.of(record.getSchema(), record), baos); + final WriteResult writeResult = writer.write(RecordSet.of(record.getSchema(), record), baos); + verify(writeResult); data = baos.toByteArray(); } try (final InputStream in = new ByteArrayInputStream(data)) { - final DataFileStream<GenericRecord> dataFileStream = new DataFileStream<>(in, new GenericDatumReader<GenericRecord>()); - final Schema avroSchema = dataFileStream.getSchema(); - GenericData.setStringType(avroSchema, StringType.String); - - final GenericRecord avroRecord = dataFileStream.next(); + final GenericRecord avroRecord = readRecord(in, schema); assertMatch(record, avroRecord); } } - private void assertMatch(final Record record, final GenericRecord avroRecord) { + protected void assertMatch(final Record record, final GenericRecord avroRecord) { for (final String fieldName : record.getSchema().getFieldNames()) { Object avroValue = avroRecord.get(fieldName); final Object recordValue = record.getValue(fieldName); http://git-wip-us.apache.org/repos/asf/nifi/blob/50ea1083/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/avro/TestWriteAvroResultWithSchema.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/avro/TestWriteAvroResultWithSchema.java b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/avro/TestWriteAvroResultWithSchema.java new file mode 100644 index 0000000..6ace012 --- /dev/null +++ b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/avro/TestWriteAvroResultWithSchema.java @@ -0,0 +1,46 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 + * + * http://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.apache.nifi.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.GenericData; +import org.apache.avro.generic.GenericData.StringType; +import org.apache.avro.generic.GenericDatumReader; +import org.apache.avro.generic.GenericRecord; + +public class TestWriteAvroResultWithSchema extends TestWriteAvroResult { + + @Override + protected WriteAvroResult createWriter(final Schema schema) { + return new WriteAvroResultWithSchema(schema); + } + + @Override + protected GenericRecord readRecord(final InputStream in, final Schema schema) throws IOException { + final DataFileStream<GenericRecord> dataFileStream = new DataFileStream<>(in, new GenericDatumReader<GenericRecord>()); + final Schema avroSchema = dataFileStream.getSchema(); + GenericData.setStringType(avroSchema, StringType.String); + final GenericRecord avroRecord = dataFileStream.next(); + + return avroRecord; + } +} http://git-wip-us.apache.org/repos/asf/nifi/blob/50ea1083/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/avro/TestWriteAvroResultWithoutSchema.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/avro/TestWriteAvroResultWithoutSchema.java b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/avro/TestWriteAvroResultWithoutSchema.java new file mode 100644 index 0000000..d40bb55 --- /dev/null +++ b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/avro/TestWriteAvroResultWithoutSchema.java @@ -0,0 +1,56 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 + * + * http://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.apache.nifi.avro; + +import java.io.IOException; +import java.io.InputStream; +import java.util.Map; + +import org.apache.avro.Schema; +import org.apache.avro.generic.GenericDatumReader; +import org.apache.avro.generic.GenericRecord; +import org.apache.avro.io.BinaryDecoder; +import org.apache.avro.io.DecoderFactory; +import org.apache.nifi.schema.access.SchemaTextAsAttribute; +import org.apache.nifi.serialization.WriteResult; +import org.junit.Assert; + +public class TestWriteAvroResultWithoutSchema extends TestWriteAvroResult { + + @Override + protected WriteAvroResult createWriter(final Schema schema) { + return new WriteAvroResultWithExternalSchema(schema, AvroTypeUtil.createSchema(schema), new SchemaTextAsAttribute()); + } + + @Override + protected GenericRecord readRecord(final InputStream in, final Schema schema) throws IOException { + final BinaryDecoder decoder = DecoderFactory.get().binaryDecoder(in, null); + final GenericDatumReader<GenericRecord> reader = new GenericDatumReader<GenericRecord>(schema); + return reader.read(null, decoder); + } + + @Override + protected void verify(final WriteResult writeResult) { + final Map<String, String> attributes = writeResult.getAttributes(); + + final String schemaText = attributes.get("avro.schema"); + Assert.assertNotNull(schemaText); + new Schema.Parser().parse(schemaText); + } + +} http://git-wip-us.apache.org/repos/asf/nifi/blob/50ea1083/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/csv/TestCSVHeaderSchemaStrategy.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/csv/TestCSVHeaderSchemaStrategy.java b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/csv/TestCSVHeaderSchemaStrategy.java new file mode 100644 index 0000000..3eed784 --- /dev/null +++ b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/csv/TestCSVHeaderSchemaStrategy.java @@ -0,0 +1,69 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 + * + * http://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.apache.nifi.csv; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.controller.ConfigurationContext; +import org.apache.nifi.schema.access.SchemaNotFoundException; +import org.apache.nifi.serialization.record.RecordFieldType; +import org.apache.nifi.serialization.record.RecordSchema; +import org.apache.nifi.util.MockConfigurationContext; +import org.junit.Test; + +public class TestCSVHeaderSchemaStrategy { + + @Test + public void testSimple() throws SchemaNotFoundException, IOException { + final CSVHeaderSchemaStrategy strategy = new CSVHeaderSchemaStrategy(); + final String headerLine = "a, b, c, d, e\\,z, f"; + final byte[] headerBytes = headerLine.getBytes(); + + final Map<PropertyDescriptor, String> properties = new HashMap<>(); + properties.put(CSVUtils.CSV_FORMAT, CSVUtils.CUSTOM.getValue()); + properties.put(CSVUtils.COMMENT_MARKER, "#"); + properties.put(CSVUtils.VALUE_SEPARATOR, ","); + properties.put(CSVUtils.TRIM_FIELDS, "true"); + properties.put(CSVUtils.QUOTE_CHAR, "\""); + properties.put(CSVUtils.ESCAPE_CHAR, "\\"); + + final ConfigurationContext context = new MockConfigurationContext(properties, null); + + final RecordSchema schema; + try (final InputStream bais = new ByteArrayInputStream(headerBytes)) { + schema = strategy.getSchema(null, bais, context); + } + + final List<String> expectedFieldNames = Arrays.asList("a", "b", "c", "d", "e,z", "f"); + assertEquals(expectedFieldNames, schema.getFieldNames()); + + assertTrue(schema.getFields().stream() + .allMatch(field -> field.getDataType().equals(RecordFieldType.STRING.getDataType()))); + } + +} http://git-wip-us.apache.org/repos/asf/nifi/blob/50ea1083/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/csv/TestWriteCSVResult.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/csv/TestWriteCSVResult.java b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/csv/TestWriteCSVResult.java index 1e8997b..9424e79 100644 --- a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/csv/TestWriteCSVResult.java +++ b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/csv/TestWriteCSVResult.java @@ -36,6 +36,7 @@ import java.util.TimeZone; import org.apache.commons.csv.CSVFormat; import org.apache.commons.csv.QuoteMode; +import org.apache.nifi.schema.access.SchemaNameAsAttribute; import org.apache.nifi.serialization.SimpleRecordSchema; import org.apache.nifi.serialization.record.DataType; import org.apache.nifi.serialization.record.MapRecord; @@ -52,7 +53,6 @@ public class TestWriteCSVResult { @Test public void testDataTypes() throws IOException { final CSVFormat csvFormat = CSVFormat.DEFAULT.withQuoteMode(QuoteMode.ALL).withRecordSeparator("\n"); - final WriteCSVResult result = new WriteCSVResult(csvFormat, RecordFieldType.DATE.getDefaultFormat(), RecordFieldType.TIME.getDefaultFormat(), RecordFieldType.TIMESTAMP.getDefaultFormat()); final StringBuilder headerBuilder = new StringBuilder(); final List<RecordField> fields = new ArrayList<>(); @@ -71,6 +71,9 @@ public class TestWriteCSVResult { } final RecordSchema schema = new SimpleRecordSchema(fields); + final WriteCSVResult result = new WriteCSVResult(csvFormat, schema, new SchemaNameAsAttribute(), + RecordFieldType.DATE.getDefaultFormat(), RecordFieldType.TIME.getDefaultFormat(), RecordFieldType.TIMESTAMP.getDefaultFormat(), true); + final long now = System.currentTimeMillis(); final Map<String, Object> valueMap = new HashMap<>(); valueMap.put("string", "string"); @@ -117,7 +120,7 @@ public class TestWriteCSVResult { expectedBuilder.append('"').append(dateValue).append('"').append(','); expectedBuilder.append('"').append(timeValue).append('"').append(','); expectedBuilder.append('"').append(timestampValue).append('"').append(','); - expectedBuilder.append(",\"48\","); + expectedBuilder.append(",\"48\",,"); final String expectedValues = expectedBuilder.toString(); assertEquals(expectedValues, values); http://git-wip-us.apache.org/repos/asf/nifi/blob/50ea1083/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/grok/TestGrokRecordReader.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/grok/TestGrokRecordReader.java b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/grok/TestGrokRecordReader.java index a741ad1..ae5d433 100644 --- a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/grok/TestGrokRecordReader.java +++ b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/grok/TestGrokRecordReader.java @@ -45,7 +45,7 @@ public class TestGrokRecordReader { grok.addPatternFromFile("src/main/resources/default-grok-patterns.txt"); grok.compile("%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:level} %{GREEDYDATA:message}"); - final GrokRecordReader deserializer = new GrokRecordReader(fis, grok, null); + final GrokRecordReader deserializer = new GrokRecordReader(fis, grok, GrokReader.createRecordSchema(grok), true); final String[] logLevels = new String[] {"INFO", "WARN", "ERROR", "FATAL", "FINE"}; final String[] messages = new String[] {"Test Message 1", "Red", "Green", "Blue", "Yellow"}; @@ -75,7 +75,7 @@ public class TestGrokRecordReader { final String msg = "2016-08-04 13:26:32,473 INFO [Leader Election Notification Thread-1] o.a.n.LoggerClass \n" + "org.apache.nifi.exception.UnitTestException: Testing to ensure we are able to capture stack traces"; final InputStream bais = new ByteArrayInputStream(msg.getBytes(StandardCharsets.UTF_8)); - final GrokRecordReader deserializer = new GrokRecordReader(bais, grok, null); + final GrokRecordReader deserializer = new GrokRecordReader(bais, grok, GrokReader.createRecordSchema(grok), true); final Object[] values = deserializer.nextRecord().getValues(); @@ -98,7 +98,7 @@ public class TestGrokRecordReader { grok.addPatternFromFile("src/main/resources/default-grok-patterns.txt"); grok.compile("%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:level} \\[%{DATA:thread}\\] %{DATA:class} %{GREEDYDATA:message}"); - final GrokRecordReader deserializer = new GrokRecordReader(fis, grok, null); + final GrokRecordReader deserializer = new GrokRecordReader(fis, grok, GrokReader.createRecordSchema(grok), true); final String[] logLevels = new String[] {"INFO", "INFO", "INFO", "WARN", "WARN"}; @@ -122,7 +122,7 @@ public class TestGrokRecordReader { grok.addPatternFromFile("src/main/resources/default-grok-patterns.txt"); grok.compile("%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:level} \\[%{DATA:thread}\\] %{DATA:class} %{GREEDYDATA:message}?"); - final GrokRecordReader deserializer = new GrokRecordReader(fis, grok, null); + final GrokRecordReader deserializer = new GrokRecordReader(fis, grok, GrokReader.createRecordSchema(grok), true); final String[] logLevels = new String[] {"INFO", "INFO", "ERROR", "WARN", "WARN"}; @@ -154,7 +154,7 @@ public class TestGrokRecordReader { grok.addPatternFromFile("src/main/resources/default-grok-patterns.txt"); grok.compile("%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:level} %{GREEDYDATA:message}"); - final GrokRecordReader deserializer = new GrokRecordReader(fis, grok, null); + final GrokRecordReader deserializer = new GrokRecordReader(fis, grok, GrokReader.createRecordSchema(grok), true); final String[] logLevels = new String[] {"INFO", "ERROR", "INFO"}; final String[] messages = new String[] {"message without stack trace", http://git-wip-us.apache.org/repos/asf/nifi/blob/50ea1083/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/json/TestJsonTreeRowRecordReader.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/json/TestJsonTreeRowRecordReader.java b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/json/TestJsonTreeRowRecordReader.java index 2422206..75e4d31 100644 --- a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/json/TestJsonTreeRowRecordReader.java +++ b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/json/TestJsonTreeRowRecordReader.java @@ -309,4 +309,27 @@ public class TestJsonTreeRowRecordReader { } } + @Test + public void testIncorrectSchema() throws IOException, MalformedRecordException { + final DataType accountType = RecordFieldType.RECORD.getRecordDataType(getAccountSchema()); + final List<RecordField> fields = getDefaultFields(); + fields.add(new RecordField("account", accountType)); + fields.remove(new RecordField("balance", RecordFieldType.DOUBLE.getDataType())); + + final RecordSchema schema = new SimpleRecordSchema(fields); + + try (final InputStream in = new FileInputStream(new File("src/test/resources/json/single-bank-account-wrong-field-type.json")); + final JsonTreeRowRecordReader reader = new JsonTreeRowRecordReader(in, Mockito.mock(ComponentLog.class), schema, dateFormat, timeFormat, timestampFormat)) { + + reader.nextRecord().getValues(); + Assert.fail("Was able to read record with invalid schema."); + + } catch (final MalformedRecordException mre) { + final String msg = mre.getCause().getMessage(); + assertTrue(msg.contains("account.balance")); + assertTrue(msg.contains("true")); + assertTrue(msg.contains("Double")); + assertTrue(msg.contains("Boolean")); + } + } } http://git-wip-us.apache.org/repos/asf/nifi/blob/50ea1083/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/json/TestWriteJsonResult.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/json/TestWriteJsonResult.java b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/json/TestWriteJsonResult.java index 6119d36..5c8bc49 100644 --- a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/json/TestWriteJsonResult.java +++ b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/json/TestWriteJsonResult.java @@ -37,6 +37,7 @@ import java.util.Map; import java.util.TimeZone; import org.apache.nifi.logging.ComponentLog; +import org.apache.nifi.schema.access.SchemaNameAsAttribute; import org.apache.nifi.serialization.SimpleRecordSchema; import org.apache.nifi.serialization.record.DataType; import org.apache.nifi.serialization.record.MapRecord; @@ -52,9 +53,6 @@ public class TestWriteJsonResult { @Test public void testDataTypes() throws IOException, ParseException { - final WriteJsonResult writer = new WriteJsonResult(Mockito.mock(ComponentLog.class), true, RecordFieldType.DATE.getDefaultFormat(), - RecordFieldType.TIME.getDefaultFormat(), RecordFieldType.TIMESTAMP.getDefaultFormat()); - final List<RecordField> fields = new ArrayList<>(); for (final RecordFieldType fieldType : RecordFieldType.values()) { if (fieldType == RecordFieldType.CHOICE) { @@ -63,16 +61,25 @@ public class TestWriteJsonResult { possibleTypes.add(RecordFieldType.LONG.getDataType()); fields.add(new RecordField(fieldType.name().toLowerCase(), fieldType.getChoiceDataType(possibleTypes))); + } else if (fieldType == RecordFieldType.MAP) { + fields.add(new RecordField(fieldType.name().toLowerCase(), fieldType.getMapDataType(RecordFieldType.INT.getDataType()))); } else { fields.add(new RecordField(fieldType.name().toLowerCase(), fieldType.getDataType())); } } final RecordSchema schema = new SimpleRecordSchema(fields); + final WriteJsonResult writer = new WriteJsonResult(Mockito.mock(ComponentLog.class), schema, new SchemaNameAsAttribute(), true, RecordFieldType.DATE.getDefaultFormat(), + RecordFieldType.TIME.getDefaultFormat(), RecordFieldType.TIMESTAMP.getDefaultFormat()); + final DateFormat df = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss.SSS"); df.setTimeZone(TimeZone.getTimeZone("gmt")); final long time = df.parse("2017/01/01 17:00:00.000").getTime(); + final Map<String, Object> map = new LinkedHashMap<>(); + map.put("height", 48); + map.put("width", 96); + final Map<String, Object> valueMap = new LinkedHashMap<>(); valueMap.put("string", "string"); valueMap.put("boolean", true); @@ -90,6 +97,7 @@ public class TestWriteJsonResult { valueMap.put("record", null); valueMap.put("array", null); valueMap.put("choice", 48L); + valueMap.put("map", map); final Record record = new MapRecord(schema, valueMap); final RecordSet rs = RecordSet.of(schema, record); http://git-wip-us.apache.org/repos/asf/nifi/blob/50ea1083/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/resources/json/output/dataTypes.json ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/resources/json/output/dataTypes.json b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/resources/json/output/dataTypes.json index 40c28dd..881925c 100644 --- a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/resources/json/output/dataTypes.json +++ b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/resources/json/output/dataTypes.json @@ -14,5 +14,9 @@ "timestamp" : "2017-01-01 17:00:00", "record" : null, "choice" : 48, - "array" : null + "array" : null, + "map" : { + "height" : 48, + "width" : 96 + } } ] \ No newline at end of file http://git-wip-us.apache.org/repos/asf/nifi/blob/50ea1083/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/resources/json/single-bank-account-wrong-field-type.json ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/resources/json/single-bank-account-wrong-field-type.json b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/resources/json/single-bank-account-wrong-field-type.json new file mode 100644 index 0000000..50d676c --- /dev/null +++ b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/resources/json/single-bank-account-wrong-field-type.json @@ -0,0 +1,13 @@ +{ + "id": 1, + "name": "John Doe", + "address": "123 My Street", + "city": "My City", + "state": "MS", + "zipCode": "11111", + "country": "USA", + "account": { + "id": 42, + "balance": true + } +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/nifi/blob/50ea1083/nifi-nar-bundles/nifi-standard-services/nifi-schema-registry-service-api/src/main/java/org/apache/nifi/schemaregistry/services/SchemaRegistry.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-standard-services/nifi-schema-registry-service-api/src/main/java/org/apache/nifi/schemaregistry/services/SchemaRegistry.java b/nifi-nar-bundles/nifi-standard-services/nifi-schema-registry-service-api/src/main/java/org/apache/nifi/schemaregistry/services/SchemaRegistry.java index 68c2461..88362b8 100644 --- a/nifi-nar-bundles/nifi-standard-services/nifi-schema-registry-service-api/src/main/java/org/apache/nifi/schemaregistry/services/SchemaRegistry.java +++ b/nifi-nar-bundles/nifi-standard-services/nifi-schema-registry-service-api/src/main/java/org/apache/nifi/schemaregistry/services/SchemaRegistry.java @@ -16,38 +16,76 @@ */ package org.apache.nifi.schemaregistry.services; -import java.util.Map; +import java.io.IOException; +import java.util.Set; import org.apache.nifi.controller.ControllerService; +import org.apache.nifi.schema.access.SchemaField; +import org.apache.nifi.schema.access.SchemaNotFoundException; import org.apache.nifi.serialization.record.RecordSchema; +import org.apache.nifi.serialization.record.SchemaIdentifier; /** * Represents {@link ControllerService} strategy to expose internal and/or * integrate with external Schema Registry */ -public interface SchemaRegistry extends ControllerService, AutoCloseable { - - public static final String SCHEMA_NAME_ATTR = "schema.name"; +public interface SchemaRegistry extends ControllerService { /** * Retrieves and returns the textual representation of the schema based on - * the provided name of the schema available in Schema Registry. Will throw - * an runtime exception if schema can not be found. + * the provided name of the schema available in Schema Registry. + * + * @return the text that corresponds to the latest version of the schema with the given name + * + * @throws IOException if unable to communicate with the backing store + * @throws SchemaNotFoundException if unable to find the schema with the given name */ - String retrieveSchemaText(String schemaName); + String retrieveSchemaText(String schemaName) throws IOException, SchemaNotFoundException; /** - * Retrieves and returns the textual representation of the schema based on - * the provided name of the schema available in Schema Registry and optional - * additional attributes. Will throw an runtime exception if schema can not - * be found. + * Retrieves the textual representation of the schema with the given ID and version + * + * @param schemaId the unique identifier for the desired schema + * @param version the version of the desired schema + * @return the textual representation of the schema with the given ID and version + * + * @throws IOException if unable to communicate with the backing store + * @throws SchemaNotFoundException if unable to find the schema with the given id and version */ - String retrieveSchemaText(String schemaName, Map<String, String> attributes); + String retrieveSchemaText(long schemaId, int version) throws IOException, SchemaNotFoundException; + /** + * Retrieves and returns the RecordSchema based on the provided name of the schema available in Schema Registry. The RecordSchema + * that is returned must have the Schema's name populated in its SchemaIdentifier. I.e., a call to + * {@link RecordSchema}.{@link RecordSchema#getIdentifier() getIdentifier()}.{@link SchemaIdentifier#getName() getName()} + * will always return an {@link Optional} that is not empty. + * + * @return the latest version of the schema with the given name, or <code>null</code> if no schema can be found with the given name. + * @throws SchemaNotFoundException if unable to find the schema with the given name + */ + RecordSchema retrieveSchema(String schemaName) throws IOException, SchemaNotFoundException; - RecordSchema retrieveSchema(String schemaName); + /** + * Retrieves the schema with the given ID and version. The RecordSchema that is returned must have the Schema's identifier and version + * populated in its SchemaIdentifier. I.e., a call to + * {@link RecordSchema}.{@link RecordSchema#getIdentifier() getIdentifier()}.{@link SchemaIdentifier#getIdentifier() getIdentifier()} + * will always return an {@link Optional} that is not empty, as will a call to + * {@link RecordSchema}.{@link RecordSchema#getIdentifier() getIdentifier()}.{@link SchemaIdentifier#getVersion() getVersion()}. + * + * @param schemaId the unique identifier for the desired schema + * @param version the version of the desired schema + * @return the schema with the given ID and version or <code>null</code> if no schema + * can be found with the given ID and version + * + * @throws IOException if unable to communicate with the backing store + * @throws SchemaNotFoundException if unable to find the schema with the given id and version + */ + RecordSchema retrieveSchema(long schemaId, int version) throws IOException, SchemaNotFoundException; - RecordSchema retrieveSchema(String schemaName, Map<String, String> attributes); + /** + * @return the set of all Schema Fields that are supplied by the RecordSchema that is returned from {@link #retrieveSchema(String)} and {@link #retrieveSchema(long, int)} + */ + Set<SchemaField> getSuppliedSchemaFields(); }
