public Collection<Descriptors.Descriptor> findDescriptorsByOption(String optionName) {
  Descriptors.FieldDescriptor fieldDescriptor = optionsCatalog.getMessageOptionByName(optionName);
  return descriptorMap.values().stream()
      .filter(
          descriptor -> {
            DescriptorProtos.MessageOptions options = descriptor.getOptions();
            UnknownFieldSet.Field unknown =
                options.getUnknownFields().getField(fieldDescriptor.getNumber());
            if (unknown.getLengthDelimitedList().size()
                    + unknown.getFixed64List().size()
                    + unknown.getFixed32List().size()
                    + unknown.getVarintList().size()
                > 0) {
              return true;
            }
            return options.getAllFields().containsKey(fieldDescriptor);
          })
      .collect(Collectors.toList());
}
 

public Collection<Descriptors.FileDescriptor> findFileDescriptorsByOption(String optionName) {
  Descriptors.FieldDescriptor fieldDescriptor = optionsCatalog.getFileOptionByName(optionName);
  return fileDescriptorMap.values().stream()
      .filter(
          descriptor -> {
            DescriptorProtos.FileOptions options = descriptor.getOptions();
            UnknownFieldSet.Field unknown =
                options.getUnknownFields().getField(fieldDescriptor.getNumber());
            if (unknown.getLengthDelimitedList().size()
                    + unknown.getFixed64List().size()
                    + unknown.getFixed32List().size()
                    + unknown.getVarintList().size()
                > 0) {
              return true;
            }
            return options.getAllFields().containsKey(fieldDescriptor);
          })
      .collect(Collectors.toList());
}
 

public Collection<Descriptors.EnumDescriptor> findEnumDescriptorsByOption(String optionName) {
  Descriptors.FieldDescriptor fieldDescriptor = optionsCatalog.getEnumOptionByName(optionName);
  return enumMap.values().stream()
      .filter(
          descriptor -> {
            DescriptorProtos.EnumOptions options = descriptor.getOptions();
            UnknownFieldSet.Field unknown =
                options.getUnknownFields().getField(fieldDescriptor.getNumber());
            if (unknown.getLengthDelimitedList().size()
                    + unknown.getFixed64List().size()
                    + unknown.getFixed32List().size()
                    + unknown.getVarintList().size()
                > 0) {
              return true;
            }
            return options.getAllFields().containsKey(fieldDescriptor);
          })
      .collect(Collectors.toList());
}
 

public Collection<Descriptors.ServiceDescriptor> findServiceDescriptorsByOption(
    String optionName) {
  Descriptors.FieldDescriptor fieldDescriptor = optionsCatalog.getServiceOptionByName(optionName);
  return serviceMap.values().stream()
      .filter(
          descriptor -> {
            DescriptorProtos.ServiceOptions options = descriptor.getOptions();
            UnknownFieldSet.Field unknown =
                options.getUnknownFields().getField(fieldDescriptor.getNumber());
            if (unknown.getLengthDelimitedList().size()
                    + unknown.getFixed64List().size()
                    + unknown.getFixed32List().size()
                    + unknown.getVarintList().size()
                > 0) {
              return true;
            }
            return options.getAllFields().containsKey(fieldDescriptor);
          })
      .collect(Collectors.toList());
}
 

private Map<Descriptors.FieldDescriptor, Object> convertUnknownFieldValue(
    UnknownFieldSet unknownFieldSet, Map<Integer, Descriptors.FieldDescriptor> optionsMap) {
  Map<Descriptors.FieldDescriptor, Object> unknownFieldValues = new LinkedHashMap<>();
  unknownFieldSet
      .asMap()
      .forEach(
          (number, field) -> {
            Descriptors.FieldDescriptor fieldDescriptor = optionsMap.get(number);
            if (fieldDescriptor.isRepeated()) {
              unknownFieldValues.put(
                  fieldDescriptor, convertUnknownFieldList(fieldDescriptor, field));
            } else {
              unknownFieldValues.put(
                  fieldDescriptor, convertUnknownFieldValue(fieldDescriptor, field));
            }
          });
  return unknownFieldValues;
}
 

