content: /*
   * Licensed to the Apache Software Foundation (ASF) under one
   * or more contributor license agreements.  See the NOTICE file
   * distributed with this work for additional information
   * regarding copyright ownership.  The ASF licenses this file
   * to you under the Apache License, Version 2.0 (the
   * "License"); you may not use this file except in compliance
   * with the License.  You may obtain a copy of the License at
   *
   *     http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  
  package org.apache.cassandra.repair;
  
  import java.util.ArrayList;
  import java.util.List;
  import java.util.concurrent.TimeUnit;
  
  import org.junit.Test;
  
  import accord.utils.Gen;
  import accord.utils.Gens;
  import org.apache.cassandra.config.DatabaseDescriptor;
  import org.apache.cassandra.config.RetrySpec;
  import org.apache.cassandra.db.ColumnFamilyStore;
  import org.apache.cassandra.repair.consistent.LocalSessions;
  import org.apache.cassandra.repair.state.Completable;
  import org.apache.cassandra.utils.Closeable;
  import org.apache.cassandra.utils.FailingBiConsumer;
  import org.assertj.core.api.Assertions;
  
  import static accord.utils.Property.qt;
  
  public class ConcurrentIrWithPreviewFuzzTest extends FuzzTestBase
  {
      @Test
      public void concurrentIrWithPreview()
      {
          // to avoid unlucky timing issues, retry until success; given enough retries we should eventually become success
          DatabaseDescriptor.getRepairRetrySpec().maxAttempts = new RetrySpec.MaxAttempt(Integer.MAX_VALUE);
          qt().withPure(false).withExamples(1).check(rs -> {
              Cluster cluster = new Cluster(rs);
              enableMessageFaults(cluster);
  
              Gen<Cluster.Node> coordinatorGen = Gens.pick(cluster.nodes.keySet()).map(cluster.nodes::get);
  
              List<Closeable> closeables = new ArrayList<>();
              for (int example = 0; example < 100; example++)
              {
                  Cluster.Node irCoordinator = coordinatorGen.next(rs);
                  Cluster.Node previewCoordinator = coordinatorGen.next(rs);
                  RepairCoordinator ir = irCoordinator.repair(KEYSPACE, irOption(rs, irCoordinator, KEYSPACE, ignore -> TABLES));
                  ir.run();
                  RepairCoordinator preview = previewCoordinator.repair(KEYSPACE, previewOption(rs, previewCoordinator, KEYSPACE, ignore -> TABLES), false);
                  preview.run();
  
                  closeables.add(cluster.nodes.get(pickParticipant(rs, previewCoordinator, preview)).doValidation(ignore -> (cfs, validator) -> addMismatch(rs, cfs, validator)));
                  // cause a delay in validation to have more failing previews
                  closeables.add(cluster.nodes.get(pickParticipant(rs, previewCoordinator, preview)).doValidation(next -> (cfs, validator) -> {
                      if (validator.desc.parentSessionId.equals(preview.state.id))
                          delayValidation(cluster, ir, next, cfs, validator);
                      else next.acceptOrFail(cfs, validator);
                  }));
                  // make sure listeners don't leak
                  closeables.add(LocalSessions::unsafeClearListeners);
  
                  cluster.processAll();
  
                  // IR will always pass, but preview is what may fail (if the coordinator is the same)
                  Assertions.assertThat(ir.state.getResult()).describedAs("Unexpected state: %s -> %s; example %d", ir.state, ir.state.getResult(), example).isEqualTo(Completable.Result.success(repairSuccessMessage(ir)));
  
                  Assertions.assertThat(preview.state.getResult()).describedAs("Unexpected state: %s; example %d", preview.state, example).isNotNull();
  
                  if (irCoordinator == previewCoordinator)
                  {
                      Assertions.assertThat(preview.state.getResult().message).describedAs("Unexpected state: %s -> %s; example %d", preview.state, preview.state.getResult(), example).contains("failed with error An incremental repair with session id");
                  }
                  else
                  {
                      assertSuccess(cluster, example, true, preview);
                  }
                  closeables.forEach(Closeable::close);
                  closeables.clear();
              }
          });
      }
  
      private void delayValidation(Cluster cluster, RepairCoordinator ir, FailingBiConsumer<ColumnFamilyStore, Validator> next, ColumnFamilyStore cfs, Validator validator)
      {
          cluster.unorderedScheduled.schedule(() -> {
              // make sure to wait for IR to complete...
              Completable.Result result = ir.state.getResult();
              if (result == null)
              {
                  delayValidation(cluster, ir, next, cfs, validator);
                  return;
              }
              next.accept(cfs, validator);
          }, 1, TimeUnit.HOURS);
      }
  }
  