commonlib/src/main/java/com/streamsets/pipeline/lib/util/ProtobufTypeUtil.java
@SuppressWarnings("unchecked")
  private static Field createField(
      Record record,
      String fieldPath,
      Descriptors.FieldDescriptor fieldDescriptor,
      Map<String, Set<Descriptors.FieldDescriptor>> messageTypeToExtensionMap,
      Object message
  ) throws DataParserException {
    Field newField;
    if (message == null) {
      // If the message does not contain required fields then builder.build() throws UninitializedMessageException
      Object defaultValue = null;
      Descriptors.FieldDescriptor.JavaType javaType = fieldDescriptor.getJavaType();
      // get default values only for optional fields and non-message types
      if (fieldDescriptor.isOptional() && fieldDescriptor.getJavaType() != Descriptors.FieldDescriptor.JavaType.MESSAGE) {
        defaultValue = fieldDescriptor.getDefaultValue();
        //Default value for byte string should be converted to byte array
        if (javaType == Descriptors.FieldDescriptor.JavaType.BYTE_STRING
            && defaultValue instanceof ByteString) {
          defaultValue = ((ByteString)defaultValue).toByteArray();
        }
      }
      newField = Field.create(getFieldType(javaType), defaultValue);
    } else if (fieldDescriptor.isMapField()) {
      // Map entry (protobuf 3 map)
      Map<String, Field> sdcMapFieldValues = new HashMap<>();
      Collection<DynamicMessage> mapEntries = (Collection<DynamicMessage>) message;
      // MapEntry
      for (DynamicMessage dynamicMessage : mapEntries) {
        // MapEntry has 2 fields, key and value
        Map<Descriptors.FieldDescriptor, Object> kv = dynamicMessage.getAllFields();
        String key = null;
        Object value = null;
        Descriptors.FieldDescriptor valueDescriptor = null;
        for (Map.Entry<Descriptors.FieldDescriptor, Object> entry : kv.entrySet()) {
          switch (entry.getKey().getName()) {
            case KEY:
              key = entry.getValue().toString();
              break;
            case VALUE:
              value = entry.getValue();
              valueDescriptor = entry.getKey();
              break;
            default:
              throw new DataParserException(Errors.PROTOBUF_09, entry.getKey().getName());
          }
        }
        if (key != null && valueDescriptor != null) {
          sdcMapFieldValues.put(
              key, createSdcField(record, fieldPath, valueDescriptor, messageTypeToExtensionMap, value)
          );
        }
      }

      newField = Field.create(sdcMapFieldValues);
    } else if (fieldDescriptor.isRepeated()) {
      // List entry (repeated)
      List<?> list = (List<?>) message;
      List<Field> listField = new ArrayList<>();
      for (int i = 0; i < list.size(); i++) {
        if (fieldDescriptor.getJavaType() == Descriptors.FieldDescriptor.JavaType.MESSAGE) {
          listField.add(
              protobufToSdcField(
                  record,
                  fieldPath + "[" + i + "]",
                  fieldDescriptor.getMessageType(),
                  messageTypeToExtensionMap,
                  list.get(i)
              )
          );
        } else {
          listField.add(
              createSdcField(
                  record,
                  fieldPath + "[" + i + "]",
                  fieldDescriptor,
                  messageTypeToExtensionMap,
                  list.get(i)
              )
          );
        }
      }
      newField = Field.create(listField);
    } else {
      // normal entry
      newField = createSdcField(record, fieldPath, fieldDescriptor, messageTypeToExtensionMap, message);
    }
    return newField;
  }
@SuppressWarnings("unchecked")
  private static Field createField(
      Record record,
      String fieldPath,
      Descriptors.FieldDescriptor fieldDescriptor,
      Map<String, Set<Descriptors.FieldDescriptor>> messageTypeToExtensionMap,
      Object message
  ) throws DataParserException {
    Field newField;
    if (message == null) {
      // If the message does not contain required fields then builder.build() throws UninitializedMessageException
      Object defaultValue = null;
      Descriptors.FieldDescriptor.JavaType javaType = fieldDescriptor.getJavaType();
      // get default values only for optional fields and non-message types
      if (fieldDescriptor.isOptional() && fieldDescriptor.getJavaType() != Descriptors.FieldDescriptor.JavaType.MESSAGE) {
        defaultValue = fieldDescriptor.getDefaultValue();
        //Default value for byte string should be converted to byte array
        if (javaType == Descriptors.FieldDescriptor.JavaType.BYTE_STRING
            && defaultValue instanceof ByteString) {
          defaultValue = ((ByteString)defaultValue).toByteArray();
        }
      }
      newField = Field.create(getFieldType(javaType), defaultValue);
    } else if (fieldDescriptor.isMapField()) {
      // Map entry (protobuf 3 map)
      Map<String, Field> sdcMapFieldValues = new HashMap<>();
      Collection<DynamicMessage> mapEntries = (Collection<DynamicMessage>) message;
      // MapEntry
      for (DynamicMessage dynamicMessage : mapEntries) {
        // MapEntry has 2 fields, key and value
        Map<Descriptors.FieldDescriptor, Object> kv = dynamicMessage.getAllFields();
        String key = null;
        Object value = null;
        Descriptors.FieldDescriptor valueDescriptor = null;
        for (Map.Entry<Descriptors.FieldDescriptor, Object> entry : kv.entrySet()) {
          switch (entry.getKey().getName()) {
            case KEY:
              key = entry.getValue().toString();
              break;
            case VALUE:
              value = entry.getValue();
              valueDescriptor = entry.getKey();
              break;
            default:
              throw new DataParserException(Errors.PROTOBUF_09, entry.getKey().getName());
          }
        }
        if (key != null && valueDescriptor != null) {
          sdcMapFieldValues.put(
              key, createSdcField(record, fieldPath, valueDescriptor, messageTypeToExtensionMap, value)
          );
        }
      }

      newField = Field.create(sdcMapFieldValues);
    } else if (fieldDescriptor.isRepeated()) {
      // List entry (repeated)
      List<?> list = (List<?>) message;
      List<Field> listField = new ArrayList<>();
      for (int i = 0; i < list.size(); i++) {
        if (fieldDescriptor.getJavaType() == Descriptors.FieldDescriptor.JavaType.MESSAGE) {
          listField.add(
              protobufToSdcField(
                  record,
                  fieldPath + "[" + i + "]",
                  fieldDescriptor.getMessageType(),
                  messageTypeToExtensionMap,
                  list.get(i)
              )
          );
        } else {
          listField.add(
              createSdcField(
                  record,
                  fieldPath + "[" + i + "]",
                  fieldDescriptor,
                  messageTypeToExtensionMap,
                  list.get(i)
              )
          );
        }
      }
      newField = Field.create(listField);
    } else {
      // normal entry
      newField = createSdcField(record, fieldPath, fieldDescriptor, messageTypeToExtensionMap, message);
    }
    return newField;
  }
