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 testCloseWithTimeout() throws IOException {
    Duration closeTimeout = Duration.millis(2000L);
    JmsIO.Read spec =
        JmsIO.read()
            .withConnectionFactory(connectionFactory)
            .withUsername(USERNAME)
            .withPassword(PASSWORD)
            .withQueue(QUEUE)
            .withCloseTimeout(closeTimeout);

    JmsIO.UnboundedJmsSource source = new JmsIO.UnboundedJmsSource(spec);

    ScheduledExecutorService mockScheduledExecutorService =
        Mockito.mock(ScheduledExecutorService.class);
    ExecutorOptions options = PipelineOptionsFactory.as(ExecutorOptions.class);
    options.setScheduledExecutorService(mockScheduledExecutorService);
    ArgumentCaptor<Runnable> runnableArgumentCaptor = ArgumentCaptor.forClass(Runnable.class);
    when(mockScheduledExecutorService.schedule(
            runnableArgumentCaptor.capture(), anyLong(), any(TimeUnit.class)))
        .thenReturn(null /* unused */);

    JmsIO.UnboundedJmsReader reader = source.createReader(options, null);
    reader.start();
    assertFalse(getDiscardedValue(reader));
    reader.close();
    assertFalse(getDiscardedValue(reader));
    verify(mockScheduledExecutorService)
        .schedule(any(Runnable.class), eq(closeTimeout.getMillis()), eq(TimeUnit.MILLISECONDS));
    runnableArgumentCaptor.getValue().run();
    assertTrue(getDiscardedValue(reader));
    verifyNoMoreInteractions(mockScheduledExecutorService);
  }