public static Read<Solace.Record> read() {
    return new Read<Solace.Record>(
        Read.Configuration.<Solace.Record>builder()
            .setTypeDescriptor(TypeDescriptor.of(Solace.Record.class))
            .setParseFn(SolaceRecordMapper::map)
            .setTimestampFn(SENDER_TIMESTAMP_FUNCTION)
            .setDeduplicateRecords(DEFAULT_DEDUPLICATE_RECORDS)
            .setWatermarkIdleDurationThreshold(DEFAULT_WATERMARK_IDLE_DURATION_THRESHOLD));
  }
@Test
  public void testReadWithCoderAndParseFnAndTimestampFn() {
    // Broker that creates input data
    MockSessionService mockClientService =
        new MockSessionService(
            index -> {
              List<BytesXMLMessage> messages =
                  ImmutableList.of(
                      SolaceDataUtils.getBytesXmlMessage("payload_test0", "450"),
                      SolaceDataUtils.getBytesXmlMessage("payload_test1", "451"),
                      SolaceDataUtils.getBytesXmlMessage("payload_test2", "452"));
              return getOrNull(index, messages);
            },
            3);
    SessionServiceFactory fakeSessionServiceFactory =
        new MockSessionServiceFactory(mockClientService);

    // Expected data
    List<SimpleRecord> expected = new ArrayList<>();
    expected.add(new SimpleRecord("payload_test0", "450"));
    expected.add(new SimpleRecord("payload_test1", "451"));
    expected.add(new SimpleRecord("payload_test2", "452"));

    // Run the pipeline
    PCollection<SimpleRecord> events =
        pipeline.apply(
            "Read from Solace",
            SolaceIO.read(
                    TypeDescriptor.of(SimpleRecord.class),
                    input ->
                        new SimpleRecord(
                            new String(input.getBytes(), StandardCharsets.UTF_8),
                            input.getApplicationMessageId()),
                    input -> Instant.ofEpochMilli(1708100477061L))
                .from(Solace.Queue.fromName("queue"))
                .withSempClientFactory(MockSempClientFactory.getDefaultMock())
                .withSessionServiceFactory(fakeSessionServiceFactory)
                .withMaxNumConnections(1));

    // Assert results
    PAssert.that(events).containsInAnyOrder(expected);
    pipeline.run();
  }