commonlib/src/main/java/com/streamsets/pipeline/lib/util/ProtobufTypeUtil.java
public static Field protobufToSdcField(
      Record record,
      String fieldPath,
      Descriptors.Descriptor descriptor,
      Map<String, Set<Descriptors.FieldDescriptor>> messageTypeToExtensionMap,
      Object message
  ) throws DataParserException {
    LinkedHashMap<String, Field> sdcRecordMapFieldValue = new LinkedHashMap<>();

    // get all the expected fields from the proto file
    Map<String, Descriptors.FieldDescriptor> protobufFields = new LinkedHashMap<>();
    for (Descriptors.FieldDescriptor fieldDescriptor : descriptor.getFields()) {
      protobufFields.put(fieldDescriptor.getName(), fieldDescriptor);
    }

    // get all fields in the read message
    Map<Descriptors.FieldDescriptor, Object> values = ((DynamicMessage) message).getAllFields();

    // for every field present in the proto definition create an sdc field.
    for (Descriptors.FieldDescriptor fieldDescriptor : protobufFields.values()) {
      Object value = values.get(fieldDescriptor);
      sdcRecordMapFieldValue.put(
          fieldDescriptor.getName(),
          createField(record, fieldPath, fieldDescriptor, messageTypeToExtensionMap, value)
      );
    }

    // handle applicable extensions for this message type
    if (messageTypeToExtensionMap.containsKey(descriptor.getFullName())) {
      for (Descriptors.FieldDescriptor fieldDescriptor : messageTypeToExtensionMap.get(descriptor.getFullName())) {
        if (values.containsKey(fieldDescriptor)) {
          Object value = values.get(fieldDescriptor);
          sdcRecordMapFieldValue.put(
              fieldDescriptor.getName(),
              createField(record, fieldPath, fieldDescriptor, messageTypeToExtensionMap, value)
          );
        }
      }
    }

    // handle unknown fields
    // unknown fields can go into the record header
    UnknownFieldSet unknownFields = ((DynamicMessage) message).getUnknownFields();
    if (!unknownFields.asMap().isEmpty()) {
      ByteArrayOutputStream bOut = new ByteArrayOutputStream();
      try {
        unknownFields.writeDelimitedTo(bOut);
        bOut.flush();
        bOut.close();
      } catch (IOException e) {
        throw new DataParserException(Errors.PROTOBUF_10, e.toString(), e);
      }
      String path = fieldPath.isEmpty() ? FORWARD_SLASH : fieldPath;
      byte[] bytes = org.apache.commons.codec.binary.Base64.encodeBase64(bOut.toByteArray());
      record.getHeader().setAttribute(PROTOBUF_UNKNOWN_FIELDS_PREFIX + path, new String(bytes, StandardCharsets.UTF_8));
    }

    return Field.createListMap(sdcRecordMapFieldValue);
  }
public static Field protobufToSdcField(
      Record record,
      String fieldPath,
      Descriptors.Descriptor descriptor,
      Map<String, Set<Descriptors.FieldDescriptor>> messageTypeToExtensionMap,
      Object message
  ) throws DataParserException {
    LinkedHashMap<String, Field> sdcRecordMapFieldValue = new LinkedHashMap<>();

    // get all the expected fields from the proto file
    Map<String, Descriptors.FieldDescriptor> protobufFields = new LinkedHashMap<>();
    for (Descriptors.FieldDescriptor fieldDescriptor : descriptor.getFields()) {
      protobufFields.put(fieldDescriptor.getName(), fieldDescriptor);
    }

    // get all fields in the read message
    Map<Descriptors.FieldDescriptor, Object> values = ((DynamicMessage) message).getAllFields();

    // for every field present in the proto definition create an sdc field.
    for (Descriptors.FieldDescriptor fieldDescriptor : protobufFields.values()) {
      Object value = values.get(fieldDescriptor);
      sdcRecordMapFieldValue.put(
          fieldDescriptor.getName(),
          createField(record, fieldPath, fieldDescriptor, messageTypeToExtensionMap, value)
      );
    }

    // handle applicable extensions for this message type
    if (messageTypeToExtensionMap.containsKey(descriptor.getFullName())) {
      for (Descriptors.FieldDescriptor fieldDescriptor : messageTypeToExtensionMap.get(descriptor.getFullName())) {
        if (values.containsKey(fieldDescriptor)) {
          Object value = values.get(fieldDescriptor);
          sdcRecordMapFieldValue.put(
              fieldDescriptor.getName(),
              createField(record, fieldPath, fieldDescriptor, messageTypeToExtensionMap, value)
          );
        }
      }
    }

    // handle unknown fields
    // unknown fields can go into the record header
    UnknownFieldSet unknownFields = ((DynamicMessage) message).getUnknownFields();
    if (!unknownFields.asMap().isEmpty()) {
      ByteArrayOutputStream bOut = new ByteArrayOutputStream();
      try {
        unknownFields.writeDelimitedTo(bOut);
        bOut.flush();
        bOut.close();
      } catch (IOException e) {
        throw new DataParserException(Errors.PROTOBUF_10, e.toString(), e);
      }
      String path = fieldPath.isEmpty() ? FORWARD_SLASH : fieldPath;
      byte[] bytes = org.apache.commons.codec.binary.Base64.encodeBase64(bOut.toByteArray());
      record.getHeader().setAttribute(PROTOBUF_UNKNOWN_FIELDS_PREFIX + path, new String(bytes, StandardCharsets.UTF_8));
    }

    return Field.createListMap(sdcRecordMapFieldValue);
  }