private UnknownFieldSet buildUnknownFieldSet(List<OptionChangeInfo> optionChanges) {
  UnknownFieldSet.Builder unknownFieldSetBuilder = UnknownFieldSet.newBuilder();
  for (OptionChangeInfo optionChange : optionChanges) {
    switch (optionChange.getChangeType()) {
      case ADDITION:
        unknownFieldSetBuilder =
            mergeUnknownField(unknownFieldSetBuilder, optionChange.getPayloadNew());
        break;
      case REMOVAL:
        unknownFieldSetBuilder.clearField(optionChange.getOptionNumber());
        break;
      case PAYLOAD_CHANGED:
        unknownFieldSetBuilder.clearField(optionChange.getOptionNumber());
        unknownFieldSetBuilder =
            mergeUnknownField(unknownFieldSetBuilder, optionChange.getPayloadNew());
        break;
    }
  }
  return unknownFieldSetBuilder.build();
}
 

private void handleMissingField(String fieldName, JsonParser parser,
                                       ExtensionRegistry extensionRegistry,
                                       UnknownFieldSet.Builder builder) throws IOException {

    JsonToken token = parser.nextToken();
    if (token.equals(JsonToken.START_OBJECT)) {
        // Message structure
    	token = parser.nextToken(); // skip name
    	while (token != null && !token.equals(JsonToken.END_OBJECT)) {
            handleMissingField(fieldName, parser, extensionRegistry, builder);
            token = parser.nextToken(); // get } or field name
        }
    } else if (token.equals(JsonToken.START_ARRAY)) {
        // Collection
        do {
            handleMissingField(fieldName, parser, extensionRegistry, builder);
            token = parser.getCurrentToken(); // got value or ]
        } while (token != null && !token.equals(JsonToken.END_ARRAY));
    } else {
        // Primitive value
    	// NULL, INT, BOOL, STRING
    	// nothing to do..
    }
}
 

private static FieldNumberTree fromUnknownFieldSet(UnknownFieldSet unknownFieldSet) {
  FieldNumberTree tree = new FieldNumberTree();
  for (int fieldNumber : unknownFieldSet.asMap().keySet()) {
    UnknownFieldSet.Field unknownField = unknownFieldSet.asMap().get(fieldNumber);
    for (UnknownFieldDescriptor unknownFieldDescriptor :
        UnknownFieldDescriptor.descriptors(fieldNumber, unknownField)) {
      Key key = Key.unknown(unknownFieldDescriptor);
      FieldNumberTree childTree = new FieldNumberTree();
      tree.children.put(key, childTree);

      if (unknownFieldDescriptor.type() == UnknownFieldDescriptor.Type.GROUP) {
        for (Object group : unknownFieldDescriptor.type().getValues(unknownField)) {
          childTree.merge(fromUnknownFieldSet((UnknownFieldSet) group));
        }
      }
    }
  }

  return tree;
}
 

private static void handleUnknownFields(
    Record record,
    String fieldPath,
    DynamicMessage.Builder builder
) throws IOException {
  String path = fieldPath.isEmpty() ? FORWARD_SLASH : fieldPath;
  String attribute = record.getHeader().getAttribute(ProtobufTypeUtil.PROTOBUF_UNKNOWN_FIELDS_PREFIX + path);
  if (attribute != null) {
    UnknownFieldSet.Builder unknownFieldBuilder = UnknownFieldSet.newBuilder();
    unknownFieldBuilder.mergeDelimitedFrom(
        new ByteArrayInputStream(
            org.apache.commons.codec.binary.Base64.decodeBase64(attribute.getBytes(StandardCharsets.UTF_8))
        )
    );
    UnknownFieldSet unknownFieldSet = unknownFieldBuilder.build();
    builder.setUnknownFields(unknownFieldSet);
  }
}
 

public static void checkRecordForUnknownFields(Record record, int i) throws IOException {
  // unknown fields are expected in paths for person and employee
  String attribute = record.getHeader().getAttribute(ProtobufTypeUtil.PROTOBUF_UNKNOWN_FIELDS_PREFIX + "/");
  UnknownFieldSet.Builder builder = UnknownFieldSet.newBuilder();
  builder.mergeDelimitedFrom(new ByteArrayInputStream(org.apache.commons.codec.binary.Base64.decodeBase64(attribute.getBytes())));
  UnknownFieldSet unknownFieldSet = builder.build();

  UnknownFieldsUtil.checkEmployeeUnknownFields(unknownFieldSet);

  if(i%2 == 0) {
    attribute = record.getHeader().getAttribute(ProtobufTypeUtil.PROTOBUF_UNKNOWN_FIELDS_PREFIX + "/engineer/person");
  } else {
    attribute = record.getHeader().getAttribute(ProtobufTypeUtil.PROTOBUF_UNKNOWN_FIELDS_PREFIX + "/exec/person");
  }
  builder = UnknownFieldSet.newBuilder();
  builder.mergeDelimitedFrom(new ByteArrayInputStream(org.apache.commons.codec.binary.Base64.decodeBase64(attribute.getBytes())));
  unknownFieldSet = builder.build();

  UnknownFieldsUtil.checkPersonUnknownFields(unknownFieldSet);
}
 

