@Override
  public void write(T record) {
    recordConsumer.startMessage();
    try {
      messageWriter.writeTopLevelMessage(record);
    } catch (RuntimeException e) {
      Message m = (record instanceof Message.Builder) ? ((Message.Builder) record).build() : (Message) record;
      LOG.error("Cannot write message {}: {}", e.getMessage(), m);
      throw e;
    }
    recordConsumer.endMessage();
  }
@Test
  public void testProto3WrappedMessageUnwrappedRoundTrip() throws Exception {
    TestProto3.WrappedMessage.Builder msg = TestProto3.WrappedMessage.newBuilder();
    msg.setWrappedDouble(DoubleValue.of(0.577));
    msg.setWrappedFloat(FloatValue.of(3.1415f));
    msg.setWrappedInt64(Int64Value.of(1_000_000_000L * 4));
    msg.setWrappedUInt64(UInt64Value.of(1_000_000_000L * 9));
    msg.setWrappedInt32(Int32Value.of(1_000_000 * 3));
    msg.setWrappedUInt32(UInt32Value.of(1_000_000 * 8));
    msg.setWrappedBool(BoolValue.of(true));
    msg.setWrappedString(StringValue.of("Good Will Hunting"));
    msg.setWrappedBytes(BytesValue.of(ByteString.copyFrom("someText", "UTF-8")));

    // Write them out and read them back
    Path tmpFilePath = TestUtils.someTemporaryFilePath();
    ParquetWriter<MessageOrBuilder> writer = ProtoParquetWriter.<MessageOrBuilder>builder(tmpFilePath)
        .withMessage(TestProto3.WrappedMessage.class)
        .config(ProtoWriteSupport.PB_UNWRAP_PROTO_WRAPPERS, "true")
        .build();
    writer.write(msg);
    writer.close();
    List<TestProto3.WrappedMessage> gotBack = TestUtils.readMessages(tmpFilePath, TestProto3.WrappedMessage.class);

    TestProto3.WrappedMessage gotBackFirst = gotBack.get(0);
    assertEquals(0.577, gotBackFirst.getWrappedDouble().getValue(), 1e-5);
    assertEquals(3.1415f, gotBackFirst.getWrappedFloat().getValue(), 1e-5f);
    assertEquals(1_000_000_000L * 4, gotBackFirst.getWrappedInt64().getValue());
    assertEquals(1_000_000_000L * 9, gotBackFirst.getWrappedUInt64().getValue());
    assertEquals(1_000_000 * 3, gotBackFirst.getWrappedInt32().getValue());
    assertEquals(1_000_000 * 8, gotBackFirst.getWrappedUInt32().getValue());
    assertEquals(BoolValue.of(true), gotBackFirst.getWrappedBool());
    assertEquals("Good Will Hunting", gotBackFirst.getWrappedString().getValue());
    assertEquals(
        ByteString.copyFrom("someText", "UTF-8"),
        gotBackFirst.getWrappedBytes().getValue());
  }