MongoDataConverter { public void addFieldSchema(Entry<String, BsonValue> keyValuesforSchema, SchemaBuilder builder) { String key = fieldNamer.fieldNameFor(keyValuesforSchema.getKey()); BsonType type = keyValuesforSchema.getValue().getBsonType(); switch (type) { case NULL: case STRING: case JAVASCRIPT: case OBJECT_ID: case DECIMAL128: builder.field(key, Schema.OPTIONAL_STRING_SCHEMA); break; case DOUBLE: builder.field(key, Schema.OPTIONAL_FLOAT64_SCHEMA); break; case BINARY: builder.field(key, Schema.OPTIONAL_BYTES_SCHEMA); break; case INT32: builder.field(key, Schema.OPTIONAL_INT32_SCHEMA); break; case INT64: builder.field(key, Schema.OPTIONAL_INT64_SCHEMA); break; case DATE_TIME: case TIMESTAMP: builder.field(key, org.apache.kafka.connect.data.Timestamp.builder().optional().build()); break; case BOOLEAN: builder.field(key, Schema.OPTIONAL_BOOLEAN_SCHEMA); break; case JAVASCRIPT_WITH_SCOPE: SchemaBuilder jswithscope = SchemaBuilder.struct().name(builder.name() + "." + key); jswithscope.field("code", Schema.OPTIONAL_STRING_SCHEMA); SchemaBuilder scope = SchemaBuilder.struct().name(jswithscope.name() + ".scope").optional(); BsonDocument jwsDocument = keyValuesforSchema.getValue().asJavaScriptWithScope().getScope().asDocument(); for (Entry<String, BsonValue> jwsDocumentKey : jwsDocument.entrySet()) { addFieldSchema(jwsDocumentKey, scope); } Schema scopeBuild = scope.build(); jswithscope.field("scope", scopeBuild).build(); builder.field(key, jswithscope); break; case REGULAR_EXPRESSION: SchemaBuilder regexwop = SchemaBuilder.struct().name(SCHEMA_NAME_REGEX).optional(); regexwop.field("regex", Schema.OPTIONAL_STRING_SCHEMA); regexwop.field("options", Schema.OPTIONAL_STRING_SCHEMA); builder.field(key, regexwop.build()); break; case DOCUMENT: SchemaBuilder builderDoc = SchemaBuilder.struct().name(builder.name() + "." + key).optional(); BsonDocument docs = keyValuesforSchema.getValue().asDocument(); for (Entry<String, BsonValue> doc : docs.entrySet()) { addFieldSchema(doc, builderDoc); } builder.field(key, builderDoc.build()); break; case ARRAY: if (keyValuesforSchema.getValue().asArray().isEmpty()) { if (sanitizeValue) { return; } switch (arrayEncoding) { case ARRAY: builder.field(key, SchemaBuilder.array(Schema.OPTIONAL_STRING_SCHEMA).optional().build()); break; case DOCUMENT: builder.field(key, SchemaBuilder.struct().name(builder.name() + "." + key).optional().build()); break; } } else { switch (arrayEncoding) { case ARRAY: BsonArray value = keyValuesforSchema.getValue().asArray(); BsonType valueType = value.get(0).getBsonType(); testType(builder, key, keyValuesforSchema.getValue(), valueType); builder.field(key, SchemaBuilder.array(subSchema(builder, key, valueType, value)).optional().build()); break; case DOCUMENT: final BsonArray array = keyValuesforSchema.getValue().asArray(); final SchemaBuilder arrayStructBuilder = SchemaBuilder.struct().name(builder.name() + "." + key).optional(); final Map<String, BsonValue> convertedArray = new HashMap<>(); for (int i = 0; i < array.size(); i++) { convertedArray.put(arrayElementStructName(i), array.get(i)); } convertedArray.entrySet().forEach(x -> addFieldSchema(x, arrayStructBuilder)); builder.field(key, arrayStructBuilder.build()); break; } } break; default: break; } } MongoDataConverter(ArrayEncoding arrayEncoding, FieldNamer<String> fieldNamer, boolean sanitizeValue);  MongoDataConverter(ArrayEncoding arrayEncoding); Struct convertRecord(Entry<String, BsonValue> keyvalueforStruct, Schema schema, Struct struct); void convertFieldValue(Entry<String, BsonValue> keyvalueforStruct, Struct struct, Schema schema); void addFieldSchema(Entry<String, BsonValue> keyValuesforSchema, SchemaBuilder builder); static final String SCHEMA_NAME_REGEX; }
@Test public void shouldCreateCorrectSchemaFromInsertJson() { for (Entry<String, BsonValue> entry : val.entrySet()) { converter.addFieldSchema(entry, builder); } Schema finalSchema = builder.build(); assertThat(finalSchema).isEqualTo( SchemaBuilder.struct().name("pub") .field("address", SchemaBuilder.struct().name("pub.address").optional() .field("building", Schema.OPTIONAL_STRING_SCHEMA) .field("floor", SchemaBuilder.struct().name("pub.address.floor").optional() .field("level", Schema.OPTIONAL_INT32_SCHEMA) .field("description", Schema.OPTIONAL_STRING_SCHEMA) .build()) .field("coord", SchemaBuilder.array(Schema.OPTIONAL_FLOAT64_SCHEMA).optional().build()) .field("street", Schema.OPTIONAL_STRING_SCHEMA) .field("zipcode", Schema.OPTIONAL_STRING_SCHEMA) .build()) .field("borough", Schema.OPTIONAL_STRING_SCHEMA) .field("cuisine", Schema.OPTIONAL_STRING_SCHEMA) .field("grades", SchemaBuilder.array(SchemaBuilder.struct().name("pub.grades").optional() .field("date", Timestamp.builder().optional().build()) .field("grade", Schema.OPTIONAL_STRING_SCHEMA) .field("score", Schema.OPTIONAL_INT32_SCHEMA) .build()) .optional() .build()) .field("name", Schema.OPTIONAL_STRING_SCHEMA) .field("restaurant_id", Schema.OPTIONAL_STRING_SCHEMA) .build()); }