@Override
  public Object toKsqlRow(final Schema connectSchema, final Object connectObject) {
    final Object avroCompatibleRow = innerTranslator.toKsqlRow(connectSchema, connectObject);
    if (avroCompatibleRow == null) {
      return null;
    }

    return ConnectSchemas.withCompatibleSchema(ksqlSchema, avroCompatibleRow);
  }
@Test
  public void shouldRejectConversionsRequiringRounding() {
    // Given:
    final Schema ksqlSchema = DecimalUtil.builder(3, 0).build();
    final Schema topicSchema = DecimalUtil.builder(4, 1).build();

    // When:
    final AvroDataTranslator translator =
        new AvroDataTranslator(ksqlSchema, AvroProperties.DEFAULT_AVRO_SCHEMA_FULL_NAME);

    // Then:
    final KsqlException e = assertThrows(
        KsqlException.class,
        () -> translator.toKsqlRow(topicSchema, new BigDecimal("123.4")));
    assertThat(
        e.getMessage(),
        containsString("Cannot fit decimal '123.4' into DECIMAL(3, 0) without rounding. (Requires 4,1)"));
  }