@Override
  protected void crawlAccounts(final Flux<Account> accounts) {
    final int maxConcurrency = getNamespace().getInt(MAX_CONCURRENCY_ARGUMENT);
    final boolean dryRun = getNamespace().getBoolean(DRY_RUN_ARGUMENT);

    final PushNotificationExperiment<T> experiment =
        experimentFactory.buildExperiment(getCommandDependencies(), getConfiguration());

    final PushNotificationExperimentSamples pushNotificationExperimentSamples =
        getCommandDependencies().pushNotificationExperimentSamples();

    log.info("Starting \"{}\" with max concurrency: {}", experiment.getExperimentName(), maxConcurrency);

    accounts
        .flatMap(account -> Flux.fromIterable(account.getDevices()).map(device -> Tuples.of(account, device)))
        .doOnNext(ignored -> DEVICE_INSPECTED_COUNTER.increment())
        .flatMap(accountAndDevice -> Mono.fromFuture(() ->
                experiment.isDeviceEligible(accountAndDevice.getT1(), accountAndDevice.getT2()))
            .mapNotNull(eligible -> eligible ? accountAndDevice : null), maxConcurrency)
        .flatMap(accountAndDevice -> {
          final UUID accountIdentifier = accountAndDevice.getT1().getIdentifier(IdentityType.ACI);
          final byte deviceId = accountAndDevice.getT2().getId();

          final Mono<Boolean> recordInitialSampleMono = dryRun
              ? Mono.just(true)
              : Mono.fromFuture(() -> {
                    try {
                      return pushNotificationExperimentSamples.recordInitialState(
                          accountIdentifier,
                          deviceId,
                          experiment.getExperimentName(),
                          isInExperimentGroup(accountIdentifier, deviceId, experiment.getExperimentName()),
                          experiment.getState(accountAndDevice.getT1(), accountAndDevice.getT2()));
                    } catch (final JsonProcessingException e) {
                      throw new UncheckedIOException(e);
                    }
                  })
                  .retryWhen(Retry.backoff(3, Duration.ofSeconds(1))
                      .onRetryExhaustedThrow(((backoffSpec, retrySignal) -> retrySignal.failure())));

          return recordInitialSampleMono.mapNotNull(stateStored -> {
                Metrics.counter(RECORD_INITIAL_SAMPLE_COUNTER_NAME,
                        DRY_RUN_TAG_NAME, String.valueOf(dryRun),
                        "initialSampleAlreadyExists", String.valueOf(!stateStored))
                    .increment();

                return stateStored ? accountAndDevice : null;
              })
              .onErrorResume(throwable -> {
                log.warn("Failed to record initial sample for {}:{} in experiment {}",
                    accountIdentifier, deviceId, experiment.getExperimentName(), throwable);

                return Mono.empty();
              });
        }, maxConcurrency)
        .flatMap(accountAndDevice -> {
          final Account account = accountAndDevice.getT1();
          final Device device = accountAndDevice.getT2();
          final boolean inExperimentGroup =
              isInExperimentGroup(account.getIdentifier(IdentityType.ACI), device.getId(), experiment.getExperimentName());

          final Mono<Void> applyTreatmentMono = dryRun
              ? Mono.empty()
              : Mono.fromFuture(() -> inExperimentGroup
                      ? experiment.applyExperimentTreatment(account, device)
                      : experiment.applyControlTreatment(account, device))
                  .onErrorResume(throwable -> {
                    log.warn("Failed to apply {} treatment for {}:{} in experiment {}",
                        inExperimentGroup ? "experimental" : " control",
                        account.getIdentifier(IdentityType.ACI),
                        device.getId(),
                        experiment.getExperimentName(),
                        throwable);

                    return Mono.empty();
                  });

          return applyTreatmentMono
                  .doOnSuccess(ignored -> Metrics.counter(APPLY_TREATMENT_COUNTER_NAME,
                      DRY_RUN_TAG_NAME, String.valueOf(dryRun),
                      "treatment", inExperimentGroup ? "experiment" : "control").increment());
        }, maxConcurrency)
        .then()
        .block();
  }
@Test
  void crawlAccountsExperimentException() {
    final Device device = mock(Device.class);
    when(device.getId()).thenReturn(DEVICE_ID);

    final Account account = mock(Account.class);
    when(account.getIdentifier(IdentityType.ACI)).thenReturn(ACCOUNT_IDENTIFIER);
    when(account.getDevices()).thenReturn(List.of(device));

    when(experiment.applyExperimentTreatment(account, device))
        .thenReturn(CompletableFuture.failedFuture(new RuntimeException()));

    assertDoesNotThrow(() -> startPushNotificationExperimentCommand.crawlAccounts(Flux.just(account)));
    verify(experiment).applyExperimentTreatment(account, device);
  }