public static SerializableFunction<byte[], Row> getProtoBytesToRowFunction(
      String fileDescriptorPath, String messageName) {

    ProtoSchemaInfo dynamicProtoDomain = getProtoDomain(fileDescriptorPath, messageName);
    ProtoDomain protoDomain = dynamicProtoDomain.getProtoDomain();
    @SuppressWarnings("unchecked")
    ProtoDynamicMessageSchema<DynamicMessage> protoDynamicMessageSchema =
        ProtoDynamicMessageSchema.forDescriptor(protoDomain, messageName);
    return new SimpleFunction<byte[], Row>() {
      @Override
      public Row apply(byte[] input) {
        try {
          List<String> messageElements = Splitter.on('.').splitToList(messageName);
          String messageTypeByName = messageElements.get(messageElements.size() - 1);
          final Descriptors.Descriptor descriptor =
              protoDomain
                  .getFileDescriptor(dynamicProtoDomain.getFileName())
                  .findMessageTypeByName(messageTypeByName);
          DynamicMessage dynamicMessage = DynamicMessage.parseFrom(descriptor, input);
          SerializableFunction<DynamicMessage, Row> res =
              protoDynamicMessageSchema.getToRowFunction();
          return res.apply(dynamicMessage);
        } catch (InvalidProtocolBufferException e) {
          LOG.error("Error parsing to DynamicMessage", e);
          throw new RuntimeException(e);
        }
      }
    };
  }
@Test(expected = java.lang.RuntimeException.class)
  public void testProtoBytesToRowFunctionReturnsRowFailure() {
    // Create a proto bytes to row function
    SerializableFunction<byte[], Row> protoBytesToRowFunction =
        ProtoByteUtils.getProtoBytesToRowFunction(DESCRIPTOR_PATH, MESSAGE_NAME);

    // Create some test input bytes that are not matching
    byte[] inputBytes = new byte[] {1, 2, 3, 4, 5};

    // Call the proto bytes to row function that should fail because the input does not match
    protoBytesToRowFunction.apply(inputBytes);
  }