public static Field protobufToSdcField ( Record record , String fieldPath , Descriptors . Descriptor descriptor , Map < String , Set < Descriptors . FieldDescriptor > > messageTypeToExtensionMap , Object message ) throws DataParserException { LinkedHashMap < String , Field > sdcRecordMapFieldValue = new LinkedHashMap <> ( ) ; // get all the expected fields from the proto file Map < String , Descriptors . FieldDescriptor > protobufFields = new LinkedHashMap <> ( ) ; for ( Descriptors . FieldDescriptor fieldDescriptor : descriptor . getFields ( ) ) { protobufFields . put ( fieldDescriptor . getName ( ) , fieldDescriptor ) ; } // get all fields in the read message Map < Descriptors . FieldDescriptor , Object > values = ( ( DynamicMessage ) message ) . getAllFields ( ) ; // for every field present in the proto definition create an sdc field. for ( Descriptors . FieldDescriptor fieldDescriptor : protobufFields . values ( ) ) { Object value = values . get ( fieldDescriptor ) ; sdcRecordMapFieldValue . put ( fieldDescriptor . getName ( ) , createField ( record , fieldPath , fieldDescriptor , messageTypeToExtensionMap , value ) ) ; } // handle applicable extensions for this message type if ( messageTypeToExtensionMap . containsKey ( descriptor . getFullName ( ) ) ) { for ( Descriptors . FieldDescriptor fieldDescriptor : messageTypeToExtensionMap . get ( descriptor . getFullName ( ) ) ) { if ( values . containsKey ( fieldDescriptor ) ) { Object value = values . get ( fieldDescriptor ) ; sdcRecordMapFieldValue . put ( fieldDescriptor . getName ( ) , createField ( record , fieldPath , fieldDescriptor , messageTypeToExtensionMap , value ) ) ; } } } // handle unknown fields // unknown fields can go into the record header UnknownFieldSet unknownFields = ( ( DynamicMessage ) message ) . getUnknownFields ( ) ; if ( ! unknownFields . asMap ( ) . isEmpty ( ) ) { ByteArrayOutputStream bOut = new ByteArrayOutputStream ( ) ; try { unknownFields . writeDelimitedTo ( bOut ) ; bOut . flush ( ) ; bOut . close ( ) ; } catch ( IOException e ) { throw new DataParserException ( Errors . PROTOBUF_10 , e . toString ( ) , e ) ; } String path = fieldPath . isEmpty ( ) ? FORWARD_SLASH : fieldPath ; byte [ ] bytes = org . apache . commons . codec . binary . Base64 . encodeBase64 ( bOut . toByteArray ( ) ) ; record . getHeader ( ) . setAttribute ( PROTOBUF_UNKNOWN_FIELDS_PREFIX + path , new String ( bytes , StandardCharsets . UTF_8 ) ) ; } return Field . createListMap ( sdcRecordMapFieldValue ) ; }
Converts a protobuf message to an SDC Record Field.

@param record                    SDC Record to add field to
@param fieldPath                 location in record where to insert field.
@param descriptor                protobuf descriptor instance
@param messageTypeToExtensionMap protobuf extensions map
@param message                   message to decode and insert into the specified field path
@return new Field instance representing the decoded message
@throws DataParserException
Converts a protobuf message to an SDC Record Field .