@Test(expectedExceptions = DataConversionException.class, expectedExceptionsMessageRegExp = "Input data does not have record.") public void testFailureWithoutRecord() throws DataConversionException, IOException { ObjectMapper objectMapper = new ObjectMapper(); HashMap<String, Object> map = new HashMap<>(); Metadata md = new Metadata(); md.getRecordMetadata().put("test1", "test2"); map.put("rMd", md); JsonNode jsonNode = objectMapper.valueToTree(map); RecordWithMetadata<byte[]> inputRecord = new RecordWithMetadata<>(jsonNode.toString().getBytes(), null); EnvelopedRecordWithMetadataToRecordWithMetadata converter = new EnvelopedRecordWithMetadataToRecordWithMetadata(); converter.convertRecord(null, inputRecord, null); }
@Override public Iterable<RecordWithMetadata<?>> convertRecord(Object outputSchema, RecordWithMetadata<byte[]> inputRecord, WorkUnitState workUnit) throws DataConversionException { try { try (JsonParser parser = jsonFactory.createJsonParser(inputRecord.getRecord())) { parser.setCodec(objectMapper); JsonNode jsonNode = parser.readValueAsTree(); if (!jsonNode.has(RECORD_KEY)) { throw new DataConversionException("Input data does not have record."); } String record = jsonNode.get(RECORD_KEY).getTextValue(); Metadata md = new Metadata(); if (jsonNode.has(METADATA_KEY) && jsonNode.get(METADATA_KEY).has(METADATA_RECORD_KEY)) { md.getRecordMetadata().putAll(objectMapper.readValue(jsonNode.get(METADATA_KEY).get(METADATA_RECORD_KEY), Map.class)); } return Collections.singleton(new RecordWithMetadata<>(record, md)); } } catch (IOException e) { throw new DataConversionException(e); } }
EnvelopedRecordWithMetadataToRecordWithMetadata extends Converter<String, Object, RecordWithMetadata<byte[]>, RecordWithMetadata<?>> { @Override public Iterable<RecordWithMetadata<?>> convertRecord(Object outputSchema, RecordWithMetadata<byte[]> inputRecord, WorkUnitState workUnit) throws DataConversionException { try { try (JsonParser parser = jsonFactory.createJsonParser(inputRecord.getRecord())) { parser.setCodec(objectMapper); JsonNode jsonNode = parser.readValueAsTree(); if (!jsonNode.has(RECORD_KEY)) { throw new DataConversionException("Input data does not have record."); } String record = jsonNode.get(RECORD_KEY).getTextValue(); Metadata md = new Metadata(); if (jsonNode.has(METADATA_KEY) && jsonNode.get(METADATA_KEY).has(METADATA_RECORD_KEY)) { md.getRecordMetadata().putAll(objectMapper.readValue(jsonNode.get(METADATA_KEY).get(METADATA_RECORD_KEY), Map.class)); } return Collections.singleton(new RecordWithMetadata<>(record, md)); } } catch (IOException e) { throw new DataConversionException(e); } } }
EnvelopedRecordWithMetadataToRecordWithMetadata extends Converter<String, Object, RecordWithMetadata<byte[]>, RecordWithMetadata<?>> { @Override public Iterable<RecordWithMetadata<?>> convertRecord(Object outputSchema, RecordWithMetadata<byte[]> inputRecord, WorkUnitState workUnit) throws DataConversionException { try { try (JsonParser parser = jsonFactory.createJsonParser(inputRecord.getRecord())) { parser.setCodec(objectMapper); JsonNode jsonNode = parser.readValueAsTree(); if (!jsonNode.has(RECORD_KEY)) { throw new DataConversionException("Input data does not have record."); } String record = jsonNode.get(RECORD_KEY).getTextValue(); Metadata md = new Metadata(); if (jsonNode.has(METADATA_KEY) && jsonNode.get(METADATA_KEY).has(METADATA_RECORD_KEY)) { md.getRecordMetadata().putAll(objectMapper.readValue(jsonNode.get(METADATA_KEY).get(METADATA_RECORD_KEY), Map.class)); } return Collections.singleton(new RecordWithMetadata<>(record, md)); } } catch (IOException e) { throw new DataConversionException(e); } }  }
EnvelopedRecordWithMetadataToRecordWithMetadata extends Converter<String, Object, RecordWithMetadata<byte[]>, RecordWithMetadata<?>> { @Override public Iterable<RecordWithMetadata<?>> convertRecord(Object outputSchema, RecordWithMetadata<byte[]> inputRecord, WorkUnitState workUnit) throws DataConversionException { try { try (JsonParser parser = jsonFactory.createJsonParser(inputRecord.getRecord())) { parser.setCodec(objectMapper); JsonNode jsonNode = parser.readValueAsTree(); if (!jsonNode.has(RECORD_KEY)) { throw new DataConversionException("Input data does not have record."); } String record = jsonNode.get(RECORD_KEY).getTextValue(); Metadata md = new Metadata(); if (jsonNode.has(METADATA_KEY) && jsonNode.get(METADATA_KEY).has(METADATA_RECORD_KEY)) { md.getRecordMetadata().putAll(objectMapper.readValue(jsonNode.get(METADATA_KEY).get(METADATA_RECORD_KEY), Map.class)); } return Collections.singleton(new RecordWithMetadata<>(record, md)); } } catch (IOException e) { throw new DataConversionException(e); } }  @Override String convertSchema(String inputSchema, WorkUnitState workUnit); @Override Iterable<RecordWithMetadata<?>> convertRecord(Object outputSchema, RecordWithMetadata<byte[]> inputRecord,
      WorkUnitState workUnit); }
EnvelopedRecordWithMetadataToRecordWithMetadata extends Converter<String, Object, RecordWithMetadata<byte[]>, RecordWithMetadata<?>> { @Override public Iterable<RecordWithMetadata<?>> convertRecord(Object outputSchema, RecordWithMetadata<byte[]> inputRecord, WorkUnitState workUnit) throws DataConversionException { try { try (JsonParser parser = jsonFactory.createJsonParser(inputRecord.getRecord())) { parser.setCodec(objectMapper); JsonNode jsonNode = parser.readValueAsTree(); if (!jsonNode.has(RECORD_KEY)) { throw new DataConversionException("Input data does not have record."); } String record = jsonNode.get(RECORD_KEY).getTextValue(); Metadata md = new Metadata(); if (jsonNode.has(METADATA_KEY) && jsonNode.get(METADATA_KEY).has(METADATA_RECORD_KEY)) { md.getRecordMetadata().putAll(objectMapper.readValue(jsonNode.get(METADATA_KEY).get(METADATA_RECORD_KEY), Map.class)); } return Collections.singleton(new RecordWithMetadata<>(record, md)); } } catch (IOException e) { throw new DataConversionException(e); } }  @Override String convertSchema(String inputSchema, WorkUnitState workUnit); @Override Iterable<RecordWithMetadata<?>> convertRecord(Object outputSchema, RecordWithMetadata<byte[]> inputRecord,
      WorkUnitState workUnit);  }