<commit_before>/*
 * 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.accumulo.test.functional;

import static com.google.common.base.Charsets.UTF_8;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

import java.util.EnumSet;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.Set;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.accumulo.core.client.AccumuloException;
import org.apache.accumulo.core.client.AccumuloSecurityException;
import org.apache.accumulo.core.client.BatchWriter;
import org.apache.accumulo.core.client.BatchWriterConfig;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.IteratorSetting;
import org.apache.accumulo.core.client.MutationsRejectedException;
import org.apache.accumulo.core.client.Scanner;
import org.apache.accumulo.core.client.TableNotFoundException;
import org.apache.accumulo.core.client.admin.ActiveScan;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Mutation;
import org.apache.accumulo.core.data.Range;
import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.iterators.IteratorUtil;
import org.apache.accumulo.core.security.Authorizations;
import org.apache.accumulo.core.security.ColumnVisibility;
import org.apache.accumulo.core.util.UtilWaitThread;
import org.apache.accumulo.harness.AccumuloClusterIT;
import org.apache.hadoop.io.Text;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * ACCUMULO-2641 Integration test. ACCUMULO-2641 Adds scan id to thrift protocol so that {@code org.apache.accumulo.core.client.admin.ActiveScan.getScanid()}
 * returns a unique scan id.
 * <p>
 * <p/>
 * The test uses the Minicluster and the {@code org.apache.accumulo.test.functional.SlowIterator} to create multiple scan sessions. The test exercises multiple
 * tablet servers with splits and multiple ranges to force the scans to occur across multiple tablet servers for completeness.
 * <p/>
 * This patch modified thrift, the TraceRepoDeserializationTest test seems to fail unless the following be added:
 * <p/>
 * private static final long serialVersionUID = -4659975753252858243l;
 * <p/>
 * back into org.apache.accumulo.trace.thrift.TInfo until that test signature is regenerated.
 */
public class ScanIdIT extends AccumuloClusterIT {

  private static final Logger log = LoggerFactory.getLogger(ScanIdIT.class);

  private static final int NUM_SCANNERS = 8;

  private static final int NUM_DATA_ROWS = 100;

  private static final Random random = new Random();

  private static final ExecutorService pool = Executors.newFixedThreadPool(NUM_SCANNERS);

  private static volatile boolean testInProgress = true;

  private static final Map<Integer,Value> resultsByWorker = new ConcurrentHashMap<Integer,Value>();

  @Override
  protected int defaultTimeoutSeconds() {
    return 60;
  }

