<commit_before>//*********************************************************
// Copyright (c) Microsoft. All rights reserved.
// Licensed under the MIT license. See LICENSE file in the project root for full license information.
//*********************************************************

package com.microsoft.kafkaavailability;

import com.microsoft.kafkaavailability.discovery.CommonUtils;
import com.microsoft.kafkaavailability.discovery.Constants;
import com.microsoft.kafkaavailability.discovery.CuratorClient;
import com.microsoft.kafkaavailability.discovery.CuratorManager;
import com.microsoft.kafkaavailability.properties.AppProperties;
import com.microsoft.kafkaavailability.properties.MetaDataManagerProperties;
import com.microsoft.kafkaavailability.threads.AvailabilityThread;
import com.microsoft.kafkaavailability.threads.ConsumerThread;
import com.microsoft.kafkaavailability.threads.LeaderInfoThread;
import com.microsoft.kafkaavailability.threads.ProducerThread;
import org.apache.commons.cli.*;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.imps.CuratorFrameworkState;
import org.apache.log4j.MDC;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.Phaser;

/***
 * Sends a canary message to every topic and partition in Kafka.
 * Reads data from the tail of every topic and partition in Kafka
 * Reports the availability and latency metrics for the above operations.
 * Availability is defined as the percentage of total partitions that respond to each operation.
 */
public class App {
    final static Logger m_logger = LoggerFactory.getLogger(App.class);

    static int m_sleepTime = 30000;
    static String m_cluster = "localhost";

    static AppProperties appProperties;
    static String m_cluster;
    static MetaDataManagerProperties metaDataProperties;
    static List<String> listServers;

    private static String registrationPath = Constants.DEFAULT_REGISTRATION_ROOT;
    private static String ip = CommonUtils.getIpAddress();
    private static String computerName = CommonUtils.getComputerName();
    private static String serviceSpec = "";

