<commit_before>/*
 * Copyright © 2014-2015 <NAME>, Inc.
 *
 * Licensed 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 co.cask.cdap.data.runtime.main;

import co.cask.cdap.api.metrics.MetricsCollectionService;
import co.cask.cdap.app.guice.AppFabricServiceRuntimeModule;
import co.cask.cdap.app.guice.ProgramRunnerRuntimeModule;
import co.cask.cdap.app.guice.ServiceStoreModules;
import co.cask.cdap.app.store.ServiceStore;
import co.cask.cdap.common.conf.CConfiguration;
import co.cask.cdap.common.conf.Constants;
import co.cask.cdap.common.guice.ConfigModule;
import co.cask.cdap.common.guice.DiscoveryRuntimeModule;
import co.cask.cdap.common.guice.IOModule;
import co.cask.cdap.common.guice.KafkaClientModule;
import co.cask.cdap.common.guice.LocationRuntimeModule;
import co.cask.cdap.common.guice.TwillModule;
import co.cask.cdap.common.guice.ZKClientModule;
import co.cask.cdap.common.kerberos.SecurityUtil;
import co.cask.cdap.common.runtime.DaemonMain;
import co.cask.cdap.common.service.RetryOnStartFailureService;
import co.cask.cdap.common.service.RetryStrategies;
import co.cask.cdap.common.utils.DirUtils;
import co.cask.cdap.data.runtime.DataFabricModules;
import co.cask.cdap.data.runtime.DataSetServiceModules;
import co.cask.cdap.data.runtime.DataSetsModules;
import co.cask.cdap.data.stream.StreamAdminModules;
import co.cask.cdap.data2.datafabric.dataset.service.DatasetService;
import co.cask.cdap.data2.util.hbase.ConfigurationTable;
import co.cask.cdap.data2.util.hbase.HBaseTableUtil;
import co.cask.cdap.explore.client.ExploreClient;
import co.cask.cdap.explore.guice.ExploreClientModule;
import co.cask.cdap.explore.service.ExploreServiceUtils;
import co.cask.cdap.gateway.auth.AuthModule;
import co.cask.cdap.internal.app.services.AppFabricServer;
import co.cask.cdap.logging.appender.LogAppenderInitializer;
import co.cask.cdap.logging.guice.LoggingModules;
import co.cask.cdap.metrics.guice.MetricsClientRuntimeModule;
import co.cask.cdap.notifications.feeds.guice.NotificationFeedServiceRuntimeModule;
import co.cask.cdap.notifications.guice.NotificationServiceRuntimeModule;
import com.google.common.base.Charsets;
import com.google.common.base.Supplier;
import com.google.common.base.Throwables;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.google.common.io.Closeables;
import com.google.common.util.concurrent.Service;
import com.google.inject.Guice;
import com.google.inject.Injector;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.security.User;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.twill.api.ElectionHandler;
import org.apache.twill.api.TwillApplication;
import org.apache.twill.api.TwillController;
import org.apache.twill.api.TwillPreparer;
import org.apache.twill.api.TwillRunnerService;
import org.apache.twill.api.logging.PrinterLogHandler;
import org.apache.twill.common.Cancellable;
import org.apache.twill.common.ServiceListenerAdapter;
import org.apache.twill.common.Threads;
import org.apache.twill.internal.zookeeper.LeaderElection;
import org.apache.twill.kafka.client.KafkaClientService;
import org.apache.twill.zookeeper.ZKClientService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.PrintWriter;
import java.io.Writer;
import java.net.InetAddress;
import java.net.URL;
import java.net.UnknownHostException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import javax.annotation.Nullable;

/**
 * Driver class for starting all master services.
 * AppFabricHttpService
 * TwillRunnables: MetricsProcessor, MetricsHttp, LogSaver, TransactionService, StreamHandler.
 */
public class MasterServiceMain extends DaemonMain {
  private static final Logger LOG = LoggerFactory.getLogger(MasterServiceMain.class);

  private static final long MAX_BACKOFF_TIME_MS = TimeUnit.MILLISECONDS.convert(10, TimeUnit.MINUTES);
  private static final long SUCCESSFUL_RUN_DURATON_MS = TimeUnit.MILLISECONDS.convert(20, TimeUnit.MINUTES);

  // Maximum time to try looking up the existing twill application
  private static final long LOOKUP_ATTEMPT_TIMEOUT_MS = 2000;

  private final CConfiguration cConf;
  private final Configuration hConf;
  private final Injector baseInjector;
  private final ZKClientService zkClient;
  private final TwillRunnerService twillRunner;
  private final KafkaClientService kafkaClient;
  private final MetricsCollectionService metricsCollectionService;
  private final ServiceStore serviceStore;
  private final LeaderElection leaderElection;
  private final TokenSecureStoreUpdater secureStoreUpdater;

  private volatile boolean stopped;

  public static void main(final String[] args) throws Exception {
    LOG.info("Starting {}", MasterServiceMain.class.getSimpleName());
    new MasterServiceMain().doMain(args);
  }

