public CompletableFuture<Void> discardSamples(final String experimentName, final int maxConcurrency) {
    final AttributeValue experimentNameAttributeValue = AttributeValue.fromS(experimentName);

    return Flux.from(dynamoDbAsyncClient.scanPaginator(ScanRequest.builder()
                .tableName(tableName)
                .filterExpression("#experiment = :experiment")
                .expressionAttributeNames(Map.of("#experiment", KEY_EXPERIMENT_NAME))
                .expressionAttributeValues(Map.of(":experiment", experimentNameAttributeValue))
                .projectionExpression(ATTR_ACI_AND_DEVICE_ID)
                .build())
            .items())
        .map(item -> item.get(ATTR_ACI_AND_DEVICE_ID))
        .flatMap(aciAndDeviceId -> Mono.fromFuture(() -> dynamoDbAsyncClient.deleteItem(DeleteItemRequest.builder()
            .tableName(tableName)
            .key(Map.of(
                KEY_EXPERIMENT_NAME, experimentNameAttributeValue,
                ATTR_ACI_AND_DEVICE_ID, aciAndDeviceId))
            .build()))
            .retryWhen(Retry.backoff(5, Duration.ofSeconds(5)))
            .onErrorResume(throwable -> {
              log.warn("Failed to delete sample for experiment {}", experimentName, throwable);
              return Mono.empty();
            }), maxConcurrency)
        .then()
        .toFuture();
  }
@Test
  void discardSamples() throws JsonProcessingException {
    final String discardSamplesExperimentName = "discard-experiment";
    final String retainSamplesExperimentName = "retain-experiment";
    final int sampleCount = 16;

    for (int i = 0; i < sampleCount; i++) {
      pushNotificationExperimentSamples.recordInitialState(UUID.randomUUID(),
              Device.PRIMARY_ID,
              discardSamplesExperimentName,
              ThreadLocalRandom.current().nextBoolean(),
              new TestDeviceState(ThreadLocalRandom.current().nextInt()))
          .join();

      pushNotificationExperimentSamples.recordInitialState(UUID.randomUUID(),
              Device.PRIMARY_ID,
              retainSamplesExperimentName,
              ThreadLocalRandom.current().nextBoolean(),
              new TestDeviceState(ThreadLocalRandom.current().nextInt()))
          .join();
    }

    pushNotificationExperimentSamples.discardSamples(discardSamplesExperimentName, 1).join();

    {
      final QueryResponse queryResponse = DYNAMO_DB_EXTENSION.getDynamoDbAsyncClient().query(QueryRequest.builder()
              .tableName(DynamoDbExtensionSchema.Tables.PUSH_NOTIFICATION_EXPERIMENT_SAMPLES.tableName())
              .keyConditionExpression("#experiment = :experiment")
              .expressionAttributeNames(Map.of("#experiment", PushNotificationExperimentSamples.KEY_EXPERIMENT_NAME))
              .expressionAttributeValues(Map.of(":experiment", AttributeValue.fromS(discardSamplesExperimentName)))
              .select(Select.COUNT)
              .build())
          .join();

      assertEquals(0, queryResponse.count());
    }

    {
      final QueryResponse queryResponse = DYNAMO_DB_EXTENSION.getDynamoDbAsyncClient().query(QueryRequest.builder()
              .tableName(DynamoDbExtensionSchema.Tables.PUSH_NOTIFICATION_EXPERIMENT_SAMPLES.tableName())
              .keyConditionExpression("#experiment = :experiment")
              .expressionAttributeNames(Map.of("#experiment", PushNotificationExperimentSamples.KEY_EXPERIMENT_NAME))
              .expressionAttributeValues(Map.of(":experiment", AttributeValue.fromS(retainSamplesExperimentName)))
              .select(Select.COUNT)
              .build())
          .join();

      assertEquals(sampleCount, queryResponse.count());
    }
  }