@Override
  public Object toKsqlRow(final Schema connectSchema, final Object connectData) {
    if (connectData == null) {
      return null;
    }

    return toKsqlValue(schema, connectSchema, connectData, "");
  }
@Test
  public void shouldTranslateArrayOfStructs() {
    // Given:
    final Schema innerSchema = SchemaBuilder
        .struct()
        .field("FIELD", Schema.OPTIONAL_INT32_SCHEMA)
        .optional()
        .build();

    final Schema rowSchema = SchemaBuilder
        .struct()
        .field("ARRAY", SchemaBuilder
            .array(innerSchema)
            .optional()
            .build())
        .build();

    final Struct connectStruct = new Struct(rowSchema);
    final Struct inner1 = new Struct(innerSchema);
    inner1.put("FIELD", 123);
    final Struct inner2 = new Struct(innerSchema);
    inner2.put("FIELD", 456);
    connectStruct.put("ARRAY", Arrays.asList(inner1, inner2));

    final ConnectDataTranslator connectToKsqlTranslator = new ConnectDataTranslator(rowSchema);

    // When:
    final Struct row = (Struct) connectToKsqlTranslator.toKsqlRow(rowSchema, connectStruct);

    // Then:
    assertThat(row.get("ARRAY"), instanceOf(List.class));

    final List<Struct> array = (List<Struct>)row.get("ARRAY");
    assertThat(array.get(0).get("FIELD"), equalTo(123));
    assertThat(array.get(1).get("FIELD"), equalTo(456));
  }