public <T> Flux<PushNotificationExperimentSample<T>> getSamples(final String experimentName, final Class<T> stateClass) {

    return Flux.from(dynamoDbAsyncClient.queryPaginator(QueryRequest.builder()
                .tableName(tableName)
                .keyConditionExpression("#experiment = :experiment")
                .expressionAttributeNames(Map.of("#experiment", KEY_EXPERIMENT_NAME))
                .expressionAttributeValues(Map.of(":experiment", AttributeValue.fromS(experimentName)))
                .build())
            .items())
        .handle((item, sink) -> {
          try {
            final Tuple2<UUID, Byte> aciAndDeviceId = parseSortKey(item.get(ATTR_ACI_AND_DEVICE_ID));

            final boolean inExperimentGroup = item.get(ATTR_IN_EXPERIMENT_GROUP).bool();
            final T initialState = parseState(item.get(ATTR_INITIAL_STATE).s(), stateClass);
            final T finalState = item.get(ATTR_FINAL_STATE) != null
                ? parseState(item.get(ATTR_FINAL_STATE).s(), stateClass)
                : null;

            sink.next(new PushNotificationExperimentSample<>(aciAndDeviceId.getT1(), aciAndDeviceId.getT2(), inExperimentGroup, initialState, finalState));
          } catch (final JsonProcessingException e) {
            sink.error(e);
          }
        });
  }
@Test
  void getSamples() throws JsonProcessingException {
    final String experimentName = "test-experiment";
    final UUID initialSampleAccountIdentifier = UUID.randomUUID();
    final byte initialSampleDeviceId = (byte) ThreadLocalRandom.current().nextInt(Device.MAXIMUM_DEVICE_ID);
    final boolean initialSampleInExperimentGroup = ThreadLocalRandom.current().nextBoolean();

    final UUID finalSampleAccountIdentifier = UUID.randomUUID();
    final byte finalSampleDeviceId = (byte) ThreadLocalRandom.current().nextInt(Device.MAXIMUM_DEVICE_ID);
    final boolean finalSampleInExperimentGroup = ThreadLocalRandom.current().nextBoolean();

    final int initialBounciness = ThreadLocalRandom.current().nextInt();
    final int finalBounciness = initialBounciness + 17;

    pushNotificationExperimentSamples.recordInitialState(initialSampleAccountIdentifier,
            initialSampleDeviceId,
            experimentName,
            initialSampleInExperimentGroup,
            new TestDeviceState(initialBounciness))
        .join();

    pushNotificationExperimentSamples.recordInitialState(finalSampleAccountIdentifier,
            finalSampleDeviceId,
            experimentName,
            finalSampleInExperimentGroup,
            new TestDeviceState(initialBounciness))
        .join();

    pushNotificationExperimentSamples.recordFinalState(finalSampleAccountIdentifier,
            finalSampleDeviceId,
            experimentName,
            new TestDeviceState(finalBounciness))
        .join();

    pushNotificationExperimentSamples.recordInitialState(UUID.randomUUID(),
            (byte) ThreadLocalRandom.current().nextInt(Device.MAXIMUM_DEVICE_ID),
            experimentName + "-different",
            ThreadLocalRandom.current().nextBoolean(),
            new TestDeviceState(ThreadLocalRandom.current().nextInt()))
        .join();

    final Set<PushNotificationExperimentSample<TestDeviceState>> expectedSamples = Set.of(
        new PushNotificationExperimentSample<>(initialSampleAccountIdentifier,
            initialSampleDeviceId,
            initialSampleInExperimentGroup,
            new TestDeviceState(initialBounciness),
            null),

        new PushNotificationExperimentSample<>(finalSampleAccountIdentifier,
            finalSampleDeviceId,
            finalSampleInExperimentGroup,
            new TestDeviceState(initialBounciness),
            new TestDeviceState(finalBounciness)));

    assertEquals(expectedSamples,
        pushNotificationExperimentSamples.getSamples(experimentName, TestDeviceState.class).collect(Collectors.toSet()).block());
  }