    public static void main(String[] args) throws IOException, MetaDataManagerException, InterruptedException {
        m_logger.info("Starting KafkaAvailability Tool");
        IPropertiesManager appPropertiesManager = new PropertiesManager<>("appProperties.json", AppProperties.class);
        IPropertiesManager metaDataPropertiesManager = new PropertiesManager<>("metadatamanagerProperties.json", MetaDataManagerProperties.class);
        appProperties = (AppProperties) appPropertiesManager.getProperties();
        metaDataProperties = (MetaDataManagerProperties) metaDataPropertiesManager.getProperties();
        Options options = new Options();
        options.addOption("r", "run", true, "Number of runs. Don't use this argument if you want to run infinitely.");
        options.addOption("s", "sleep", true, "Time (in milliseconds) to sleep between each run. Default is 30000");
        options.addOption("c", "cluster", true, "Cluster name. will pull from here if appProperties is null");

        CommandLineParser parser = new DefaultParser();
        HelpFormatter formatter = new HelpFormatter();
        final CuratorFramework curatorFramework = CuratorClient.getCuratorFramework(metaDataProperties.zooKeeperHosts);

        try {
            // parse the command line arguments
            CommandLine line = parser.parse(options, args);
            int howManyRuns;

            if((appProperties.environmentName == null || appProperties.environmentName.equals("")) && line.hasOption("cluster")) {
                appProperties.environmentName = line.getOptionValue("cluster");
            } else {
                throw new IllegalArgumentException("cluster name must be provided either on the command line or in the app properties");
            }

            MDC.put("cluster", appProperties.environmentName);

            MDC.put("computerName", computerName);
            CuratorManager curatorManager = CallRegister(curatorFramework);

            if (line.hasOption("sleep")) {
                m_sleepTime = Integer.parseInt(line.getOptionValue("sleep"));
            }

            if (line.hasOption("run")) {
                howManyRuns = Integer.parseInt(line.getOptionValue("run"));
                for (int i = 0; i < howManyRuns; i++) {
                    waitForChanges(curatorManager);
                    RunOnce(curatorFramework);
                    Thread.sleep(m_sleepTime);
                }
            } else {
                while (true) {
                    waitForChanges(curatorManager);
                    RunOnce(curatorFramework);
                    Thread.sleep(m_sleepTime);
                }
            }
        } catch (ParseException exp) {
            // oops, something went wrong<commit_msg>fixed bug so environment can be provided in appProperties.json<commit_after>//*********************************************************
// Copyright (c) Microsoft. All rights reserved.
// Licensed under the MIT license. See LICENSE file in the project root for full license information.
//*********************************************************

package com.microsoft.kafkaavailability;

import com.microsoft.kafkaavailability.discovery.CommonUtils;
import com.microsoft.kafkaavailability.discovery.Constants;
import com.microsoft.kafkaavailability.discovery.CuratorClient;
import com.microsoft.kafkaavailability.discovery.CuratorManager;
import com.microsoft.kafkaavailability.properties.AppProperties;
import com.microsoft.kafkaavailability.properties.MetaDataManagerProperties;
import com.microsoft.kafkaavailability.threads.AvailabilityThread;
import com.microsoft.kafkaavailability.threads.ConsumerThread;
import com.microsoft.kafkaavailability.threads.LeaderInfoThread;
import com.microsoft.kafkaavailability.threads.ProducerThread;
import org.apache.commons.cli.*;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.imps.CuratorFrameworkState;
import org.apache.log4j.MDC;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.Phaser;

/***
 * Sends a canary message to every topic and partition in Kafka.
 * Reads data from the tail of every topic and partition in Kafka
 * Reports the availability and latency metrics for the above operations.
 * Availability is defined as the percentage of total partitions that respond to each operation.
 */
public class App {
    final static Logger m_logger = LoggerFactory.getLogger(App.class);

    static int m_sleepTime = 30000;

    static AppProperties appProperties;
    static MetaDataManagerProperties metaDataProperties;
    static List<String> listServers;

    private static String registrationPath = Constants.DEFAULT_REGISTRATION_ROOT;
    private static String ip = CommonUtils.getIpAddress();
    private static String computerName = CommonUtils.getComputerName();
    private static String serviceSpec = "";

    public static void main(String[] args) throws IOException, MetaDataManagerException, InterruptedException {
        m_logger.info("Starting KafkaAvailability Tool");
        IPropertiesManager appPropertiesManager = new PropertiesManager<>("appProperties.json", AppProperties.class);
        IPropertiesManager metaDataPropertiesManager = new PropertiesManager<>("metadatamanagerProperties.json", MetaDataManagerProperties.class);
        appProperties = (AppProperties) appPropertiesManager.getProperties();
        metaDataProperties = (MetaDataManagerProperties) metaDataPropertiesManager.getProperties();
        Options options = new Options();
        options.addOption("r", "run", true, "Number of runs. Don't use this argument if you want to run infinitely.");
        options.addOption("s", "sleep", true, "Time (in milliseconds) to sleep between each run. Default is 30000");
        options.addOption("c", "cluster", true, "Cluster name. will pull from here if appProperties is null");

        CommandLineParser parser = new DefaultParser();
        HelpFormatter formatter = new HelpFormatter();
        final CuratorFramework curatorFramework = CuratorClient.getCuratorFramework(metaDataProperties.zooKeeperHosts);

        try {
            // parse the command line arguments
            CommandLine line = parser.parse(options, args);
            int howManyRuns;

            if((appProperties.environmentName == null || appProperties.environmentName.equals(""))) {
                if(line.hasOption("cluster")) {
                    appProperties.environmentName = line.getOptionValue("cluster");   
                } else {
                    throw new IllegalArgumentException("cluster name must be provided either on the command line or in the app properties");
                }
            } 
            
            MDC.put("cluster", appProperties.environmentName);

            MDC.put("computerName", computerName);
            CuratorManager curatorManager = CallRegister(curatorFramework);

            if (line.hasOption("sleep")) {
                m_sleepTime = Integer.parseInt(line.getOptionValue("sleep"));
            }

            if (line.hasOption("run")) {
                howManyRuns = Integer.parseInt(line.getOptionValue("run"));
                for (int i = 0; i < howManyRuns; i++) {
                    waitForChanges(curatorManager);
                    RunOnce(curatorFramework);
                    Thread.sleep(m_sleepTime);
                }
            } else {
                while (true) {
                    waitForChanges(curatorManager);
                    RunOnce(curatorFramework);
                    Thread.sleep(m_sleepTime);
                }
            }
        } catch (ParseException exp) {
            // oops, something went wrong
{'fraction_non_alphanumeric': 0.10652304009575105, 'fraction_numerical': 0.0027927388789148215, 'mean_word_length': 3.3920280332895314, 'pattern_counts': {'":': 0, '<': 11, '<?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': '3397809', 'n_tokens_mistral': 2481, 'n_tokens_neox': 2333, 'n_words': 640}