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

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

    final Schema rowSchema = SchemaBuilder
        .struct()
        .field("MAP", SchemaBuilder
            .map(Schema.STRING_SCHEMA, 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("MAP", ImmutableMap.of("k1", inner1, "k2", inner2));

    final ConnectDataTranslator connectToKsqlTranslator = new ConnectDataTranslator(rowSchema);

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


    assertThat(row.get("MAP"), instanceOf(Map.class));

    final Map<String, Struct> map = (Map<String, Struct>)row.get("MAP");
    assertThat(map.get("k1").get("FIELD"), equalTo(123));
    assertThat(map.get("k2").get("FIELD"), equalTo(456));
  }