@Test public void convertRecordTest() throws Exception { when(mockInputRecord.get("payloadSchemaId")).thenReturn(SCHEMA_KEY); when(mockOutputRecord.getSchema()).thenReturn(mockSchema); WorkUnitState workUnitState = new WorkUnitState(); workUnitState.setProp(ConfigurationKeys.EXTRACT_TABLE_NAME_KEY, "testEvent"); workUnitState.setProp("kafka.schema.registry.url", "testUrl"); workUnitState.setProp(KafkaSchemaRegistryFactory.KAFKA_SCHEMA_REGISTRY_FACTORY_CLASS, KafkaAvroSchemaRegistryForTest.Factory.class.getName()); EnvelopeSchemaConverterForTest converter = new EnvelopeSchemaConverterForTest(); converter.init(workUnitState); GenericRecord output = converter.convertRecord(null, mockInputRecord, workUnitState).iterator().next(); Assert.assertEquals(output, mockOutputRecord); }
@Override public Iterable<GenericRecord> convertRecord(String outputSchema, GenericRecord inputRecord, WorkUnitState workUnit) throws DataConversionException { try { String schemaIdField = workUnit.contains(PAYLOAD_SCHEMA_ID_FIELD) ? workUnit.getProp(PAYLOAD_SCHEMA_ID_FIELD) : DEFAULT_PAYLOAD_SCHEMA_ID_FIELD; String payloadField = workUnit.contains(PAYLOAD_FIELD) ? workUnit.getProp(PAYLOAD_FIELD) : DEFAULT_PAYLOAD_FIELD; String schemaKey = String.valueOf(inputRecord.get(schemaIdField)); Schema payloadSchema = (Schema) this.registry.getSchemaByKey(schemaKey); byte[] payload = getPayload(inputRecord, payloadField); GenericRecord outputRecord = deserializePayload(payload, payloadSchema); if (this.fieldRemover.isPresent()) { payloadSchema = this.fieldRemover.get().removeFields(payloadSchema); } return new SingleRecordIterable<>(AvroUtils.convertRecordSchema(outputRecord, payloadSchema)); } catch (IOException | ExecutionException e) { throw new DataConversionException(e); } }
EnvelopeSchemaConverter extends Converter<Schema, String, GenericRecord, GenericRecord> { @Override public Iterable<GenericRecord> convertRecord(String outputSchema, GenericRecord inputRecord, WorkUnitState workUnit) throws DataConversionException { try { String schemaIdField = workUnit.contains(PAYLOAD_SCHEMA_ID_FIELD) ? workUnit.getProp(PAYLOAD_SCHEMA_ID_FIELD) : DEFAULT_PAYLOAD_SCHEMA_ID_FIELD; String payloadField = workUnit.contains(PAYLOAD_FIELD) ? workUnit.getProp(PAYLOAD_FIELD) : DEFAULT_PAYLOAD_FIELD; String schemaKey = String.valueOf(inputRecord.get(schemaIdField)); Schema payloadSchema = (Schema) this.registry.getSchemaByKey(schemaKey); byte[] payload = getPayload(inputRecord, payloadField); GenericRecord outputRecord = deserializePayload(payload, payloadSchema); if (this.fieldRemover.isPresent()) { payloadSchema = this.fieldRemover.get().removeFields(payloadSchema); } return new SingleRecordIterable<>(AvroUtils.convertRecordSchema(outputRecord, payloadSchema)); } catch (IOException | ExecutionException e) { throw new DataConversionException(e); } } }
EnvelopeSchemaConverter extends Converter<Schema, String, GenericRecord, GenericRecord> { @Override public Iterable<GenericRecord> convertRecord(String outputSchema, GenericRecord inputRecord, WorkUnitState workUnit) throws DataConversionException { try { String schemaIdField = workUnit.contains(PAYLOAD_SCHEMA_ID_FIELD) ? workUnit.getProp(PAYLOAD_SCHEMA_ID_FIELD) : DEFAULT_PAYLOAD_SCHEMA_ID_FIELD; String payloadField = workUnit.contains(PAYLOAD_FIELD) ? workUnit.getProp(PAYLOAD_FIELD) : DEFAULT_PAYLOAD_FIELD; String schemaKey = String.valueOf(inputRecord.get(schemaIdField)); Schema payloadSchema = (Schema) this.registry.getSchemaByKey(schemaKey); byte[] payload = getPayload(inputRecord, payloadField); GenericRecord outputRecord = deserializePayload(payload, payloadSchema); if (this.fieldRemover.isPresent()) { payloadSchema = this.fieldRemover.get().removeFields(payloadSchema); } return new SingleRecordIterable<>(AvroUtils.convertRecordSchema(outputRecord, payloadSchema)); } catch (IOException | ExecutionException e) { throw new DataConversionException(e); } }  }
EnvelopeSchemaConverter extends Converter<Schema, String, GenericRecord, GenericRecord> { @Override public Iterable<GenericRecord> convertRecord(String outputSchema, GenericRecord inputRecord, WorkUnitState workUnit) throws DataConversionException { try { String schemaIdField = workUnit.contains(PAYLOAD_SCHEMA_ID_FIELD) ? workUnit.getProp(PAYLOAD_SCHEMA_ID_FIELD) : DEFAULT_PAYLOAD_SCHEMA_ID_FIELD; String payloadField = workUnit.contains(PAYLOAD_FIELD) ? workUnit.getProp(PAYLOAD_FIELD) : DEFAULT_PAYLOAD_FIELD; String schemaKey = String.valueOf(inputRecord.get(schemaIdField)); Schema payloadSchema = (Schema) this.registry.getSchemaByKey(schemaKey); byte[] payload = getPayload(inputRecord, payloadField); GenericRecord outputRecord = deserializePayload(payload, payloadSchema); if (this.fieldRemover.isPresent()) { payloadSchema = this.fieldRemover.get().removeFields(payloadSchema); } return new SingleRecordIterable<>(AvroUtils.convertRecordSchema(outputRecord, payloadSchema)); } catch (IOException | ExecutionException e) { throw new DataConversionException(e); } }  @Override EnvelopeSchemaConverter init(WorkUnitState workUnit); @Override String convertSchema(Schema inputSchema, WorkUnitState workUnit); @Override Iterable<GenericRecord> convertRecord(String outputSchema, GenericRecord inputRecord, WorkUnitState workUnit); byte[] getPayload(GenericRecord inputRecord, String payloadFieldName); GenericRecord deserializePayload(byte[] payload, Schema payloadSchema); }
EnvelopeSchemaConverter extends Converter<Schema, String, GenericRecord, GenericRecord> { @Override public Iterable<GenericRecord> convertRecord(String outputSchema, GenericRecord inputRecord, WorkUnitState workUnit) throws DataConversionException { try { String schemaIdField = workUnit.contains(PAYLOAD_SCHEMA_ID_FIELD) ? workUnit.getProp(PAYLOAD_SCHEMA_ID_FIELD) : DEFAULT_PAYLOAD_SCHEMA_ID_FIELD; String payloadField = workUnit.contains(PAYLOAD_FIELD) ? workUnit.getProp(PAYLOAD_FIELD) : DEFAULT_PAYLOAD_FIELD; String schemaKey = String.valueOf(inputRecord.get(schemaIdField)); Schema payloadSchema = (Schema) this.registry.getSchemaByKey(schemaKey); byte[] payload = getPayload(inputRecord, payloadField); GenericRecord outputRecord = deserializePayload(payload, payloadSchema); if (this.fieldRemover.isPresent()) { payloadSchema = this.fieldRemover.get().removeFields(payloadSchema); } return new SingleRecordIterable<>(AvroUtils.convertRecordSchema(outputRecord, payloadSchema)); } catch (IOException | ExecutionException e) { throw new DataConversionException(e); } }  @Override EnvelopeSchemaConverter init(WorkUnitState workUnit); @Override String convertSchema(Schema inputSchema, WorkUnitState workUnit); @Override Iterable<GenericRecord> convertRecord(String outputSchema, GenericRecord inputRecord, WorkUnitState workUnit); byte[] getPayload(GenericRecord inputRecord, String payloadFieldName); GenericRecord deserializePayload(byte[] payload, Schema payloadSchema); static final String PAYLOAD_SCHEMA_ID_FIELD; static final String PAYLOAD_FIELD; static final String DEFAULT_PAYLOAD_SCHEMA_ID_FIELD; static final String DEFAULT_PAYLOAD_FIELD; static final String DEFAULT_KAFKA_SCHEMA_REGISTRY_FACTORY_CLASS; }