<commit_before>package com.continuuity.metrics.service;

import com.continuuity.common.conf.CConfiguration;
import com.continuuity.common.conf.Constants;
import com.continuuity.common.service.ServerException;
import com.continuuity.metrics.stubs.FlowMonitor;
import com.continuuity.observer.StateChangeCallback;
import com.continuuity.observer.StateChangeListener;
import com.continuuity.observer.StateChangeListenerException;
import com.continuuity.observer.internal.StateChange;
import com.google.inject.Inject;
import com.netflix.curator.framework.CuratorFramework;
import com.netflix.curator.framework.CuratorFrameworkFactory;
import com.netflix.curator.retry.RetryUntilElapsed;
import org.apache.thrift.server.THsHaServer;
import org.apache.thrift.transport.TNonblockingServerSocket;
import org.apache.thrift.transport.TTransportException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 *
 */
public class MetricsSingleNodeServer implements MetricsServer {
  private static final Logger Log = LoggerFactory.getLogger(MetricsSingleNodeServer.class);
  private static final String STATE_CHANGE_QUEUE = "/continuuity/system/queue/statechange";
  private THsHaServer server;
  private CuratorFramework client;
  private ExecutorService executorService;
  private final MetricsHandler handler;
  private final StateChangeCallback callback;
  private StateChangeListener listener;
  private static final String DEFAULT_JDBC_URI="jdbc:hsqldb:mem:fmdb";

  @Inject
  public MetricsSingleNodeServer(MetricsHandler handler, StateChangeCallback callback) {
    this.handler = handler;
    this.callback = callback;
  }

