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

    Trees.WideTree.Builder msg = Trees.WideTree.newBuilder();
    Trees.WideTree.Builder cur = msg;

    for (int i = 0; i < 10; ++i) {
      cur.getValueBuilder().setTypeUrl("" + i);
      cur = cur.addChildrenBuilder();
    }
    Trees.WideTree built = msg.build();
    spyWriter.write(built);

    InOrder inOrder = Mockito.inOrder(readConsumerMock);

    inOrder.verify(readConsumerMock).startMessage();
    inOrder.verify(readConsumerMock).startField("value", 0);
    inOrder.verify(readConsumerMock).startGroup();
    inOrder.verify(readConsumerMock).startField("type_url", 0);
    inOrder.verify(readConsumerMock).addBinary(Binary.fromConstantByteArray("0".getBytes()));
    inOrder.verify(readConsumerMock).endField("type_url", 0);
    inOrder.verify(readConsumerMock).startField("value", 1);
    inOrder.verify(readConsumerMock).addBinary(Binary.fromConstantByteArray("".getBytes()));
    inOrder.verify(readConsumerMock).endField("value", 1);
    inOrder.verify(readConsumerMock).endGroup();
    inOrder.verify(readConsumerMock).endField("value", 0);
    inOrder.verify(readConsumerMock).startField("children", 1);
    inOrder.verify(readConsumerMock).startGroup();
    inOrder.verify(readConsumerMock).startField("value", 0);
    inOrder.verify(readConsumerMock).startGroup();
    inOrder.verify(readConsumerMock).startField("type_url", 0);
    inOrder.verify(readConsumerMock).addBinary(Binary.fromConstantByteArray("1".getBytes()));
    inOrder.verify(readConsumerMock).endField("type_url", 0);
    inOrder.verify(readConsumerMock).startField("value", 1);
    inOrder.verify(readConsumerMock).addBinary(Binary.fromConstantByteArray("".getBytes()));
    inOrder.verify(readConsumerMock).endField("value", 1);
    inOrder.verify(readConsumerMock).endGroup();
    inOrder.verify(readConsumerMock).endField("value", 0);
    inOrder.verify(readConsumerMock).startField("children", 1);
    inOrder.verify(readConsumerMock)
        .addBinary(Binary.fromConstantByteArray(
            built.getChildren(0).getChildren(0).toByteArray()));
    inOrder.verify(readConsumerMock).endField("children", 1);
    inOrder.verify(readConsumerMock).endGroup();
    inOrder.verify(readConsumerMock).endField("children", 1);
    inOrder.verify(readConsumerMock).endMessage();

    Mockito.verifyNoMoreInteractions(readConsumerMock);
  }