@Override
  public byte[] serialize(final String topicName, final T record) {
    return serializer.get().serialize(topicName, record);
  }
@Test
  public void shouldUseAThreadLocalSerializer() throws InterruptedException {
    final List<Serializer<GenericRow>> serializers = new LinkedList<>();

    final ThreadLocalSerializer<GenericRow> serializer = new ThreadLocalSerializer<>(
        () -> {
          final Serializer<GenericRow> local = mock(Serializer.class);
          serializers.add(local);
          expect(local.serialize(anyString(), anyObject(GenericRow.class)))
              .andReturn(new byte[32])
              .times(1);
          replay(local);
          return serializers.get(serializers.size() - 1);
        }
    );

    for (int i = 0; i < 3; i++) {
      final Thread t = new Thread(
          () -> serializer.serialize("foo", new GenericRow())
      );
      t.start();
      t.join();
      assertThat(serializers.size(), equalTo(i + 1));
      serializers.forEach(EasyMock::verify);
    }
  }