@Nullable
  @Override
  public GenericRow decode(byte[] payload, GenericRow destination) {
    try {
      destination = (GenericRow) _decodeMethod.invoke(null, payload, destination);
    } catch (Exception e) {
      throw new RuntimeException(e);
    }
    return destination;
  }
@Test
  public void testCompositeMessage()
      throws Exception {
    ProtoBufCodeGenMessageDecoder messageDecoder = setupDecoder("composite_types.jar",
        "org.apache.pinot.plugin.inputformat.protobuf.CompositeTypes$CompositeMessage",
        ImmutableSet.of("test_message", "sample_record"));
    Map<String, Object> inputRecord = createComplexTypeRecord();
    GenericRow destination = new GenericRow();
    ComplexTypes.TestMessage testMessage = getComplexTypeObject(inputRecord);
    Sample.SampleRecord sampleRecord = getSampleRecordMessage();
    CompositeTypes.CompositeMessage compositeMessage =
        CompositeTypes.CompositeMessage.newBuilder().setTestMessage(testMessage).setSampleRecord(sampleRecord).build();
    messageDecoder.decode(compositeMessage.toByteArray(), destination);

    assertNotNull(destination.getValue("test_message"));

    for (String col : getSourceFieldsForComplexType()) {
      assertNotNull(((Map<String, Object>) destination.getValue("test_message")).get(col));
    }

    assertNotNull(((Map<String, Object>) destination.getValue("sample_record")).get("email"));
    assertNotNull(((Map<String, Object>) destination.getValue("sample_record")).get("name"));
    assertNotNull(((Map<String, Object>) destination.getValue("sample_record")).get("id"));

    assertEquals(((Map<String, Object>) destination.getValue("sample_record")).get("email"), "foobar@hello.com");
    assertEquals(((Map<String, Object>) destination.getValue("sample_record")).get("name"), "Alice");
    assertEquals(((Map<String, Object>) destination.getValue("sample_record")).get("id"), 18);
  }