@Test public void testFormatCorrectsFieldNames() { byte[] data = Json.createObjectNode() .put("test", true) .put("test-hyphen", true) .put("test.dot", true) .put("-test-prefix-hyphen", true) .put("$test-bad-symbol", true) .put("0-test-prefix-number", true) .toString().getBytes(StandardCharsets.UTF_8); PubsubMessage message = new PubsubMessage(data, Collections.emptyMap()); Schema schema = SchemaBuilder.record("root").fields() .name("test").type().booleanType().noDefault() .name("test_hyphen").type().booleanType().noDefault() .name("test_dot").type().booleanType().noDefault() .name("_test_prefix_hyphen").type().booleanType().noDefault() .name("_0_test_prefix_number").type().booleanType().noDefault() .endRecord(); PubsubMessageRecordFormatter formatter = new PubsubMessageRecordFormatter(); GenericRecord record = formatter.formatRecord(message, schema); assertEquals(true, record.get("test")); assertEquals(true, record.get("test_hyphen")); assertEquals(true, record.get("test_dot")); assertEquals(true, record.get("_test_prefix_hyphen")); assertEquals(true, record.get("_0_test_prefix_number")); assertEquals(null, record.get("$test_bad_symbol")); }
@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);  }