StreamlineEventSerializer implements Serializer<StreamlineEvent> { static Object getAvroRecord (StreamlineEvent streamlineEvent, Schema schema) { if (streamlineEvent.containsKey(StreamlineEvent.PRIMITIVE_PAYLOAD_FIELD)) { if (streamlineEvent.keySet().size() > 1) { throw new RuntimeException("Invalid schema, primitive schema can contain only one field."); } return streamlineEvent.get(StreamlineEvent.PRIMITIVE_PAYLOAD_FIELD); } GenericRecord result; result = new GenericData.Record(schema); for (Map.Entry<String, Object> entry: streamlineEvent.entrySet()) { result.put(entry.getKey(), getAvroValue(entry.getValue(), schema.getField(entry.getKey()).schema())); } return result; } StreamlineEventSerializer(); @Override void configure(Map<String, ?> configs, boolean isKey); @Override byte[] serialize(String topic, StreamlineEvent streamlineEvent); @Override void close();  }
@Test public void testRecordWithFixed () { List<Schema.Field> fields = new ArrayList<>(); Map<String, Object> data = new HashMap<>(); for (int i = 0; i < PRIMITIVE_VALUES.length; ++i) { Schema.Field field = new Schema.Field(NAMES[i], Schema.create(SCHEMA_TYPES[i]), null, null); fields.add(field); data.put(NAMES[i], PRIMITIVE_VALUES[i]); } fields.add(new Schema.Field("fixed", Schema.createFixed("fixedSchema", null, null, 10), null, null)); data.put("fixed", "bytes".getBytes()); fields.add(new Schema.Field("array", Schema.createArray(Schema.create(Schema.Type.INT)), null, null)); List<Integer> integerList = new ArrayList<>(); integerList.add(1); integerList.add(2); data.put("array", integerList); Schema schema = Schema.createRecord(fields); GenericRecord expected = new GenericData.Record(schema); for (int i = 0; i < PRIMITIVE_VALUES.length; ++i) { expected.put(NAMES[i], PRIMITIVE_VALUES[i]); } expected.put("fixed", new GenericData.Fixed(Schema.createFixed("fixedSchema", null, null, 10), "bytes".getBytes())); expected.put("array", new GenericData.Array<Integer>(Schema.createArray(Schema.create(Schema.Type.INT)), integerList)); StreamlineEvent streamlineEvent = StreamlineEventImpl.builder().fieldsAndValues(data).dataSourceId("dataSourceId").build(); Assert.assertEquals(expected, StreamlineEventSerializer.getAvroRecord(streamlineEvent, schema)); }