public static Read<byte[], byte[]> readBytes() {
    return KafkaIO.<byte[], byte[]>read()
        .withKeyDeserializer(ByteArrayDeserializer.class)
        .withValueDeserializer(ByteArrayDeserializer.class);
  }
@Test
  public void testSourceWithExplicitPartitionsDisplayData() {
    KafkaIO.Read<byte[], byte[]> read =
        KafkaIO.readBytes()
            .withBootstrapServers("myServer1:9092,myServer2:9092")
            .withTopicPartitions(
                ImmutableList.of(new TopicPartition("test", 5), new TopicPartition("test", 6)))
            .withConsumerFactoryFn(
                new ConsumerFactoryFn(
                    Lists.newArrayList("test"),
                    10,
                    10,
                    OffsetResetStrategy.EARLIEST)); // 10 partitions

    DisplayData displayData = DisplayData.from(read);

    assertThat(displayData, hasDisplayItem("topicPartitions", "test-5,test-6"));
    assertThat(displayData, hasDisplayItem("enable.auto.commit", false));
    assertThat(displayData, hasDisplayItem("bootstrap.servers", "myServer1:9092,myServer2:9092"));
    assertThat(displayData, hasDisplayItem("auto.offset.reset", "latest"));
    assertThat(displayData, hasDisplayItem("receive.buffer.bytes", 524288));
  }