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

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

    // Then:
    assertThat(
        translator.toKsqlRow(topicSchema, new BigDecimal("12.1")),
        is(new BigDecimal("12.10")));
  }