  public MasterServiceMain() {
    this.cConf = CConfiguration.create();
    this.cConf.set(Constants.Dataset.Manager.ADDRESS, getLocalHost().getCanonicalHostName());
    this.hConf = HBaseConfiguration.create();

    Injector injector = createBaseInjector(cConf, hConf);
    this.baseInjector = injector;
    this.zkClient = injector.getInstance(ZKClientService.class);
    this.twillRunner = injector.getInstance(TwillRunnerService.class);
    this.kafkaClient = injector.getInstance(KafkaClientService.class);
    this.metricsCollectionService = injector.getInstance(MetricsCollectionService.class);
    this.serviceStore = injector.getInstance(ServiceStore.class);
    this.secureStoreUpdater = baseInjector.getInstance(TokenSecureStoreUpdater.class);
    this.leaderElection = createLeaderElection();
  }

  @Override
  public void init(String[] args) {
    cleanupTempDir();

    checkExploreRequirements();
    login();
  }

  @Override
  public void start() {
    createSystemHBaseNamespace();
    updateConfigurationTable();

    LogAppenderInitializer logAppenderInitializer = baseInjector.getInstance(LogAppenderInitializer.class);
    logAppenderInitializer.initialize();

    zkClient.startAndWait();
    twillRunner.startAndWait();
    kafkaClient.startAndWait();
    metricsCollectionService.startAndWait();
    serviceStore.startAndWait();
    leaderElection.startAndWait();
  }

  @Override
  public void stop() {
    LOG.info("Stopping {}", Constants.Service.MASTER_SERVICES);
    stopped = true;

    stopQuietly(leaderElection);
    stopQuietly(serviceStore);
    stopQuietly(metricsCollectionService);
    stopQuietly(kafkaClient);
    stopQuietly(twillRunner);
    stopQuietly(zkClient);

    if (cConf.getBoolean(Constants.Explore.EXPLORE_ENABLED)) {
      Closeables.closeQuietly(baseInjector.getInstance(ExploreClient.class));
    }
  }

  @Override
  public void destroy() {
  }
<commit_msg>(CDAP-2569) Always try to create the root ZK node on Master service startup - The “root” node can be namespace by the zk connect string, hence may not exist yet.   - E.g. host:2181/cdap   - By creating “/“ with the given connection string, it make sure the relative root exist.<commit_after>/*
 * Copyright © 2014-2015 <NAME>, Inc.
 *
 * Licensed 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 co.cask.cdap.data.runtime.main;

import co.cask.cdap.api.metrics.MetricsCollectionService;
import co.cask.cdap.app.guice.AppFabricServiceRuntimeModule;
import co.cask.cdap.app.guice.ProgramRunnerRuntimeModule;
import co.cask.cdap.app.guice.ServiceStoreModules;
import co.cask.cdap.app.store.ServiceStore;
import co.cask.cdap.common.conf.CConfiguration;
import co.cask.cdap.common.conf.Constants;
import co.cask.cdap.common.guice.ConfigModule;
import co.cask.cdap.common.guice.DiscoveryRuntimeModule;
import co.cask.cdap.common.guice.IOModule;
import co.cask.cdap.common.guice.KafkaClientModule;
import co.cask.cdap.common.guice.LocationRuntimeModule;
import co.cask.cdap.common.guice.TwillModule;
import co.cask.cdap.common.guice.ZKClientModule;
import co.cask.cdap.common.kerberos.SecurityUtil;
import co.cask.cdap.common.runtime.DaemonMain;
import co.cask.cdap.common.service.RetryOnStartFailureService;
import co.cask.cdap.common.service.RetryStrategies;
import co.cask.cdap.common.utils.DirUtils;
import co.cask.cdap.data.runtime.DataFabricModules;
import co.cask.cdap.data.runtime.DataSetServiceModules;
import co.cask.cdap.data.runtime.DataSetsModules;
import co.cask.cdap.data.stream.StreamAdminModules;
import co.cask.cdap.data2.datafabric.dataset.service.DatasetService;
import co.cask.cdap.data2.util.hbase.ConfigurationTable;
import co.cask.cdap.data2.util.hbase.HBaseTableUtil;
import co.cask.cdap.explore.client.ExploreClient;
import co.cask.cdap.explore.guice.ExploreClientModule;
import co.cask.cdap.explore.service.ExploreServiceUtils;
import co.cask.cdap.gateway.auth.AuthModule;
import co.cask.cdap.internal.app.services.AppFabricServer;
import co.cask.cdap.logging.appender.LogAppenderInitializer;
import co.cask.cdap.logging.guice.LoggingModules;
import co.cask.cdap.metrics.guice.MetricsClientRuntimeModule;
import co.cask.cdap.notifications.feeds.guice.NotificationFeedServiceRuntimeModule;
import co.cask.cdap.notifications.guice.NotificationServiceRuntimeModule;
import com.google.common.base.Charsets;
import com.google.common.base.Supplier;
import com.google.common.base.Throwables;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.google.common.io.Closeables;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.Service;
import com.google.inject.Guice;
import com.google.inject.Injector;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.security.User;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.twill.api.ElectionHandler;
import org.apache.twill.api.TwillApplication;
import org.apache.twill.api.TwillController;
import org.apache.twill.api.TwillPreparer;
import org.apache.twill.api.TwillRunnerService;
import org.apache.twill.api.logging.PrinterLogHandler;
import org.apache.twill.common.Cancellable;
import org.apache.twill.common.ServiceListenerAdapter;
import org.apache.twill.common.Threads;
import org.apache.twill.internal.zookeeper.LeaderElection;
import org.apache.twill.kafka.client.KafkaClientService;
import org.apache.twill.zookeeper.ZKClientService;
import org.apache.twill.zookeeper.ZKOperations;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.PrintWriter;
import java.io.Writer;
import java.net.InetAddress;
import java.net.URL;
import java.net.UnknownHostException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import javax.annotation.Nullable;

/**
 * Driver class for starting all master services.
 * AppFabricHttpService
 * TwillRunnables: MetricsProcessor, MetricsHttp, LogSaver, TransactionService, StreamHandler.
 */
public class MasterServiceMain extends DaemonMain {
  private static final Logger LOG = LoggerFactory.getLogger(MasterServiceMain.class);