  /**
   * @throws Exception
   *           any exception is a test failure.
   */
  @Test
  public void testScanId() throws Exception {

    final String tableName = getUniqueNames(1)[0];
    Connector conn = getConnector();
    conn.tableOperations().create(tableName);

    addSplits(conn, tableName);

    generateSampleData(conn, tableName);

    attachSlowIterator(conn, tableName);

    for (int scannerIndex = 0; scannerIndex < NUM_SCANNERS; scannerIndex++) {
      ScannerThread st = new ScannerThread(conn, scannerIndex, tableName);
      pool.submit(st);
    }

    // wait for scanners to report a result.
    while (testInProgress) {

      if (resultsByWorker.size() < NUM_SCANNERS) {
        log.trace("Results reported {}", resultsByWorker.size());
        UtilWaitThread.sleep(750);
      } else {
        // each worker has reported at least one result.
        testInProgress = false;

        log.debug("Final result count {}", resultsByWorker.size());

        // delay to allow scanners to react to end of test and cleanly close.
        UtilWaitThread.sleep(1000);
      }

    }

    // all scanner have reported at least 1 result, so check for unique scan ids.
    Set<Long> scanIds = new HashSet<Long>();

    List<String> tservers = conn.instanceOperations().getTabletServers();

    log.debug("tablet servers {}", tservers.toString());

    for (String tserver : tservers) {

      List<ActiveScan> activeScans = conn.instanceOperations().getActiveScans(tserver);

      log.debug("TServer {} has {} active scans", tserver, activeScans.size());
<commit_msg>ACCUMULO-3889 Make ScanIdIT a bit more reliable for TNFE<commit_after>/*
 * 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.accumulo.test.functional;

import static com.google.common.base.Charsets.UTF_8;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

import java.util.EnumSet;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.Set;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.accumulo.core.client.AccumuloException;
import org.apache.accumulo.core.client.AccumuloSecurityException;
import org.apache.accumulo.core.client.BatchWriter;
import org.apache.accumulo.core.client.BatchWriterConfig;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.IteratorSetting;
import org.apache.accumulo.core.client.MutationsRejectedException;
import org.apache.accumulo.core.client.Scanner;
import org.apache.accumulo.core.client.TableNotFoundException;
import org.apache.accumulo.core.client.admin.ActiveScan;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Mutation;
import org.apache.accumulo.core.data.Range;
import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.iterators.IteratorUtil;
import org.apache.accumulo.core.security.Authorizations;
import org.apache.accumulo.core.security.ColumnVisibility;
import org.apache.accumulo.core.util.UtilWaitThread;
import org.apache.accumulo.harness.AccumuloClusterIT;
import org.apache.hadoop.io.Text;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * ACCUMULO-2641 Integration test. ACCUMULO-2641 Adds scan id to thrift protocol so that {@code org.apache.accumulo.core.client.admin.ActiveScan.getScanid()}
 * returns a unique scan id.
 * <p>
 * <p/>
 * The test uses the Minicluster and the {@code org.apache.accumulo.test.functional.SlowIterator} to create multiple scan sessions. The test exercises multiple
 * tablet servers with splits and multiple ranges to force the scans to occur across multiple tablet servers for completeness.
 * <p/>
 * This patch modified thrift, the TraceRepoDeserializationTest test seems to fail unless the following be added:
 * <p/>
 * private static final long serialVersionUID = -4659975753252858243l;
 * <p/>
 * back into org.apache.accumulo.trace.thrift.TInfo until that test signature is regenerated.
 */
public class ScanIdIT extends AccumuloClusterIT {

  private static final Logger log = LoggerFactory.getLogger(ScanIdIT.class);

  private static final int NUM_SCANNERS = 8;

  private static final int NUM_DATA_ROWS = 100;

  private static final Random random = new Random();

  private static final ExecutorService pool = Executors.newFixedThreadPool(NUM_SCANNERS);

  private static volatile boolean testInProgress = true;

  private static final Map<Integer,Value> resultsByWorker = new ConcurrentHashMap<Integer,Value>();

  @Override
  protected int defaultTimeoutSeconds() {
    return 60;
  }

  /**
   * @throws Exception
   *           any exception is a test failure.
   */
  @Test
  public void testScanId() throws Exception {

    final String tableName = getUniqueNames(1)[0];
    Connector conn = getConnector();
    conn.tableOperations().create(tableName);

    addSplits(conn, tableName);

    generateSampleData(conn, tableName);

    attachSlowIterator(conn, tableName);

    for (int scannerIndex = 0; scannerIndex < NUM_SCANNERS; scannerIndex++) {
      ScannerThread st = new ScannerThread(conn, scannerIndex, tableName);
      pool.submit(st);
    }

    // wait for scanners to report a result.
    while (testInProgress) {

      if (resultsByWorker.size() < NUM_SCANNERS) {
        log.trace("Results reported {}", resultsByWorker.size());
        UtilWaitThread.sleep(750);
      } else {
        // each worker has reported at least one result.
        testInProgress = false;

        log.debug("Final result count {}", resultsByWorker.size());

        // delay to allow scanners to react to end of test and cleanly close.
        UtilWaitThread.sleep(1000);
      }

    }

    // all scanner have reported at least 1 result, so check for unique scan ids.
    Set<Long> scanIds = new HashSet<Long>();

    List<String> tservers = conn.instanceOperations().getTabletServers();

    log.debug("tablet servers {}", tservers.toString());

    for (String tserver : tservers) {

      List<ActiveScan> activeScans = null;
      for (int i = 0; i < 10; i++) {
        try {
          activeScans = conn.instanceOperations().getActiveScans(tserver);
          break;
        } catch (AccumuloException e) {
          if (e.getCause() instanceof TableNotFoundException) {
            log.debug("Got TableNotFoundException, will retry");
            Thread.sleep(200);
            continue;
          }
          throw e;
        }
      }

      assertNotNull("Repeatedly got exception trying to active scans", activeScans);

      log.debug("TServer {} has {} active scans", tserver, activeScans.size());

{'fraction_non_alphanumeric': 0.09508025122121423, 'fraction_numerical': 0.009769713886950453, 'mean_word_length': 4.052886734244161, 'pattern_counts': {'":': 0, '<': 30, '<?xml version=': 0, '>': 25, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2277371', 'n_tokens_mistral': 3378, 'n_tokens_neox': 3201, 'n_words': 969}