@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 shouldNotTransformOtherType() {
    // Given:
    final Schema schema = SchemaBuilder.struct()
        .field("f1", SchemaBuilder.OPTIONAL_STRING_SCHEMA)
        .field("f2", SchemaBuilder.OPTIONAL_INT32_SCHEMA)
        .field("f3", SchemaBuilder.OPTIONAL_INT64_SCHEMA)
        .build();
    final List<Integer> list = Collections.emptyList();

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

    // Then:
    assertThat(object, sameInstance(list));
  }