static BsonTimestamp startAtTimestamp(Map<String, String> options) {
        String startAtValue = options.get(START_AT_OPTION);
        if (isNullOrEmpty(startAtValue)) {
            throw QueryException.error("startAt property is required for MongoDB stream. " + POSSIBLE_VALUES);
        }
        if ("now".equalsIgnoreCase(startAtValue)) {
            return MongoUtilities.bsonTimestampFromTimeMillis(System.currentTimeMillis());
        } else {
            try {
                return MongoUtilities.bsonTimestampFromTimeMillis(Long.parseLong(startAtValue));
            } catch (NumberFormatException e) {
                try {
                    return MongoUtilities.bsonTimestampFromTimeMillis(Instant.parse(startAtValue).toEpochMilli());
                } catch (DateTimeParseException ex) {
                    throw QueryException.error("Invalid startAt value: '" + startAtValue + "'. " + POSSIBLE_VALUES);
                }
            }
        }
    }
@Test
    public void parses_milis_startAt() {
        // given
        long time = System.currentTimeMillis();
        LocalDateTime timeDate = LocalDateTime.ofEpochSecond(time / 1000, 0, UTC);

        // when
        BsonTimestamp startAt = Options.startAtTimestamp(ImmutableMap.of(Options.START_AT_OPTION, String.valueOf(time)));

        // then
        LocalDateTime instant = LocalDateTime.ofEpochSecond(startAt.getTime(), 0, UTC);

        assertThat(instant).isEqualToIgnoringNanos(timeDate);
    }