public static Read read() {
    return new AutoValue_MqttIO_Read.Builder()
        .setMaxReadTime(null)
        .setMaxNumRecords(Long.MAX_VALUE)
        .build();
  }
@Test(timeout = 60 * 1000)
  @Ignore("https://github.com/apache/beam/issues/18723 Test timeout failure.")
  public void testReadNoClientId() throws Exception {
    final String topicName = "READ_TOPIC_NO_CLIENT_ID";
    Read mqttReader =
        MqttIO.read()
            .withConnectionConfiguration(
                MqttIO.ConnectionConfiguration.create("tcp://localhost:" + port, topicName))
            .withMaxNumRecords(10);
    PCollection<byte[]> output = pipeline.apply(mqttReader);
    PAssert.that(output)
        .containsInAnyOrder(
            "This is test 0".getBytes(StandardCharsets.UTF_8),
            "This is test 1".getBytes(StandardCharsets.UTF_8),
            "This is test 2".getBytes(StandardCharsets.UTF_8),
            "This is test 3".getBytes(StandardCharsets.UTF_8),
            "This is test 4".getBytes(StandardCharsets.UTF_8),
            "This is test 5".getBytes(StandardCharsets.UTF_8),
            "This is test 6".getBytes(StandardCharsets.UTF_8),
            "This is test 7".getBytes(StandardCharsets.UTF_8),
            "This is test 8".getBytes(StandardCharsets.UTF_8),
            "This is test 9".getBytes(StandardCharsets.UTF_8));

    // produce messages on the brokerService in another thread
    // This thread prevents to block the pipeline waiting for new messages
    MQTT client = new MQTT();
    client.setHost("tcp://localhost:" + port);
    final BlockingConnection publishConnection = client.blockingConnection();
    publishConnection.connect();
    Thread publisherThread =
        new Thread(
            () -> {
              try {
                LOG.info(
                    "Waiting pipeline connected to the MQTT broker before sending "
                        + "messages ...");
                boolean pipelineConnected = false;
                while (!pipelineConnected) {
                  Thread.sleep(1000);
                  for (Connection connection : brokerService.getBroker().getClients()) {
                    if (!connection.getConnectionId().isEmpty()) {
                      pipelineConnected = true;
                    }
                  }
                }
                for (int i = 0; i < 10; i++) {
                  publishConnection.publish(
                      topicName,
                      ("This is test " + i).getBytes(StandardCharsets.UTF_8),
                      QoS.EXACTLY_ONCE,
                      false);
                }
              } catch (Exception e) {
                // nothing to do
              }
            });
    publisherThread.start();
    pipeline.run();

    publishConnection.disconnect();
    publisherThread.join();
  }