@Override
  public T deserialize(final String topic, final byte[] data) {
    final List<?> values = inner.deserialize(topic, data);
    if (values == null) {
      return null;
    }

    SerdeUtils.throwOnColumnCountMismatch(numColumns, values.size(), false, topic);

    return factory.apply(values);
  }
@Test
  public void shouldThrowOnDeserializeOnColumnCountMismatch() {
    // Given:
    givenInnerDeserializerReturns(ImmutableList.of("too", "many", "columns"));

    // When:
    final Exception e = assertThrows(
        SerializationException.class,
        () -> deserializer.deserialize("topicName", SERIALIZED)
    );

    // Then:
    assertThat(e.getMessage(), is("Column count mismatch on deserialization."
        + " topic: topicName"
        + ", expected: 2"
        + ", got: 3"
    ));
  }