[GOBBLIN-238] Implement EnvelopePayloadExtractor and EnvelopePayloadDeserializer

• Type: Task
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Fix Version/s: 0.12.0
• Labels: Core:Converter
• Sprint: Apache Gobblin 170807, Apache Gobblin 170821, Apache Gobblin 170905
• Assignee: Zhixiong Chen
• Reporter: Zhixiong Chen
• Created: 01/Sep/17 22:37
• Updated: 03/Jan/18 06:45
• Resolved: 12/Sep/17 01:15

Description

Why

The current implementation of EnvelopeSchemaConverter has several flaws:

• Assumes top level payload schema field
• Output record is the schema'ed payload but output schema is a String

To address the issues and improve envelope schema conversion, the task implements two types of EnvelopeSchemaConverter: EnvelopePayloadExtractor and EnvelopePayloadDeserializer.

EnvelopePayloadExtractor

This is a replacement of the deprecated `EvenlopeSchemaConverter`. Given an envelope record, the output schema will be the latest payload schema fetched from a kafka registry. The output record will be the deserialized payload with the latest schema

EnvelopePayloadDeserializer

Given an envelope record, the output schema will set the payload field to have the latest schema fetched from a kafka registry and set the other fields as they are from the input schema. The output record will set the payload to be the deserialized object with the latest schema and set the other fields as they are from the input record

Configurations

One configuration is required to set for any of the converters to work. It has no default value.

// The topic to fetch the latest schema of the payload from a kafka registry
converter.envelopeSchemaConverter.payloadSchemaTopic=

The converter supports nested schema id

converter.envelopeSchemaConverter.schemaIdField="metadata.payloadSchemaId"

Issue Links

• is a clone of: GOBBLIN-87 Gobblin runOnce not working correctly