public static UnknownFieldSet getEmployeeUnknownFields() {
  // add unknown fields
  UnknownFieldSet.Field unknownStringField = UnknownFieldSet.Field.newBuilder()
    .addLengthDelimited(ByteString.copyFromUtf8("Hello San FRancisco!"))
    .build();
  UnknownFieldSet.Field unknownVarIntField = UnknownFieldSet.Field.newBuilder()
    .addVarint(123456789)
    .build();

  UnknownFieldSet employeeUnknownFields = UnknownFieldSet.newBuilder()
    .addField(345, unknownStringField)
    .addField(456, unknownVarIntField)
    .build();

  return employeeUnknownFields;
}
 

/**
 * https://developers.google.com/protocol-buffers/docs/encoding#structure
 *
 * @param fieldDescriptor
 * @param field
 * @return
 */
private Object convertUnknownFieldList(
    Descriptors.FieldDescriptor fieldDescriptor, UnknownFieldSet.Field field) {
  List list = new ArrayList();
  if (field.getLengthDelimitedList().size() > 0) {
    field
        .getLengthDelimitedList()
        .forEach(value -> list.add(convertFieldValue(fieldDescriptor, value)));
  }
  if (field.getFixed32List().size() > 0) {
    field
        .getFixed32List()
        .forEach(value -> list.add(convertFieldValue(fieldDescriptor, value)));
  }
  if (field.getFixed64List().size() > 0) {
    field
        .getFixed64List()
        .forEach(value -> list.add(convertFieldValue(fieldDescriptor, value)));
  }
  if (field.getVarintList().size() > 0) {
    field.getVarintList().forEach(value -> list.add(convertFieldValue(fieldDescriptor, value)));
  }
  if (field.getGroupList().size() > 0) {
    throw new RuntimeException("Groups are not implemented");
  }
  return list;
}
 

private Object convertUnknownFieldValue(
    Descriptors.FieldDescriptor fieldDescriptor, UnknownFieldSet.Field field) {

  if (field.getLengthDelimitedList().size() > 0) {
    if (field.getLengthDelimitedList().size() > 1) {
      throw new RuntimeException(
          "Single value should not contrain more then 1 value in the unknown field");
    }
    return convertFieldValue(fieldDescriptor, field.getLengthDelimitedList().get(0));
  }
  if (field.getFixed32List().size() > 0) {
    if (field.getFixed32List().size() > 1) {
      throw new RuntimeException(
          "Single value should not contrain more then 1 value in the unknown field");
    }
    return convertFieldValue(fieldDescriptor, field.getFixed32List().get(0));
  }
  if (field.getFixed64List().size() > 0) {
    if (field.getFixed64List().size() > 1) {
      throw new RuntimeException(
          "Single value should not contrain more then 1 value in the unknown field");
    }
    return convertFieldValue(fieldDescriptor, field.getFixed64List().get(0));
  }
  if (field.getVarintList().size() > 0) {
    if (field.getVarintList().size() > 1) {
      throw new RuntimeException(
          "Single value should not contrain more then 1 value in the unknown field");
    }
    return convertFieldValue(fieldDescriptor, field.getVarintList().get(0));
  }
  if (field.getGroupList().size() > 0) {
    throw new RuntimeException("Groups are not implemented");
  }
  return null;
}
 

private DescriptorProtos.FileDescriptorProto.Builder applyFileOptionChanges(
    Descriptors.FileDescriptor fileDescriptor, List<OptionChangeInfo> optionChanges) {
  DescriptorProtos.FileDescriptorProto.Builder newDescriptorBuilder =
      fileDescriptor.toProto().toBuilder();
  UnknownFieldSet unknownFieldSet = buildUnknownFieldSet(optionChanges);
  DescriptorProtos.FileOptions fileOptions =
      DescriptorProtos.FileOptions.newBuilder().setUnknownFields(unknownFieldSet).build();
  return newDescriptorBuilder.setOptions(fileOptions).clearMessageType();
}
 

