@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 shouldRejectUnmatchingDecimalSchema() {
    // Given:
    final Schema ksqlSchema = DecimalUtil.builder(4, 2).build();
    final Schema topicSchema = DecimalUtil.builder(6, 3).build();

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

    // Then:
    final ArithmeticException e = assertThrows(
        ArithmeticException.class,
        () -> translator.toKsqlRow(topicSchema, new BigDecimal("123.456")));
    assertThat(
        e.getMessage(),
        containsString("Numeric field overflow: A field with precision 4 and scale 2 must round to an absolute value less than 10^2"));
  }