<commit_before>/**
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You 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 ly.stealth.mesos.kafka.executor

import java.io.{File, FileInputStream}
import java.net.{URL, URLClassLoader}
import java.util
import java.util.Properties
import ly.stealth.mesos.kafka.Util.BindAddress
import ly.stealth.mesos.kafka.Broker
import net.elodina.mesos.util.{IO, Version}
import org.apache.log4j.Logger
import scala.collection.JavaConversions._


case class LaunchConfig(
  id: String,
  options: Map[String, String] = Map(),
  syslog: Boolean = false,
  log4jOptions: Map[String, String] = Map(),
  bindAddress: BindAddress = null,
  defaults: Map[String, String] = Map()) {

  def interpolatedOptions: Map[String, String] = {
    var result = Map[String, String]()
    if (defaults != null)
      result ++= defaults

    result ++= options

    if (bindAddress != null)
      result += ("host.name" -> bindAddress.resolve())

    result.mapValues(v => v.replace("$id", id))
  }

  def interpolatedLog4jOptions: Map[String, String] = {
    log4jOptions.mapValues(v => v.replace("$id", id))
  }
}


abstract class BrokerServer {
  def isStarted: Boolean
  def start(launchInfo: LaunchConfig, send: Broker.Metrics => Unit): Broker.Endpoint
  def stop(): Unit
  def waitFor(): Unit
  def getClassLoader: ClassLoader
}

class KafkaServer extends BrokerServer {
  val logger = Logger.getLogger(classOf[BrokerServer])
  @volatile var server: Object = null
  @volatile var collector: MetricCollectorProxy = null

  def isStarted: Boolean = server != null

  def start(config: LaunchConfig, send: Broker.Metrics => Unit): Broker.Endpoint = {
    if (isStarted) throw new IllegalStateException("started")

    BrokerServer.distro.configureLog4j(config)
    val options = config.interpolatedOptions
    BrokerServer.distro.startReporters(options)

    logger.info("Starting KafkaServer")
    server = BrokerServer.distro.newServer(options)
    server.getClass.getMethod("startup").invoke(server)

    collector = BrokerServer.distro.startCollector(server, send)

    new Broker.Endpoint(options("host.name"), Integer.parseInt(options("port")))
  }

  def stop(): Unit = {
    if (!isStarted) throw new IllegalStateException("!started")

    logger.info("Stopping KafkaServer")
    collector.shutdown()
    server.getClass.getMethod("shutdown").invoke(server)

    waitFor()
    server = null
    collector = null
  }

  def waitFor(): Unit = {
    if (server != null)
      server.getClass.getMethod("awaitShutdown").invoke(server)
  }

  def getClassLoader: ClassLoader = KafkaServer.Distro.loader
}

trait BaseDistro {
  def newServer(options: Map[String, String]): Object
  def startReporters(options: Map[String, String]): Object
  def startCollector(server: AnyRef, send: Broker.Metrics => Unit, interval: Int = 60): MetricCollectorProxy
  def configureLog4j(config: LaunchConfig): Unit
  val dir: File
}

object KafkaServer {
  object Distro extends BaseDistro {
    val (dir, loader) = init()
<commit_msg>Fixed bug with stop() being called twice<commit_after>/**
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You 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 ly.stealth.mesos.kafka.executor

import java.io.{File, FileInputStream}
import java.net.{URL, URLClassLoader}
import java.util
import java.util.Properties
import java.util.concurrent.atomic.AtomicBoolean
import ly.stealth.mesos.kafka.Util.BindAddress
import ly.stealth.mesos.kafka.Broker
import net.elodina.mesos.util.{IO, Version}
import org.apache.log4j.Logger
import scala.collection.JavaConversions._


case class LaunchConfig(
  id: String,
  options: Map[String, String] = Map(),
  syslog: Boolean = false,
  log4jOptions: Map[String, String] = Map(),
  bindAddress: BindAddress = null,
  defaults: Map[String, String] = Map()) {

  def interpolatedOptions: Map[String, String] = {
    var result = Map[String, String]()
    if (defaults != null)
      result ++= defaults

    result ++= options

    if (bindAddress != null)
      result += ("host.name" -> bindAddress.resolve())

    result.mapValues(v => v.replace("$id", id))
  }

  def interpolatedLog4jOptions: Map[String, String] = {
    log4jOptions.mapValues(v => v.replace("$id", id))
  }
}


abstract class BrokerServer {
  def isStarted: Boolean
  def start(launchInfo: LaunchConfig, send: Broker.Metrics => Unit): Broker.Endpoint
  def stop(): Unit
  def waitFor(): Unit
  def getClassLoader: ClassLoader
}

class KafkaServer extends BrokerServer {
  val logger = Logger.getLogger(classOf[BrokerServer])
  @volatile var server: Object = _
  @volatile private var collector: MetricCollectorProxy = _
  private val stopping = new AtomicBoolean(false)

  def isStarted: Boolean = server != null

  def start(config: LaunchConfig, send: Broker.Metrics => Unit): Broker.Endpoint = {
    if (isStarted) throw new IllegalStateException("started")

    BrokerServer.distro.configureLog4j(config)
    val options = config.interpolatedOptions
    BrokerServer.distro.startReporters(options)

    logger.info("Starting KafkaServer")
    server = BrokerServer.distro.newServer(options)
    server.getClass.getMethod("startup").invoke(server)

    collector = BrokerServer.distro.startCollector(server, send)

    new Broker.Endpoint(options("host.name"), Integer.parseInt(options("port")))
  }

  def stop(): Unit = {
    val wasStopping = stopping.getAndSet(true)
    if (!isStarted || wasStopping)
      return

    logger.info("Stopping KafkaServer")
    Option(collector).foreach { _.shutdown() }
    Option(server).foreach { s => s.getClass.getMethod("shutdown").invoke(s) }

    waitFor()
    server = null
    collector = null
  }

  def waitFor(): Unit = {
    if (server != null)
      server.getClass.getMethod("awaitShutdown").invoke(server)
  }

  def getClassLoader: ClassLoader = KafkaServer.Distro.loader
}

trait BaseDistro {
  def newServer(options: Map[String, String]): Object
  def startReporters(options: Map[String, String]): Object
  def startCollector(server: AnyRef, send: Broker.Metrics => Unit, interval: Int = 60): MetricCollectorProxy
  def configureLog4j(config: LaunchConfig): Unit
  val dir: File
}

object KafkaServer {
  object Distro extends BaseDistro {
    val (dir, loader) = init()

{'fraction_non_alphanumeric': 0.10722186076772934, 'fraction_numerical': 0.0031229668184775536, 'mean_word_length': 4.253588516746412, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 16, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '2892355', 'n_tokens_mistral': 2302, 'n_tokens_neox': 2241, 'n_words': 675}