  /**
   * Starts the {@link com.continuuity.common.service.Server}
   *
   * @param args arguments for the service
   * @param conf instance of configuration object.
   */
  @Override
  public void start(String[] args, CConfiguration conf) throws ServerException {
    String uri = conf.get("overlord.jdbc.uri", DEFAULT_JDBC_URI);

    try {
      handler.init(uri);
      callback.init(uri);
    } catch (Exception e) {
      throw new ServerException(e.getMessage());
    }

    String zkEnsemble = conf.get(Constants.CFG_ZOOKEEPER_ENSEMBLE, Constants.DEFAULT_ZOOKEEPER_ENSEMBLE);
    try {
      executorService = Executors.newCachedThreadPool();
      client = CuratorFrameworkFactory.newClient(zkEnsemble, new RetryUntilElapsed(30000, 2000));
      client.start();

      listener = StateChange.Server.newListener(client);
      listener.listen(STATE_CHANGE_QUEUE, callback);

      String portProperty = conf.get(Constants.CFG_FLOW_MONITOR_SERVER_PORT,
        Constants.DEFAULT_FLOW_MONITOR_SERVER_PORT);
      int port = Integer.valueOf(portProperty);

      String threadCntProperty = conf.get(Constants.CFG_FLOW_MONITOR_SERVER_THREADS,
        Constants.DEFAULT_FLOW_MONITOR_SERVER_THREADS);
      int threads = Integer.valueOf(threadCntProperty);

      MetricsImpl serviceImpl = new MetricsImpl(handler);
      THsHaServer.Args serverArgs =
        new THsHaServer
          .Args(new TNonblockingServerSocket(port))
          .executorService(executorService)
          .processor(new FlowMonitor.Processor(serviceImpl))
          .workerThreads(threads);
      server = new THsHaServer(serverArgs);
      new Thread(new Runnable() {
        @Override
        public void run() {
          server.serve();
        }
      }).start();
    } catch (IOException e) {
      Log.error("Failed to create FARService server. Reason : {}", e.getMessage());
      throw new ServerException("Failed to create FARService server", e);
    } catch (TTransportException e) {
      Log.error("Non-blocking server error. Reason : {}", e.getMessage());
      throw new ServerException("Non-blocking server error.", e);
    } catch (StateChangeListenerException e) {<commit_msg>ENG-532 Removed the IOException catch block that was causing compilation errors. This is break fix and more investigation needed to figure out which dependecy caused this issue<commit_after>package com.continuuity.metrics.service;

import com.continuuity.common.conf.CConfiguration;
import com.continuuity.common.conf.Constants;
import com.continuuity.common.service.ServerException;
import com.continuuity.metrics.stubs.FlowMonitor;
import com.continuuity.observer.StateChangeCallback;
import com.continuuity.observer.StateChangeListener;
import com.continuuity.observer.StateChangeListenerException;
import com.continuuity.observer.internal.StateChange;
import com.google.inject.Inject;
import com.netflix.curator.framework.CuratorFramework;
import com.netflix.curator.framework.CuratorFrameworkFactory;
import com.netflix.curator.retry.RetryUntilElapsed;
import org.apache.thrift.server.THsHaServer;
import org.apache.thrift.transport.TNonblockingServerSocket;
import org.apache.thrift.transport.TTransportException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 *
 */
public class MetricsSingleNodeServer implements MetricsServer {
  private static final Logger Log = LoggerFactory.getLogger(MetricsSingleNodeServer.class);
  private static final String STATE_CHANGE_QUEUE = "/continuuity/system/queue/statechange";
  private THsHaServer server;
  private CuratorFramework client;
  private ExecutorService executorService;
  private final MetricsHandler handler;
  private final StateChangeCallback callback;
  private StateChangeListener listener;
  private static final String DEFAULT_JDBC_URI="jdbc:hsqldb:mem:fmdb";

  @Inject
  public MetricsSingleNodeServer(MetricsHandler handler, StateChangeCallback callback) {
    this.handler = handler;
    this.callback = callback;
  }

  /**
   * Starts the {@link com.continuuity.common.service.Server}
   *
   * @param args arguments for the service
   * @param conf instance of configuration object.
   */
  @Override
  public void start(String[] args, CConfiguration conf) throws ServerException {
    String uri = conf.get("overlord.jdbc.uri", DEFAULT_JDBC_URI);

    try {
      handler.init(uri);
      callback.init(uri);
    } catch (Exception e) {
      throw new ServerException(e.getMessage());
    }

    String zkEnsemble = conf.get(Constants.CFG_ZOOKEEPER_ENSEMBLE, Constants.DEFAULT_ZOOKEEPER_ENSEMBLE);
    try {
      executorService = Executors.newCachedThreadPool();
      client = CuratorFrameworkFactory.newClient(zkEnsemble, new RetryUntilElapsed(30000, 2000));
      client.start();

      listener = StateChange.Server.newListener(client);
      listener.listen(STATE_CHANGE_QUEUE, callback);

      String portProperty = conf.get(Constants.CFG_FLOW_MONITOR_SERVER_PORT,
        Constants.DEFAULT_FLOW_MONITOR_SERVER_PORT);
      int port = Integer.valueOf(portProperty);

      String threadCntProperty = conf.get(Constants.CFG_FLOW_MONITOR_SERVER_THREADS,
        Constants.DEFAULT_FLOW_MONITOR_SERVER_THREADS);
      int threads = Integer.valueOf(threadCntProperty);

      MetricsImpl serviceImpl = new MetricsImpl(handler);
      THsHaServer.Args serverArgs =
        new THsHaServer
          .Args(new TNonblockingServerSocket(port))
          .executorService(executorService)
          .processor(new FlowMonitor.Processor(serviceImpl))
          .workerThreads(threads);
      server = new THsHaServer(serverArgs);
      new Thread(new Runnable() {
        @Override
        public void run() {
          server.serve();
        }
      }).start();
    } catch (TTransportException e) {
      Log.error("Non-blocking server error. Reason : {}", e.getMessage());
      throw new ServerException("Non-blocking server error.", e);
    } catch (StateChangeListenerException e) {
{'fraction_non_alphanumeric': 0.09017548354041245, 'fraction_numerical': 0.0032022543870885104, 'mean_word_length': 4.437325905292479, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '5219564', 'n_tokens_mistral': 2238, 'n_tokens_neox': 2187, 'n_words': 456}