@Test public void testFormatFlatSchema() { byte[] data = Json.createObjectNode() .put("test_bool", true).put("test_long", -7).put("test_double", 0.99) .put("test_string", "hello world") .toString().getBytes(StandardCharsets.UTF_8); PubsubMessage message = new PubsubMessage(data, Collections.emptyMap()); Schema schema = SchemaBuilder.record("root").fields() .name("test_bool").type().booleanType().noDefault() .name("test_long").type().longType().noDefault() .name("test_double").type().doubleType().noDefault() .name("test_string").type().stringType().noDefault() .endRecord(); PubsubMessageRecordFormatter formatter = new PubsubMessageRecordFormatter(); GenericRecord record = formatter.formatRecord(message, schema); assertEquals(true, record.get("test_bool")); assertEquals(-7L, record.get("test_long")); assertEquals(0.99, record.get("test_double")); assertEquals("hello world", record.get("test_string").toString()); }
@Override public GenericRecord formatRecord(PubsubMessage element, Schema schema) { GenericRecord result = null; try { InputStream in = new ByteArrayInputStream(element.getPayload()); DatumReader<GenericRecord> reader = new GenericDatumReader<>(schema); Decoder decoder = new GuidedJsonDecoder(schema, in); result = reader.read(null, decoder); } catch (IOException e) { throw new UncheckedIOException(e); } catch (AvroTypeException e) { throw e; } return result; }
PubsubMessageRecordFormatter implements RecordFormatter<PubsubMessage> { @Override public GenericRecord formatRecord(PubsubMessage element, Schema schema) { GenericRecord result = null; try { InputStream in = new ByteArrayInputStream(element.getPayload()); DatumReader<GenericRecord> reader = new GenericDatumReader<>(schema); Decoder decoder = new GuidedJsonDecoder(schema, in); result = reader.read(null, decoder); } catch (IOException e) { throw new UncheckedIOException(e); } catch (AvroTypeException e) { throw e; } return result; } }
PubsubMessageRecordFormatter implements RecordFormatter<PubsubMessage> { @Override public GenericRecord formatRecord(PubsubMessage element, Schema schema) { GenericRecord result = null; try { InputStream in = new ByteArrayInputStream(element.getPayload()); DatumReader<GenericRecord> reader = new GenericDatumReader<>(schema); Decoder decoder = new GuidedJsonDecoder(schema, in); result = reader.read(null, decoder); } catch (IOException e) { throw new UncheckedIOException(e); } catch (AvroTypeException e) { throw e; } return result; }  }
PubsubMessageRecordFormatter implements RecordFormatter<PubsubMessage> { @Override public GenericRecord formatRecord(PubsubMessage element, Schema schema) { GenericRecord result = null; try { InputStream in = new ByteArrayInputStream(element.getPayload()); DatumReader<GenericRecord> reader = new GenericDatumReader<>(schema); Decoder decoder = new GuidedJsonDecoder(schema, in); result = reader.read(null, decoder); } catch (IOException e) { throw new UncheckedIOException(e); } catch (AvroTypeException e) { throw e; } return result; }  @Override GenericRecord formatRecord(PubsubMessage element, Schema schema); }
PubsubMessageRecordFormatter implements RecordFormatter<PubsubMessage> { @Override public GenericRecord formatRecord(PubsubMessage element, Schema schema) { GenericRecord result = null; try { InputStream in = new ByteArrayInputStream(element.getPayload()); DatumReader<GenericRecord> reader = new GenericDatumReader<>(schema); Decoder decoder = new GuidedJsonDecoder(schema, in); result = reader.read(null, decoder); } catch (IOException e) { throw new UncheckedIOException(e); } catch (AvroTypeException e) { throw e; } return result; }  @Override GenericRecord formatRecord(PubsubMessage element, Schema schema);  }