@Override
  public Object toConnectRow(final Object ksqlData) {
    if (!(ksqlData instanceof Struct)) {
      return ksqlData;
    }

    final Schema schema = getSchema();
    final Struct struct = new Struct(schema);
    Struct originalData = (Struct) ksqlData;
    Schema originalSchema = originalData.schema();
    if (originalSchema.name() == null && schema.name() != null) {
      originalSchema = AvroSchemas.getAvroCompatibleConnectSchema(
          originalSchema, schema.name()
      );
      originalData = ConnectSchemas.withCompatibleRowSchema(originalData, originalSchema);
    }

    validate(originalSchema, schema);

    copyStruct(originalData, originalSchema, struct, schema);

    return struct;
  }
@Test
  public void shouldTransformStructWithMapOfStructs() {
    // Given:
    final Schema innerStructSchemaWithoutOptional = getInnerStructSchema(false);
    final Schema innerStructSchemaWithOptional = getInnerStructSchema(true);

    // Physical Schema retrieved from SR
    final Schema schema = SchemaBuilder.struct()
        .field("string_field", SchemaBuilder.STRING_SCHEMA)
        .field("map_field", SchemaBuilder.map(Schema.STRING_SCHEMA, innerStructSchemaWithoutOptional))
        .build();

    // Logical Schema created by Ksql
    final Schema ORIGINAL_SCHEMA = SchemaBuilder.struct()
        .field("string_field", SchemaBuilder.OPTIONAL_STRING_SCHEMA)
        .field("map_field", SchemaBuilder.map(Schema.OPTIONAL_STRING_SCHEMA, innerStructSchemaWithOptional))
        .optional()
        .build();

    final Struct struct = new Struct(ORIGINAL_SCHEMA)
        .put("string_field", "abc")
        .put("map_field", ImmutableMap.of(
            "key1", getNestedData(innerStructSchemaWithOptional),
            "key2", getNestedData(innerStructSchemaWithOptional)));

    final Map<String, Object> mapWithoutOptional = ImmutableMap.of(
        "key1", getNestedData(innerStructSchemaWithoutOptional),
        "key2", getNestedData(innerStructSchemaWithoutOptional));

    // When:
    final Object object = new AvroSRSchemaDataTranslator(schema).toConnectRow(struct);

    // Then:
    assertThat(object, instanceOf(Struct.class));
    assertThat(((Struct) object).schema(), sameInstance(schema));
    assertThat(((Struct) object).get("string_field"), is("abc"));
    assertThat(((Struct) object).get("map_field"), equalTo(mapWithoutOptional));
  }