public static Read read() {
    return new AutoValue_MqttIO_Read.Builder()
        .setMaxReadTime(null)
        .setMaxNumRecords(Long.MAX_VALUE)
        .build();
  }
@Test(timeout = 30 * 1000)
  public void testReceiveWithTimeoutAndNoData() throws Exception {
    pipeline.apply(
        MqttIO.read()
            .withConnectionConfiguration(
                MqttIO.ConnectionConfiguration.create("tcp://localhost:" + port, "READ_TOPIC")
                    .withClientId("READ_PIPELINE"))
            .withMaxReadTime(Duration.standardSeconds(2)));

    // should stop before the test timeout
    pipeline.run();
  }