<commit_before>package com.hubspot.singularity.data;

import com.google.inject.Inject;
import java.util.HashMap;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.curator.CuratorZookeeperClient;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.WatcherRemoveCuratorFramework;
import org.apache.curator.framework.api.CreateBuilder;
import org.apache.curator.framework.api.CuratorListener;
import org.apache.curator.framework.api.DeleteBuilder;
import org.apache.curator.framework.api.ExistsBuilder;
import org.apache.curator.framework.api.GetACLBuilder;
import org.apache.curator.framework.api.GetChildrenBuilder;
import org.apache.curator.framework.api.GetConfigBuilder;
import org.apache.curator.framework.api.GetDataBuilder;
import org.apache.curator.framework.api.ReconfigBuilder;
import org.apache.curator.framework.api.RemoveWatchesBuilder;
import org.apache.curator.framework.api.SetACLBuilder;
import org.apache.curator.framework.api.SetDataBuilder;
import org.apache.curator.framework.api.SyncBuilder;
import org.apache.curator.framework.api.UnhandledErrorListener;
import org.apache.curator.framework.api.transaction.CuratorMultiTransaction;
import org.apache.curator.framework.api.transaction.CuratorTransaction;
import org.apache.curator.framework.api.transaction.TransactionOp;
import org.apache.curator.framework.imps.CuratorFrameworkState;
import org.apache.curator.framework.listen.Listenable;
import org.apache.curator.framework.schema.SchemaSet;
import org.apache.curator.framework.state.ConnectionStateErrorPolicy;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.curator.utils.EnsurePath;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.server.quorum.flexible.QuorumVerifier;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class LoggingCuratorFramework implements CuratorFramework {
  private final CuratorFramework curator;

  private final Map<String, AtomicLong> counters;

  private final Long interval = 60000L * 3; // 3 minutes

  private static final Logger LOG = LoggerFactory.getLogger(
    LoggingCuratorFramework.class
  );

  private class ClearCounter extends TimerTask {

    @Override
    public void run() {
      logAndClear();
    }
  }

  @Inject
  public LoggingCuratorFramework(CuratorFramework curator) {
    this.curator = curator;
    counters = new HashMap<>();
    Timer timer = new Timer();

    timer.schedule(new ClearCounter(), interval, interval);
  }

  private String getCaller() {
    StackTraceElement[] stackTraceElements = Thread.currentThread().getStackTrace();
    long threadId = Thread.currentThread().getId();

    int levelInStack = 1;
    if (
      stackTraceElements[1].getClassName()
        .equals("com.hubspot.singularity.data.CuratorManager") ||
      stackTraceElements[1].getClassName()
        .equals("com.hubspot.singularity.data.CuratorAsyncManager")
    ) {
      levelInStack = 2;
    }

    String longClassName = stackTraceElements[levelInStack].getClassName();
    String className = longClassName.substring(longClassName.lastIndexOf("."));

    return (
      threadId +
      ": " +
      className +
      "\'s " +
      stackTraceElements[levelInStack].getMethodName()
    );
  }

  private void setCounter() {
    String caller = getCaller();
    AtomicLong counter = counters.containsKey(caller)
      ? counters.get(caller)
      : new AtomicLong(0);
    counter.getAndIncrement();
    counters.put(caller, counter);
  }

  public void logAndClear() {
    for (Map.Entry<String, AtomicLong> entry : counters.entrySet()) {
      AtomicLong callCount = entry.getValue();
      LOG.info(
        "{} called ZK {} times in {} milliseconds",
        entry.getKey(),
        callCount.get(),
        interval
      );

      counters.remove(entry.getKey());
    }
  }

  @Override
  public void start() {
    curator.start();
  }

  @Override
  public void close() {
    curator.close();
  }<commit_msg>concurrent hash map for thread safety<commit_after>package com.hubspot.singularity.data;

import com.google.inject.Inject;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
import org.apache.curator.CuratorZookeeperClient;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.WatcherRemoveCuratorFramework;
import org.apache.curator.framework.api.CreateBuilder;
import org.apache.curator.framework.api.CuratorListener;
import org.apache.curator.framework.api.DeleteBuilder;
import org.apache.curator.framework.api.ExistsBuilder;
import org.apache.curator.framework.api.GetACLBuilder;
import org.apache.curator.framework.api.GetChildrenBuilder;
import org.apache.curator.framework.api.GetConfigBuilder;
import org.apache.curator.framework.api.GetDataBuilder;
import org.apache.curator.framework.api.ReconfigBuilder;
import org.apache.curator.framework.api.RemoveWatchesBuilder;
import org.apache.curator.framework.api.SetACLBuilder;
import org.apache.curator.framework.api.SetDataBuilder;
import org.apache.curator.framework.api.SyncBuilder;
import org.apache.curator.framework.api.UnhandledErrorListener;
import org.apache.curator.framework.api.transaction.CuratorMultiTransaction;
import org.apache.curator.framework.api.transaction.CuratorTransaction;
import org.apache.curator.framework.api.transaction.TransactionOp;
import org.apache.curator.framework.imps.CuratorFrameworkState;
import org.apache.curator.framework.listen.Listenable;
import org.apache.curator.framework.schema.SchemaSet;
import org.apache.curator.framework.state.ConnectionStateErrorPolicy;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.curator.utils.EnsurePath;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.server.quorum.flexible.QuorumVerifier;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class LoggingCuratorFramework implements CuratorFramework {
  private final CuratorFramework curator;

  private final ConcurrentMap<String, Integer> counters;

  private final Long interval = 60000L * 3; // 3 minutes

  private static final Logger LOG = LoggerFactory.getLogger(
    LoggingCuratorFramework.class
  );

  private class ClearCounter extends TimerTask {

    @Override
    public void run() {
      logAndClear();
    }
  }

  @Inject
  public LoggingCuratorFramework(CuratorFramework curator) {
    this.curator = curator;
    counters = new ConcurrentHashMap<>();
    Timer timer = new Timer();

    timer.schedule(new ClearCounter(), interval, interval);
  }

  private String getCaller() {
    StackTraceElement[] stackTraceElements = Thread.currentThread().getStackTrace();
    long threadId = Thread.currentThread().getId();

    int levelInStack = 0;
    String className = "";
    for (int i = 0; i < stackTraceElements.length; i++) {
      String longClassName = stackTraceElements[levelInStack].getClassName();
      className = longClassName.substring(longClassName.lastIndexOf("."));

      if (
        !stackTraceElements[i].getClassName().equals("LoggingCuratorFramework") &&
        (
          !stackTraceElements[i].getClassName().equals("CuratorManager") ||
          !stackTraceElements[i].getClassName().equals("CuratorAsyncManager")
        )
      ) {
        levelInStack = i;
        break;
      }
    }

    return (
      threadId +
      ": " +
      className +
      "\'s " +
      stackTraceElements[levelInStack].getMethodName()
    );
  }

  private void setCounter() {
    String caller = getCaller();
    int counter = counters.getOrDefault(caller, 0);
    counters.put(caller, ++counter);
  }

  public void logAndClear() {
    counters.forEach(
      (caller, count) -> {
        LOG.info("{} called ZK {} times in {} milliseconds", caller, count, interval);

        counters.remove(caller);
      }
    );
  }

  @Override
  public void start() {
    curator.start();
  }

  @Override
  public void close() {
    curator.close();
  }
{'fraction_non_alphanumeric': 0.10882635882635883, 'fraction_numerical': 0.0031265031265031266, 'mean_word_length': 4.351994851994852, 'pattern_counts': {'":': 2, '<': 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': '3320881', 'n_tokens_mistral': 2468, 'n_tokens_neox': 2403, 'n_words': 423}