<commit_before>package com.continuuity.data2.transaction.queue.hbase.coprocessor;

import com.continuuity.common.conf.CConfiguration;
import com.continuuity.data2.transaction.queue.QueueConstants;
import com.continuuity.data2.transaction.queue.QueueEntryRow;
import com.continuuity.data2.util.hbase.ConfigurationTable;
import com.google.common.collect.Maps;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;

import javax.annotation.Nullable;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.NavigableMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ConcurrentSkipListMap;

/**
 * Provides a RegionServer shared cache for all instances of {@link HBaseQueueRegionObserver} of the recent
 * queue consumer configuration.
 */
public class ConsumerConfigCache {
  private static final Log LOG = LogFactory.getLog(ConsumerConfigCache.class);

  private static final int LONG_BYTES = Long.SIZE / Byte.SIZE;
  // update interval for CConfiguration
  private static final long CONFIG_UPDATE_FREQUENCY = 300 * 1000L;

  private static ConcurrentMap<byte[], ConsumerConfigCache> instances =
    new ConcurrentSkipListMap<byte[], ConsumerConfigCache>(Bytes.BYTES_COMPARATOR);

  private final byte[] configTableName;
  private final Configuration hConf;

  private Thread refreshThread;
  private long lastUpdated;
  private volatile Map<byte[], QueueConsumerConfig> configCache = Maps.newTreeMap(Bytes.BYTES_COMPARATOR);
  private long configCacheUpdateFrequency = QueueConstants.DEFAULT_QUEUE_CONFIG_UPDATE_FREQUENCY;
  private ConfigurationTable configTable;
  private String tableNamespace;
  private CConfiguration conf;
  // timestamp of the last update from the configuration table
  private long lastConfigUpdate;

  ConsumerConfigCache(Configuration hConf, byte[] configTableName) {
    this.hConf = hConf;
    this.configTableName = configTableName;
    String queueConfigTableName = Bytes.toString(configTableName);
    String[] parts = queueConfigTableName.split("\\.", 2);
    this.tableNamespace = parts[0];
    this.configTable = new ConfigurationTable(hConf);
  }

  private void init() {
    startRefreshThread();
  }

  @Nullable
  public QueueConsumerConfig getConsumerConfig(byte[] queueName) {
    return configCache.get(queueName);
  }

  private void updateConfig() {
    long now = System.currentTimeMillis();<commit_msg>fix for queue unit-test<commit_after>package com.continuuity.data2.transaction.queue.hbase.coprocessor;

import com.continuuity.common.conf.CConfiguration;
import com.continuuity.data2.transaction.queue.QueueConstants;
import com.continuuity.data2.transaction.queue.QueueEntryRow;
import com.continuuity.data2.util.hbase.ConfigurationTable;
import com.google.common.collect.Maps;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;

import javax.annotation.Nullable;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.NavigableMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ConcurrentSkipListMap;

/**
 * Provides a RegionServer shared cache for all instances of {@link HBaseQueueRegionObserver} of the recent
 * queue consumer configuration.
 */
public class ConsumerConfigCache {
  private static final Log LOG = LogFactory.getLog(ConsumerConfigCache.class);

  private static final int LONG_BYTES = Long.SIZE / Byte.SIZE;
  // update interval for CConfiguration
  private static final long CONFIG_UPDATE_FREQUENCY = 300 * 1000L;

  private static ConcurrentMap<byte[], ConsumerConfigCache> instances =
    new ConcurrentSkipListMap<byte[], ConsumerConfigCache>(Bytes.BYTES_COMPARATOR);

  private final byte[] configTableName;
  private final Configuration hConf;

  private Thread refreshThread;
  private long lastUpdated;
  private volatile Map<byte[], QueueConsumerConfig> configCache = Maps.newTreeMap(Bytes.BYTES_COMPARATOR);
  private long configCacheUpdateFrequency = QueueConstants.DEFAULT_QUEUE_CONFIG_UPDATE_FREQUENCY;
  private ConfigurationTable configTable;
  private String tableNamespace;
  private CConfiguration conf;
  // timestamp of the last update from the configuration table
  private long lastConfigUpdate;

  ConsumerConfigCache(Configuration hConf, byte[] configTableName) {
    this.hConf = hConf;
    this.configTableName = configTableName;
    String queueConfigTableName = Bytes.toString(configTableName);
    String[] parts = queueConfigTableName.split("\\.", 2);
    this.tableNamespace = parts[0];
    this.configTable = new ConfigurationTable(hConf);
  }

  private void init() {
    startRefreshThread();
  }

  @Nullable
  public QueueConsumerConfig getConsumerConfig(byte[] queueName) {
    return configCache.get(queueName);
  }

  // NOTE: this one is synchronized to support direct invocation from unit-test
  private synchronized void updateConfig() {
    long now = System.currentTimeMillis();
{'fraction_non_alphanumeric': 0.08831497200650172, 'fraction_numerical': 0.0046956835831677805, 'mean_word_length': 5.492379835873388, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '7595659', 'n_tokens_mistral': 1579, 'n_tokens_neox': 1541, 'n_words': 366}