@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 testProto3WrappedMessageWithNullsRoundTrip() throws Exception {
    TestProto3.WrappedMessage.Builder msg = TestProto3.WrappedMessage.newBuilder();
    msg.setWrappedFloat(FloatValue.of(3.1415f));
    msg.setWrappedString(StringValue.of("Good Will Hunting"));
    msg.setWrappedInt32(Int32Value.of(0));

    // 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);
    assertFalse(gotBackFirst.hasWrappedDouble());
    assertEquals(3.1415f, gotBackFirst.getWrappedFloat().getValue(), 1e-5f);

    // double-check that nulls are honored
    assertTrue(gotBackFirst.hasWrappedFloat());
    assertFalse(gotBackFirst.hasWrappedInt64());
    assertFalse(gotBackFirst.hasWrappedUInt64());
    assertTrue(gotBackFirst.hasWrappedInt32());
    assertFalse(gotBackFirst.hasWrappedUInt32());
    assertEquals(0, gotBackFirst.getWrappedUInt32().getValue());
    assertFalse(gotBackFirst.hasWrappedBool());
    assertEquals("Good Will Hunting", gotBackFirst.getWrappedString().getValue());
    assertFalse(gotBackFirst.hasWrappedBytes());
  }