@ SuppressWarnings ( "unchecked" ) private static Field createField ( Record record , String fieldPath , Descriptors . FieldDescriptor fieldDescriptor , Map < String , Set < Descriptors . FieldDescriptor > > messageTypeToExtensionMap , Object message ) throws DataParserException { Field newField ; if ( message == null ) { // If the message does not contain required fields then builder.build() throws UninitializedMessageException Object defaultValue = null ; Descriptors . FieldDescriptor . JavaType javaType = fieldDescriptor . getJavaType ( ) ; // get default values only for optional fields and non-message types if ( fieldDescriptor . isOptional ( ) && fieldDescriptor . getJavaType ( ) != Descriptors . FieldDescriptor . JavaType . MESSAGE ) { defaultValue = fieldDescriptor . getDefaultValue ( ) ; //Default value for byte string should be converted to byte array if ( javaType == Descriptors . FieldDescriptor . JavaType . BYTE_STRING && defaultValue instanceof ByteString ) { defaultValue = ( ( ByteString ) defaultValue ) . toByteArray ( ) ; } } newField = Field . create ( getFieldType ( javaType ) , defaultValue ) ; } else if ( fieldDescriptor . isMapField ( ) ) { // Map entry (protobuf 3 map) Map < String , Field > sdcMapFieldValues = new HashMap <> ( ) ; Collection < DynamicMessage > mapEntries = ( Collection < DynamicMessage > ) message ; // MapEntry for ( DynamicMessage dynamicMessage : mapEntries ) { // MapEntry has 2 fields, key and value Map < Descriptors . FieldDescriptor , Object > kv = dynamicMessage . getAllFields ( ) ; String key = null ; Object value = null ; Descriptors . FieldDescriptor valueDescriptor = null ; for ( Map . Entry < Descriptors . FieldDescriptor , Object > entry : kv . entrySet ( ) ) { switch ( entry . getKey ( ) . getName ( ) ) { case KEY : key = entry . getValue ( ) . toString ( ) ; break ; case VALUE : value = entry . getValue ( ) ; valueDescriptor = entry . getKey ( ) ; break ; default : throw new DataParserException ( Errors . PROTOBUF_09 , entry . getKey ( ) . getName ( ) ) ; } } if ( key != null && valueDescriptor != null ) { sdcMapFieldValues . put ( key , createSdcField ( record , fieldPath , valueDescriptor , messageTypeToExtensionMap , value ) ) ; } } newField = Field . create ( sdcMapFieldValues ) ; } else if ( fieldDescriptor . isRepeated ( ) ) { // List entry (repeated) List < ? > list = ( List < ? > ) message ; List < Field > listField = new ArrayList <> ( ) ; for ( int i = 0 ; i < list . size ( ) ; i ++ ) { if ( fieldDescriptor . getJavaType ( ) == Descriptors . FieldDescriptor . JavaType . MESSAGE ) { listField . add ( protobufToSdcField ( record , fieldPath + "[" + i + "]" , fieldDescriptor . getMessageType ( ) , messageTypeToExtensionMap , list . get ( i ) ) ) ; } else { listField . add ( createSdcField ( record , fieldPath + "[" + i + "]" , fieldDescriptor , messageTypeToExtensionMap , list . get ( i ) ) ) ; } } newField = Field . create ( listField ) ; } else { // normal entry newField = createSdcField ( record , fieldPath , fieldDescriptor , messageTypeToExtensionMap , message ) ; } return newField ; }
Creates an SDC Record Field from the provided protobuf message and descriptor.

@param record                    record to be augmented
@param fieldPath                 field path in the record to insert new Field
@param fieldDescriptor           descriptor for this message
@param messageTypeToExtensionMap protobuf type extensions
@param message                   protobuf message to decode
@return reference to the Field added to the record.
@throws DataParserException
Creates an SDC Record Field from the provided protobuf message and descriptor .