@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 testMapRecursion() {
    RecordConsumer readConsumerMock = Mockito.mock(RecordConsumer.class);
    Configuration conf = new Configuration();
    ProtoSchemaConverter.setMaxRecursion(conf, 1);
    ProtoWriteSupport<Value> spyWriter = createReadConsumerInstance(Value.class, readConsumerMock, conf);

    // Need to build it backwards due to clunky Struct.Builder interface.
    Value.Builder msg = Value.newBuilder().setStringValue("last");
    for (int i = 10; i > -1; --i) {
      Value.Builder next = Value.newBuilder();
      next.getStructValueBuilder().putFields("" + i, msg.build());
      msg = next;
    }
    Value built = msg.build();
    spyWriter.write(built);

    InOrder inOrder = Mockito.inOrder(readConsumerMock);

    inOrder.verify(readConsumerMock).startMessage();
    inOrder.verify(readConsumerMock).startField("struct_value", 4);
    inOrder.verify(readConsumerMock).startGroup();
    inOrder.verify(readConsumerMock).startField("fields", 0);
    inOrder.verify(readConsumerMock).startGroup();
    inOrder.verify(readConsumerMock).startField("key", 0);
    inOrder.verify(readConsumerMock).addBinary(Binary.fromConstantByteArray("0".getBytes()));
    inOrder.verify(readConsumerMock).endField("key", 0);
    inOrder.verify(readConsumerMock).startField("value", 1);
    inOrder.verify(readConsumerMock).startGroup();
    inOrder.verify(readConsumerMock).startField("struct_value", 4);
    inOrder.verify(readConsumerMock).startGroup();
    inOrder.verify(readConsumerMock).startField("fields", 0);
    inOrder.verify(readConsumerMock).startGroup();
    inOrder.verify(readConsumerMock).startField("key", 0);
    inOrder.verify(readConsumerMock).addBinary(Binary.fromConstantByteArray("1".getBytes()));
    inOrder.verify(readConsumerMock).endField("key", 0);
    inOrder.verify(readConsumerMock).startField("value", 1);
    inOrder.verify(readConsumerMock)
        .addBinary(Binary.fromConstantByteArray(built.getStructValue()
            .getFieldsOrThrow("0")
            .getStructValue()
            .getFieldsOrThrow("1")
            .toByteArray()));
    inOrder.verify(readConsumerMock).endField("value", 1);
    inOrder.verify(readConsumerMock).endGroup();
    inOrder.verify(readConsumerMock).endField("fields", 0);
    inOrder.verify(readConsumerMock).endGroup();
    inOrder.verify(readConsumerMock).endField("struct_value", 4);
    inOrder.verify(readConsumerMock).endGroup();
    inOrder.verify(readConsumerMock).endField("value", 1);
    inOrder.verify(readConsumerMock).endGroup();
    inOrder.verify(readConsumerMock).endField("fields", 0);
    inOrder.verify(readConsumerMock).endGroup();
    inOrder.verify(readConsumerMock).endField("struct_value", 4);
    inOrder.verify(readConsumerMock).endMessage();

    Mockito.verifyNoMoreInteractions(readConsumerMock);
  }