@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 testMapIntMessage() throws Exception {
    RecordConsumer readConsumerMock = Mockito.mock(RecordConsumer.class);
    ProtoWriteSupport<TestProtobuf.MapIntMessage> instance =
        createReadConsumerInstance(TestProtobuf.MapIntMessage.class, readConsumerMock);

    TestProtobuf.MapIntMessage.Builder msg = TestProtobuf.MapIntMessage.newBuilder();
    msg.putMapInt(123, 1);
    msg.putMapInt(234, 2);
    instance.write(msg.build());

    InOrder inOrder = Mockito.inOrder(readConsumerMock);

    inOrder.verify(readConsumerMock).startMessage();
    inOrder.verify(readConsumerMock).startField("mapInt", 0);

    inOrder.verify(readConsumerMock).startGroup();
    inOrder.verify(readConsumerMock).startField("key", 0);
    inOrder.verify(readConsumerMock).addInteger(123);
    inOrder.verify(readConsumerMock).endField("key", 0);
    inOrder.verify(readConsumerMock).startField("value", 1);
    inOrder.verify(readConsumerMock).addInteger(1);
    inOrder.verify(readConsumerMock).endField("value", 1);
    inOrder.verify(readConsumerMock).endGroup();

    inOrder.verify(readConsumerMock).startGroup();
    inOrder.verify(readConsumerMock).startField("key", 0);
    inOrder.verify(readConsumerMock).addInteger(234);
    inOrder.verify(readConsumerMock).endField("key", 0);
    inOrder.verify(readConsumerMock).startField("value", 1);
    inOrder.verify(readConsumerMock).addInteger(2);
    inOrder.verify(readConsumerMock).endField("value", 1);
    inOrder.verify(readConsumerMock).endGroup();

    inOrder.verify(readConsumerMock).endField("mapInt", 0);
    inOrder.verify(readConsumerMock).endMessage();
    Mockito.verifyNoMoreInteractions(readConsumerMock);
  }