ProtobufData cannot convert back google.protobuf.Struct

[jorgebo10]

Hi, i need to understand if it is possible to serialize and deserialize google.protobuf.Struct so in order to do that i added some code to ProtobufConverterTest.java , however it seems that it is not possible. Am i doing something wrong?

Thanks,
Jorge

@Test
   public void testToConnectDataForProtobufStruct() throws Exception {
       Value value = Value.newBuilder()
               .setStringValue("mi_value")
               .build();

       com.google.protobuf.Struct struct = com.google.protobuf.Struct.newBuilder()
               .putFields("key", value)
               .build();

       Map<String, Object> configs = new HashMap<>();
       configs.put(KafkaProtobufSerializerConfig.SCHEMA_REGISTRY_URL_CONFIG, "bogus");
       configs.put(ProtobufDataConfig.ENHANCED_PROTOBUF_SCHEMA_SUPPORT_CONFIG, true);
       converter.configure(configs, false);
       // extra bytes for message index
       final byte[] input = concat(new byte[]{0, 0, 0, 0, 1, 0},
               struct.toByteArray());
       schemaRegistry.register("my-topic-value", getSchema(com.google.protobuf.Struct.getDescriptor()));
       SchemaAndValue result = converter.toConnectData("my-topic", input);

       Struct data1 = (Struct) result.value();
       Map<String, Object> map = (Map<String, Object>) data1.get("fields");
       Struct data2 = (Struct) map.get("key");
      Struct data3 = (Struct) data2.get("kind_0");
       assertEquals("mi_value", data3.get("string_value"));

       byte[] output = converter.fromConnectData("my-topic-value", result.schema(), result.value());

      assertEquals(input, output)
   }

Getting this error

java.lang.StackOverflowError
	at com.google.protobuf.DescriptorProtos$DescriptorProto$Builder.<init>(DescriptorProtos.java:6913)
	at com.google.protobuf.DescriptorProtos$DescriptorProto$Builder.<init>(DescriptorProtos.java:6894)
	at com.google.protobuf.DescriptorProtos$DescriptorProto.toBuilder(DescriptorProtos.java:6882)
	at com.google.protobuf.DescriptorProtos$DescriptorProto.newBuilder(DescriptorProtos.java:6874)
	at io.confluent.kafka.schemaregistry.protobuf.dynamic.MessageDefinition$Builder.<init>(MessageDefinition.java:237)
	at io.confluent.kafka.schemaregistry.protobuf.dynamic.MessageDefinition.newBuilder(MessageDefinition.java:43)
	at io.confluent.connect.protobuf.ProtobufData.messageDefinitionFromConnectSchema(ProtobufData.java:728)
	at io.confluent.connect.protobuf.ProtobufData.fieldDefinitionFromConnectSchema(ProtobufData.java:814)