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

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

    final Schema rowSchema = SchemaBuilder
        .struct()
        .field("STRUCT", structSchema)
        .optional()
        .build();

    final Schema dataStructSchema = SchemaBuilder
        .struct()
        .field("INT", SchemaBuilder.OPTIONAL_STRING_SCHEMA)
        .optional()
        .build();

    final Schema dataRowSchema = SchemaBuilder
        .struct()
        .field("STRUCT", dataStructSchema)
        .optional()
        .build();

    final Struct connectStruct = new Struct(dataRowSchema);
    final Struct structColumn = new Struct(dataStructSchema);
    structColumn.put("INT", "123");
    connectStruct.put("STRUCT", structColumn);

    final ConnectDataTranslator connectToKsqlTranslator = new ConnectDataTranslator(rowSchema);

    // When:
    final DataException e = assertThrows(
        DataException.class,
        () -> connectToKsqlTranslator.toKsqlRow(dataRowSchema, connectStruct)
    );

    // Then:
    assertThat(e.getMessage(), containsString(Schema.Type.INT32.getName()));
    assertThat(e.getMessage(), containsString(Schema.Type.STRING.getName()));
    assertThat(e.getMessage(), containsString("STRUCT->INT"));
  }