private DescriptorProtos.DescriptorProto applyMessageOptionChanges(
    DescriptorProtos.DescriptorProto.Builder newDescriptorBuilder,
    Descriptors.Descriptor descriptor,
    List<OptionChangeInfo> optionChanges) {
  UnknownFieldSet unknownFieldSet = buildUnknownFieldSet(optionChanges);
  DescriptorProtos.MessageOptions messageOptions =
      DescriptorProtos.MessageOptions.newBuilder().setUnknownFields(unknownFieldSet).build();
  return newDescriptorBuilder.setOptions(messageOptions).build();
}
 

/**
    * Like {@code print()}, but writes directly to a {@code String} and returns it.
    */
public String printToString(final UnknownFieldSet fields) {
	try {
           ByteArrayOutputStream out = new ByteArrayOutputStream();
           print(fields, out, defaultCharset);
           out.flush();
           return out.toString();
       } catch (IOException e) {
           throw new RuntimeException("Writing to a StringBuilder threw an IOException (should never happen).",
                                      e);
       }
}
 

private UnknownFieldSet.Builder mergeUnknownField(
    UnknownFieldSet.Builder globalUnknownFieldSetBuilder, ByteString payload) {
  try {
    return globalUnknownFieldSetBuilder.mergeFrom(payload);
  } catch (Exception e) {
    throw new RuntimeException("merge of unknown field failed", e);
  }
}
 

/**
 * Parses a serialized HyperLogLog++ {@link AggregatorStateProto} and populates this object's
 * fields. If the {@code input} supports aliasing (for byte arrays and {@link ByteBuffer}, see
 * {@link CodedInputStream#enableAliasing(boolean) for details}), {@link #data} and {@link
 * #sparseData} will <em>alias</em> the given bytes &mdash; sharing the same memory.
 *
 * @throws IOException If the stream does not contain a serialized {@link AggregatorStateProto} or
 *     if fields are set that would typically not belong
 */
public void parse(CodedInputStream input) throws IOException {
  // Reset defaults as values set to the default will not be encoded in the protocol buffer.
  clear();

  UnknownFieldSet.Builder ignoredFields = UnknownFieldSet.newBuilder();
  while (!input.isAtEnd()) {
    int tag = input.readTag();
    switch (tag) {
      case TYPE_TAG:
        type = AggregatorType.forNumber(input.readEnum());
        break;
      case NUM_VALUES_TAG:
        numValues = input.readInt64();
        break;
      case ENCODING_VERSION_TAG:
        encodingVersion = input.readInt32();
        break;
      case VALUE_TYPE_TAG:
        valueType = ValueType.forNumber(input.readEnum());
        break;
      case HYPERLOGLOGPLUS_UNIQUE_STATE_TAG:
        parseHll(input, input.readInt32());
        break;
      default:
        ignoredFields.mergeFieldFrom(tag, input);
    }
  }
}
 

/**
 * Parses a {@link HyperLogLogPlusUniqueStateProto} message. Since the message is nested within an
 * {@link AggregatorStateProto}, we limit ourselves to reading only the bytes of the specified
 * message length.
 */
private void parseHll(CodedInputStream input, int size) throws IOException {
  int limit = input.getTotalBytesRead() + size;

  ByteBuffer buffer;
  UnknownFieldSet.Builder ignoredFields = UnknownFieldSet.newBuilder();
  while (input.getTotalBytesRead() < limit && !input.isAtEnd()) {
    int tag = input.readTag();
    switch (tag) {
      case SPARSE_SIZE_TAG:
        sparseSize = input.readInt32();
        break;
      case PRECISION_OR_NUM_BUCKETS_TAG:
        precision = input.readInt32();
        break;
      case SPARSE_PRECISION_OR_NUM_BUCKETS_TAG:
        sparsePrecision = input.readInt32();
        break;
      case DATA_TAG:
        buffer = input.readByteBuffer();
        data = ByteSlice.copyOnWrite(buffer);
        break;
      case SPARSE_DATA_TAG:
        buffer = input.readByteBuffer();
        sparseData = GrowingByteSlice.copyOnWrite(buffer);
        break;
      default:
        ignoredFields.mergeFieldFrom(tag, input);
    }
  }
}
 