  private static final long MAX_BACKOFF_TIME_MS = TimeUnit.MILLISECONDS.convert(10, TimeUnit.MINUTES);
  private static final long SUCCESSFUL_RUN_DURATON_MS = TimeUnit.MILLISECONDS.convert(20, TimeUnit.MINUTES);

  // Maximum time to try looking up the existing twill application
  private static final long LOOKUP_ATTEMPT_TIMEOUT_MS = 2000;

  private final CConfiguration cConf;
  private final Configuration hConf;
  private final Injector baseInjector;
  private final ZKClientService zkClient;
  private final TwillRunnerService twillRunner;
  private final KafkaClientService kafkaClient;
  private final MetricsCollectionService metricsCollectionService;
  private final ServiceStore serviceStore;
  private final LeaderElection leaderElection;
  private final TokenSecureStoreUpdater secureStoreUpdater;

  private volatile boolean stopped;

  public static void main(final String[] args) throws Exception {
    LOG.info("Starting {}", MasterServiceMain.class.getSimpleName());
    new MasterServiceMain().doMain(args);
  }

  public MasterServiceMain() {
    this.cConf = CConfiguration.create();
    this.cConf.set(Constants.Dataset.Manager.ADDRESS, getLocalHost().getCanonicalHostName());
    this.hConf = HBaseConfiguration.create();

    Injector injector = createBaseInjector(cConf, hConf);
    this.baseInjector = injector;
    this.zkClient = injector.getInstance(ZKClientService.class);
    this.twillRunner = injector.getInstance(TwillRunnerService.class);
    this.kafkaClient = injector.getInstance(KafkaClientService.class);
    this.metricsCollectionService = injector.getInstance(MetricsCollectionService.class);
    this.serviceStore = injector.getInstance(ServiceStore.class);
    this.secureStoreUpdater = baseInjector.getInstance(TokenSecureStoreUpdater.class);
    this.leaderElection = createLeaderElection();
  }

  @Override
  public void init(String[] args) {
    cleanupTempDir();

    checkExploreRequirements();
    login();
  }

  @Override
  public void start() {
    createSystemHBaseNamespace();
    updateConfigurationTable();

    LogAppenderInitializer logAppenderInitializer = baseInjector.getInstance(LogAppenderInitializer.class);
    logAppenderInitializer.initialize();

    zkClient.startAndWait();
    twillRunner.startAndWait();
    kafkaClient.startAndWait();
    metricsCollectionService.startAndWait();
    serviceStore.startAndWait();
    leaderElection.startAndWait();

    // Tries to create the ZK root node (which can be namespaced through the zk connection string)
    Futures.getUnchecked(ZKOperations.ignoreError(zkClient.create("/", null, CreateMode.PERSISTENT),
                                                  KeeperException.NodeExistsException.class, null));
  }

  @Override
  public void stop() {
    LOG.info("Stopping {}", Constants.Service.MASTER_SERVICES);
    stopped = true;

    stopQuietly(leaderElection);
    stopQuietly(serviceStore);
    stopQuietly(metricsCollectionService);
    stopQuietly(kafkaClient);
    stopQuietly(twillRunner);
    stopQuietly(zkClient);

    if (cConf.getBoolean(Constants.Explore.EXPLORE_ENABLED)) {
      Closeables.closeQuietly(baseInjector.getInstance(ExploreClient.class));
    }
  }

  @Override
  public void destroy() {
  }

{'fraction_non_alphanumeric': 0.0985546136488382, 'fraction_numerical': 0.0035372324205647375, 'mean_word_length': 5.40546875, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7663290', 'n_tokens_mistral': 5104, 'n_tokens_neox': 5060, 'n_words': 957}