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

    final T single = extractOnlyColumn(values, topic);
    return inner.serialize(topic, single);
  }
@Test
  public void shouldSerializeNewStyle() {
    // Given:
    final List<?> values = ImmutableList.of(DATA);

    // When:
    final byte[] result = serializer.serialize(TOPIC, HEADERS, values);

    // Then:
    verify(inner).serialize(TOPIC, HEADERS, DATA);
    assertThat(result, is(SERIALIZED));
  }