public static Read<JmsRecord> read() {
    return new AutoValue_JmsIO_Read.Builder<JmsRecord>()
        .setMaxNumRecords(Long.MAX_VALUE)
        .setCoder(SerializableCoder.of(JmsRecord.class))
        .setCloseTimeout(DEFAULT_CLOSE_TIMEOUT)
        .setRequiresDeduping(false)
        .setMessageMapper(
            new MessageMapper<JmsRecord>() {
              @Override
              public JmsRecord mapMessage(Message message) throws Exception {
                TextMessage textMessage = (TextMessage) message;
                Map<String, Object> properties = new HashMap<>();
                @SuppressWarnings("rawtypes")
                Enumeration propertyNames = textMessage.getPropertyNames();
                while (propertyNames.hasMoreElements()) {
                  String propertyName = (String) propertyNames.nextElement();
                  properties.put(propertyName, textMessage.getObjectProperty(propertyName));
                }

                return new JmsRecord(
                    textMessage.getJMSMessageID(),
                    textMessage.getJMSTimestamp(),
                    textMessage.getJMSCorrelationID(),
                    textMessage.getJMSReplyTo(),
                    textMessage.getJMSDestination(),
                    textMessage.getJMSDeliveryMode(),
                    textMessage.getJMSRedelivered(),
                    textMessage.getJMSType(),
                    textMessage.getJMSExpiration(),
                    textMessage.getJMSPriority(),
                    properties,
                    textMessage.getText());
              }
            })
        .build();
  }
@Test
  public void testCheckpointMarkSafety() throws Exception {

    final int messagesToProcess = 100;

    // we are using no prefetch here
    // prefetch is an ActiveMQ feature: to make efficient use of network resources the broker
    // utilizes a 'push' model to dispatch messages to consumers. However, in the case of our
    // test, it means that we can have some latency between the receiveNoWait() method used by
    // the consumer and the prefetch buffer populated by the broker. Using a prefetch to 0 means
    // that the consumer will poll for message, which is exactly what we want for the test.
    // We are also sending message acknowledgements synchronously to ensure that they are
    // processed before any subsequent assertions.
    Connection connection =
        connectionFactoryWithSyncAcksAndWithoutPrefetch.createConnection(USERNAME, PASSWORD);
    connection.start();
    Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);

    // Fill the queue with messages
    MessageProducer producer = session.createProducer(session.createQueue(QUEUE));
    for (int i = 0; i < messagesToProcess; i++) {
      producer.send(session.createTextMessage("test " + i));
    }
    producer.close();
    session.close();
    connection.close();

    Function jmsMessageAck = getJmsMessageAck(this.connectionFactoryClass);

    // create a JmsIO.Read with a decorated ConnectionFactory which will introduce a delay in
    // sending
    // acknowledgements - this should help uncover threading issues around checkpoint management.
    JmsIO.Read spec =
        JmsIO.read()
            .withConnectionFactory(
                withSlowAcks(connectionFactoryWithSyncAcksAndWithoutPrefetch, jmsMessageAck))
            .withUsername(USERNAME)
            .withPassword(PASSWORD)
            .withQueue(QUEUE);
    JmsIO.UnboundedJmsSource source = new JmsIO.UnboundedJmsSource(spec);
    JmsIO.UnboundedJmsReader reader = source.createReader(PipelineOptionsFactory.create(), null);

    // start the reader and move to the first record
    assertTrue(reader.start());

    // consume half the messages (NB: start already consumed the first message)
    for (int i = 0; i < (messagesToProcess / 2) - 1; i++) {
      assertTrue(reader.advance());
    }

    // the messages are still pending in the queue (no ACK yet)
    assertEquals(messagesToProcess, count(QUEUE));

    // we finalize the checkpoint for the already-processed messages while simultaneously consuming
    // the remainder of
    // messages from the queue
    Thread runner =
        new Thread(
            () -> {
              try {
                for (int i = 0; i < messagesToProcess / 2; i++) {
                  assertTrue(reader.advance());
                }
              } catch (IOException ex) {
                throw new RuntimeException(ex);
              }
            });
    runner.start();
    reader.getCheckpointMark().finalizeCheckpoint();

    // Concurrency issues would cause an exception to be thrown before this method exits, failing
    // the test
    runner.join();
  }