@Test public void testFromConnectComplex() { Schema schema = SchemaBuilder.struct() .field("int8", SchemaBuilder.int8().defaultValue((byte) 2).doc("int8 field").build()) .field("int16", Schema.INT16_SCHEMA) .field("int32", Schema.INT32_SCHEMA) .field("int64", Schema.INT64_SCHEMA) .field("float32", Schema.FLOAT32_SCHEMA) .field("float64", Schema.FLOAT64_SCHEMA) .field("boolean", Schema.BOOLEAN_SCHEMA) .field("string", Schema.STRING_SCHEMA) .field("bytes", Schema.BYTES_SCHEMA) .field("array", SchemaBuilder.array(Schema.STRING_SCHEMA).build()) .field("map", SchemaBuilder.map(Schema.STRING_SCHEMA, Schema.INT32_SCHEMA).build()) .field("mapNonStringKeys", SchemaBuilder.map(Schema.INT32_SCHEMA, Schema.INT32_SCHEMA).build() ) .field("enum", SchemaBuilder.string() .name("Status") .optional() .parameter(ProtobufData.PROTOBUF_TYPE_ENUM, "Status") .parameter(ProtobufData.PROTOBUF_TYPE_ENUM_PREFIX + "ACTIVE", "0") .parameter(ProtobufData.PROTOBUF_TYPE_ENUM_PREFIX + "INACTIVE", "1") .build() ) .build(); Struct struct = new Struct(schema).put("int8", (byte) 12) .put("int16", (short) 12) .put("int32", 12) .put("int64", 12L) .put("float32", 12.2f) .put("float64", 12.2) .put("boolean", true) .put("string", "foo") .put("bytes", ByteBuffer.wrap("foo".getBytes())) .put("array", Arrays.asList("a", "b", "c")) .put("map", Collections.singletonMap("field", 1)) .put("mapNonStringKeys", Collections.singletonMap(1, 1)) .put("enum", "INACTIVE"); ProtobufSchemaAndValue convertedRecord = new ProtobufData().fromConnectData(schema, struct); ProtobufSchema protobufSchema = convertedRecord.getSchema(); DynamicMessage message = (DynamicMessage) convertedRecord.getValue(); MessageElement messageElem = (MessageElement) protobufSchema.rawSchema().getTypes().get(0); FieldElement fieldElem = messageElem.getFields().get(0); assertEquals("int8", fieldElem.getName()); assertEquals("int32", fieldElem.getType()); fieldElem = messageElem.getFields().get(1); assertEquals("int16", fieldElem.getName()); assertEquals("int32", fieldElem.getType()); fieldElem = messageElem.getFields().get(2); assertEquals("int32", fieldElem.getName()); assertEquals("int32", fieldElem.getType()); fieldElem = messageElem.getFields().get(3); assertEquals("int64", fieldElem.getName()); assertEquals("int64", fieldElem.getType()); fieldElem = messageElem.getFields().get(4); assertEquals("float32", fieldElem.getName()); assertEquals("float", fieldElem.getType()); fieldElem = messageElem.getFields().get(5); assertEquals("float64", fieldElem.getName()); assertEquals("double", fieldElem.getType()); fieldElem = messageElem.getFields().get(6); assertEquals("boolean", fieldElem.getName()); assertEquals("bool", fieldElem.getType()); fieldElem = messageElem.getFields().get(7); assertEquals("string", fieldElem.getName()); assertEquals("string", fieldElem.getType()); fieldElem = messageElem.getFields().get(8); assertEquals("bytes", fieldElem.getName()); assertEquals("bytes", fieldElem.getType()); fieldElem = messageElem.getFields().get(9); assertEquals("array", fieldElem.getName()); assertEquals("string", fieldElem.getType()); fieldElem = messageElem.getFields().get(10); assertEquals("map", fieldElem.getName()); assertEquals("ConnectDefault2Entry", fieldElem.getType()); fieldElem = messageElem.getFields().get(11); assertEquals("mapNonStringKeys", fieldElem.getName()); assertEquals("ConnectDefault3Entry", fieldElem.getType()); fieldElem = messageElem.getFields().get(12); assertEquals("enum", fieldElem.getName()); assertEquals("Status", fieldElem.getType()); Descriptors.FieldDescriptor fieldDescriptor = message.getDescriptorForType() .findFieldByName("int8"); assertEquals(12, message.getField(fieldDescriptor)); fieldDescriptor = message.getDescriptorForType().findFieldByName("int16"); assertEquals(12, message.getField(fieldDescriptor)); fieldDescriptor = message.getDescriptorForType().findFieldByName("int32"); assertEquals(12, message.getField(fieldDescriptor)); fieldDescriptor = message.getDescriptorForType().findFieldByName("int64"); assertEquals(12L, message.getField(fieldDescriptor)); fieldDescriptor = message.getDescriptorForType().findFieldByName("float32"); assertEquals(12.2f, message.getField(fieldDescriptor)); fieldDescriptor = message.getDescriptorForType().findFieldByName("float64"); assertEquals(12.2, message.getField(fieldDescriptor)); fieldDescriptor = message.getDescriptorForType().findFieldByName("boolean"); assertEquals(true, message.getField(fieldDescriptor)); fieldDescriptor = message.getDescriptorForType().findFieldByName("string"); assertEquals("foo", message.getField(fieldDescriptor)); fieldDescriptor = message.getDescriptorForType().findFieldByName("bytes"); assertEquals(ByteString.copyFrom("foo".getBytes()), message.getField(fieldDescriptor)); fieldDescriptor = message.getDescriptorForType().findFieldByName("array"); assertEquals(Arrays.asList("a", "b", "c"), message.getField(fieldDescriptor)); fieldDescriptor = message.getDescriptorForType().findFieldByName("map"); Object value = message.getField(fieldDescriptor); DynamicMessage dynamicMessage = ((List<DynamicMessage>) value).get(0); fieldDescriptor = dynamicMessage.getDescriptorForType().findFieldByName("key"); assertEquals("field", dynamicMessage.getField(fieldDescriptor)); fieldDescriptor = dynamicMessage.getDescriptorForType().findFieldByName("value"); assertEquals(1, dynamicMessage.getField(fieldDescriptor)); fieldDescriptor = message.getDescriptorForType().findFieldByName("mapNonStringKeys"); value = message.getField(fieldDescriptor); dynamicMessage = ((List<DynamicMessage>) value).get(0); fieldDescriptor = dynamicMessage.getDescriptorForType().findFieldByName("key"); assertEquals(1, dynamicMessage.getField(fieldDescriptor)); fieldDescriptor = dynamicMessage.getDescriptorForType().findFieldByName("value"); assertEquals(1, dynamicMessage.getField(fieldDescriptor)); fieldDescriptor = message.getDescriptorForType().findFieldByName("enum"); EnumValueDescriptor enumValueDescriptor = (EnumValueDescriptor) message.getField(fieldDescriptor); assertEquals("INACTIVE", enumValueDescriptor.getName()); }
public ProtobufSchemaAndValue fromConnectData(Schema schema, Object value) { ProtobufSchema protobufSchema = fromConnectSchema(schema); Object ctx = null; if (schema != null) { String name = schema.name(); if (name == null) { name = DEFAULT_SCHEMA_NAME + "1"; } ctx = protobufSchema.toDescriptor(name); } return new ProtobufSchemaAndValue( protobufSchema, fromConnectData(ctx, schema, "", value, protobufSchema) ); }
ProtobufData { public ProtobufSchemaAndValue fromConnectData(Schema schema, Object value) { ProtobufSchema protobufSchema = fromConnectSchema(schema); Object ctx = null; if (schema != null) { String name = schema.name(); if (name == null) { name = DEFAULT_SCHEMA_NAME + "1"; } ctx = protobufSchema.toDescriptor(name); } return new ProtobufSchemaAndValue( protobufSchema, fromConnectData(ctx, schema, "", value, protobufSchema) ); } }
ProtobufData { public ProtobufSchemaAndValue fromConnectData(Schema schema, Object value) { ProtobufSchema protobufSchema = fromConnectSchema(schema); Object ctx = null; if (schema != null) { String name = schema.name(); if (name == null) { name = DEFAULT_SCHEMA_NAME + "1"; } ctx = protobufSchema.toDescriptor(name); } return new ProtobufSchemaAndValue( protobufSchema, fromConnectData(ctx, schema, "", value, protobufSchema) ); } ProtobufData();  ProtobufData(int cacheSize);  ProtobufData(ProtobufDataConfig protobufDataConfig); }
ProtobufData { public ProtobufSchemaAndValue fromConnectData(Schema schema, Object value) { ProtobufSchema protobufSchema = fromConnectSchema(schema); Object ctx = null; if (schema != null) { String name = schema.name(); if (name == null) { name = DEFAULT_SCHEMA_NAME + "1"; } ctx = protobufSchema.toDescriptor(name); } return new ProtobufSchemaAndValue( protobufSchema, fromConnectData(ctx, schema, "", value, protobufSchema) ); } ProtobufData();  ProtobufData(int cacheSize);  ProtobufData(ProtobufDataConfig protobufDataConfig); ProtobufSchemaAndValue fromConnectData(Schema schema, Object value); ProtobufSchema fromConnectSchema(Schema schema); SchemaAndValue toConnectData(ProtobufSchema protobufSchema, Message message); Schema toConnectSchema(ProtobufSchema schema); }
ProtobufData { public ProtobufSchemaAndValue fromConnectData(Schema schema, Object value) { ProtobufSchema protobufSchema = fromConnectSchema(schema); Object ctx = null; if (schema != null) { String name = schema.name(); if (name == null) { name = DEFAULT_SCHEMA_NAME + "1"; } ctx = protobufSchema.toDescriptor(name); } return new ProtobufSchemaAndValue( protobufSchema, fromConnectData(ctx, schema, "", value, protobufSchema) ); } ProtobufData();  ProtobufData(int cacheSize);  ProtobufData(ProtobufDataConfig protobufDataConfig); ProtobufSchemaAndValue fromConnectData(Schema schema, Object value); ProtobufSchema fromConnectSchema(Schema schema); SchemaAndValue toConnectData(ProtobufSchema protobufSchema, Message message); Schema toConnectSchema(ProtobufSchema schema); static final String NAMESPACE; static final String DEFAULT_SCHEMA_NAME; static final String MAP_ENTRY_SUFFIX; static final String KEY_FIELD; static final String VALUE_FIELD; static final String PROTOBUF_TYPE_ENUM; static final String PROTOBUF_TYPE_ENUM_PREFIX; static final String PROTOBUF_TYPE_UNION; static final String PROTOBUF_TYPE_UNION_PREFIX; static final String PROTOBUF_TYPE_TAG; static final String GOOGLE_PROTOBUF_PACKAGE; static final String GOOGLE_PROTOBUF_TIMESTAMP_NAME; static final String GOOGLE_PROTOBUF_TIMESTAMP_FULL_NAME; static final String GOOGLE_PROTOBUF_TIMESTAMP_LOCATION; }