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

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

    final Schema badSchema = SchemaBuilder.struct()
        .field("FIELD", SchemaBuilder.OPTIONAL_STRING_SCHEMA)
        .optional()
        .build();

    final Struct badData = new Struct(badSchema);
    badData.put("FIELD", "fubar");

    final ConnectDataTranslator connectToKsqlTranslator = new ConnectDataTranslator(schema);

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

    // Then:
    assertThat(e.getMessage(), containsString(Schema.Type.STRING.getName()));
    assertThat(e.getMessage(), containsString(Schema.Type.INT32.getName()));
    assertThat(e.getMessage(), containsString("FIELD"));
  }