@Nonnull
protected DynamicMessage deserializeUnion(@Nonnull final Descriptors.Descriptor unionDescriptor,
                                          @Nonnull final Tuple primaryKey,
                                          @Nonnull final byte[] serialized,
                                          int metaDataVersion) {
    final DynamicMessage unionMessage = deserializeFromBytes(unionDescriptor, serialized);
    final Map<Descriptors.FieldDescriptor, Object> allFields = unionMessage.getAllFields();
    final Map<Integer, UnknownFieldSet.Field> unknownFields = unionMessage.getUnknownFields().asMap();
    if (!(allFields.size() == 1 && unknownFields.isEmpty())) {
        final String detailedMessage;
        if (!unknownFields.isEmpty()) {
            detailedMessage = " because there are unknown fields";
        } else if (allFields.size() > 1) {
            detailedMessage = " because there are extra known fields";
        } else {
            detailedMessage = " because there are no fields";
        }
        final String message = "Could not deserialize union message" + detailedMessage;
        final RecordSerializationException ex = new RecordSerializationException(message)
                .addLogInfo("unknownFields", unknownFields.keySet())
                .addLogInfo("fields", getFieldNames(allFields.keySet()))
                .addLogInfo("primaryKey", primaryKey)
                .addLogInfo("metaDataVersion", metaDataVersion);
        throw ex;
    }
    return unionMessage;
}
 

/**
 * Outputs a textual representation of {@code fields} to {@code output}.
 */
public static void print(UnknownFieldSet fields, Appendable output, boolean selfType)
        throws IOException {
    JsonGenerator generator = new JsonGenerator(output);
    generator.print("{");
    printUnknownFields(fields, generator, selfType);
    generator.print("}");
}
 

/**
 * Like {@code print()}, but writes directly to a {@code String} and returns it.
 */
public static String printToString(UnknownFieldSet fields, boolean selfType) {
    try {
        StringBuilder text = new StringBuilder();
        print(fields, text, selfType);
        return text.toString();
    } catch (IOException e) {
        throw new RuntimeException(
                "Writing to a StringBuilder threw an IOException (should never happen).",
                e);
    }
}
 

private boolean compareUnknownFields(
    Message message1, Message message2, @Nullable Reporter reporter, List<SpecificField> stack) {
  UnknownFieldSet unknownFieldSet1 = message1.getUnknownFields();
  UnknownFieldSet unknownFieldSet2 = message2.getUnknownFields();
  return compareUnknownFields(
      message1, message2, unknownFieldSet1, unknownFieldSet2, reporter, stack);
}
 

@Override
public void print(UnknownFieldSet fields, OutputStream output, Charset cs)
		throws IOException {
	OutputStreamWriter writer = new OutputStreamWriter(output, cs);
	print(fields, writer);
	writer.flush();
}
 

/**
 * Outputs a Smile representation of {@code fields} to {@code output}.
 */
public void print(final UnknownFieldSet fields, OutputStream output, Charset cs) throws IOException {
    try {
        XMLStreamWriter generator = createGenerator(output);
        generator.writeStartElement(MESSAGE_ELEMENT);
    	
    	printUnknownFields(fields, generator);
    	generator.writeEndElement();
        generator.close();
    } catch (XMLStreamException e) {
        throw new IOException(e);
    }
}
 

/**
 * Outputs a textual representation of {@code fields} to {@code output}.
 */
public void print(final UnknownFieldSet fields, Appendable output) throws IOException {
    CouchDBGenerator generator = new CouchDBGenerator(output);
    generator.print("{");
    printUnknownFields(fields, generator);
    generator.print("}");
}
 

@Override
public void print(UnknownFieldSet fields, OutputStream output, Charset cs)
		throws IOException {
	OutputStreamWriter writer = new OutputStreamWriter(output, cs);
	print(fields, writer);
	writer.flush();
}
 

/**
    * Like {@code print()}, but writes directly to a {@code String} and returns it.
    */
public String printToString(final UnknownFieldSet fields) {
	try {
           ByteArrayOutputStream out = new ByteArrayOutputStream();
           print(fields, out, defaultCharset);
           out.flush();
           return out.toString();
       } catch (IOException e) {
           throw new RuntimeException("Writing to a StringBuilder threw an IOException (should never happen).",
                                      e);
       }
}
 

/**
 * Outputs a textual representation of {@code fields} to {@code output}.
 */
public void print(final UnknownFieldSet fields, Appendable output) throws IOException {
    XmlGenerator generator = new XmlGenerator(output);
    generator.print("<message>");
    printUnknownFields(fields, generator);
    generator.print("</message>");
}
 

/**
 * Outputs a Smile representation of {@code fields} to {@code output}.
 */
public void print(final UnknownFieldSet fields, OutputStream output, Charset cs) throws IOException {
	JsonGenerator generator = createGenerator(output);
	generator.writeStartObject();
	printUnknownFields(fields, generator);
    generator.writeEndObject();
    generator.close();
}