@Override
  public T deserialize(final String topic, final byte[] bytes) {
    try {
      final SchemaAndValue schemaAndValue = converter.toConnectData(topic, bytes);

      final Object val = translator.toKsqlRow(schemaAndValue.schema(), schemaAndValue.value());

      return SerdeUtils.castToTargetType(val, targetType);
    } catch (final Exception e) {
      throw new SerializationException(
          "Error deserializing message from topic: " + topic, e);
    }
  }
@Test
  public void shouldDeserializeRecordsCorrectly() {
    // When:
    final String deserialized = connectDeserializer.deserialize(TOPIC, BYTES);

    // Then:
    verify(converter, times(1)).toConnectData(TOPIC, BYTES);
    verify(dataTranslator, times(1)).toKsqlRow(schema, value);
    assertThat(deserialized, is("some-data"));
  }