@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 shouldTransformStructWithNestedStructs() {
    // Given:
    final Schema innerStructSchemaWithoutOptional = getInnerStructSchema(false);
    final Schema innerStructSchemaWithOptional = getInnerStructSchema(true);

    Struct innerInnerStructWithOptional = getNestedData(innerStructSchemaWithOptional);
    Struct innerInnerStructWithoutOptional = getNestedData(innerStructSchemaWithoutOptional);

    final Schema structSchemaInnerWithOptional = getStructSchemaWithNestedStruct(innerStructSchemaWithOptional, true);
    final Schema structSchemaInnerWithOutOptional = getStructSchemaWithNestedStruct(innerStructSchemaWithoutOptional, false);

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

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

    Struct innerStructWithoutOptional = getInnerStructData(structSchemaInnerWithOutOptional, innerInnerStructWithoutOptional);
    Struct innerStructWithOptional = getInnerStructData(structSchemaInnerWithOptional, innerInnerStructWithOptional);

    final Struct struct = new Struct(ORIGINAL_SCHEMA)
        .put("string_field", "abc")
        .put("struct_field", innerStructWithOptional);

    // 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("struct_field"), equalTo(innerStructWithoutOptional));
  }