<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

import net.elodina.mesos.util.{Strings, Period, Version, Repr}
import java.util.concurrent.ConcurrentHashMap
import org.apache.mesos.Protos._
import org.apache.mesos.{MesosSchedulerDriver, SchedulerDriver}
import java.util
import com.google.protobuf.ByteString
import java.util.{Collections, Date}
import scala.collection.JavaConversions._
import org.apache.log4j._
import scala.Some
import org.apache.mesos.Protos.Environment.Variable
import scala.collection.mutable


object Scheduler extends org.apache.mesos.Scheduler {
  private val logger: Logger = Logger.getLogger(this.getClass)
  val version: Version = new Version("0.9.5.1")

  val cluster: Cluster = new Cluster()
  private var driver: SchedulerDriver = null

  val logs = new ConcurrentHashMap[Long, Option[String]]()

  private[kafka] def newExecutor(broker: Broker): ExecutorInfo = {
    var cmd = "java -cp " + HttpServer.jar.getName
    cmd += " -Xmx" + broker.heap + "m"
    if (broker.jvmOptions != null) cmd += " " + broker.jvmOptions.replace("$id", broker.id)

    if (Config.debug) cmd += " -Ddebug"
    cmd += " ly.stealth.mesos.kafka.Executor"

    val commandBuilder = CommandInfo.newBuilder
    if (Config.jre != null) {
      commandBuilder.addUris(CommandInfo.URI.newBuilder().setValue(Config.api + "/jre/" + Config.jre.getName))
      cmd = "jre/bin/" + cmd
    }

    val env = new mutable.HashMap[String, String]()
    env("MESOS_SYSLOG_TAG") = Config.frameworkName + "-" + broker.id
    if (broker.syslog) env("MESOS_SYSLOG") = "true"

    val envBuilder = Environment.newBuilder()
    for ((name, value) <- env)
      envBuilder.addVariables(Variable.newBuilder().setName(name).setValue(value))
    commandBuilder.setEnvironment(envBuilder)

    commandBuilder
      .addUris(CommandInfo.URI.newBuilder().setValue(Config.api + "/jar/" + HttpServer.jar.getName).setExtract(false))
      .addUris(CommandInfo.URI.newBuilder().setValue(Config.api + "/kafka/" + HttpServer.kafkaDist.getName))
      .setValue(cmd)

    ExecutorInfo.newBuilder()
      .setExecutorId(ExecutorID.newBuilder.setValue(Broker.nextExecutorId(broker)))
      .setCommand(commandBuilder)
      .setName("broker-" + broker.id)
      .build()
  }

  private[kafka] def newTask(broker: Broker, offer: Offer, reservation: Broker.Reservation): TaskInfo = {
    def taskData: ByteString = {
      var defaults: Map[String, String] = Map(
        "broker.id" -> broker.id,
        "port" -> ("" + reservation.port),
        "log.dirs" -> "kafka-logs",
        "log.retention.bytes" -> ("" + 10l * 1024 * 1024 * 1024),

        "zookeeper.connect" -> Config.zk,
        "host.name" -> offer.getHostname
      )

      if (HttpServer.kafkaVersion.compareTo(new Version("0.9")) >= 0)
        defaults += ("listeners" -> s"PLAINTEXT://:${reservation.port}")

      if (reservation.volume != null)
        defaults += ("log.dirs" -> "data/kafka-logs")

      val data = new util.HashMap[String, String]()
      data.put("broker", "" + broker.toJson(false))
      data.put("defaults", Strings.formatMap(defaults))
      ByteString.copyFromUtf8(Strings.formatMap(data))
    }

    val taskBuilder: TaskInfo.Builder = TaskInfo.newBuilder
      .setName(Config.frameworkName + "-" + broker.id)
      .setTaskId(TaskID.newBuilder.setValue(Broker.nextTaskId(broker)).build)<commit_msg>Changed executor name and set source to framework name and broker id<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

import net.elodina.mesos.util.{Strings, Period, Version, Repr}
import java.util.concurrent.ConcurrentHashMap
import org.apache.mesos.Protos._
import org.apache.mesos.{MesosSchedulerDriver, SchedulerDriver}
import java.util
import com.google.protobuf.ByteString
import java.util.{Collections, Date}
import scala.collection.JavaConversions._
import org.apache.log4j._
import scala.Some
import org.apache.mesos.Protos.Environment.Variable
import scala.collection.mutable


object Scheduler extends org.apache.mesos.Scheduler {
  private val logger: Logger = Logger.getLogger(this.getClass)
  val version: Version = new Version("0.9.5.1")

  val cluster: Cluster = new Cluster()
  private var driver: SchedulerDriver = null

  val logs = new ConcurrentHashMap[Long, Option[String]]()

  private[kafka] def newExecutor(broker: Broker): ExecutorInfo = {
    var cmd = "java -cp " + HttpServer.jar.getName
    cmd += " -Xmx" + broker.heap + "m"
    if (broker.jvmOptions != null) cmd += " " + broker.jvmOptions.replace("$id", broker.id)

    if (Config.debug) cmd += " -Ddebug"
    cmd += " ly.stealth.mesos.kafka.Executor"

    val commandBuilder = CommandInfo.newBuilder
    if (Config.jre != null) {
      commandBuilder.addUris(CommandInfo.URI.newBuilder().setValue(Config.api + "/jre/" + Config.jre.getName))
      cmd = "jre/bin/" + cmd
    }

    val env = new mutable.HashMap[String, String]()
    env("MESOS_SYSLOG_TAG") = Config.frameworkName + "-" + broker.id
    if (broker.syslog) env("MESOS_SYSLOG") = "true"

    val envBuilder = Environment.newBuilder()
    for ((name, value) <- env)
      envBuilder.addVariables(Variable.newBuilder().setName(name).setValue(value))
    commandBuilder.setEnvironment(envBuilder)

    commandBuilder
      .addUris(CommandInfo.URI.newBuilder().setValue(Config.api + "/jar/" + HttpServer.jar.getName).setExtract(false))
      .addUris(CommandInfo.URI.newBuilder().setValue(Config.api + "/kafka/" + HttpServer.kafkaDist.getName))
      .setValue(cmd)

    ExecutorInfo.newBuilder()
      .setExecutorId(ExecutorID.newBuilder.setValue(Broker.nextExecutorId(broker)))
      .setCommand(commandBuilder)
      .setName("kafka.broker")
      .setSource(Config.frameworkName + ".broker." + broker.id)
      .build()
  }

  private[kafka] def newTask(broker: Broker, offer: Offer, reservation: Broker.Reservation): TaskInfo = {
    def taskData: ByteString = {
      var defaults: Map[String, String] = Map(
        "broker.id" -> broker.id,
        "port" -> ("" + reservation.port),
        "log.dirs" -> "kafka-logs",
        "log.retention.bytes" -> ("" + 10l * 1024 * 1024 * 1024),

        "zookeeper.connect" -> Config.zk,
        "host.name" -> offer.getHostname
      )

      if (HttpServer.kafkaVersion.compareTo(new Version("0.9")) >= 0)
        defaults += ("listeners" -> s"PLAINTEXT://:${reservation.port}")

      if (reservation.volume != null)
        defaults += ("log.dirs" -> "data/kafka-logs")

      val data = new util.HashMap[String, String]()
      data.put("broker", "" + broker.toJson(false))
      data.put("defaults", Strings.formatMap(defaults))
      ByteString.copyFromUtf8(Strings.formatMap(data))
    }

    val taskBuilder: TaskInfo.Builder = TaskInfo.newBuilder
      .setName(Config.frameworkName + "-" + broker.id)
      .setTaskId(TaskID.newBuilder.setValue(Broker.nextTaskId(broker)).build)
{'fraction_non_alphanumeric': 0.1286086133459536, 'fraction_numerical': 0.006389020350212968, 'mean_word_length': 3.657300275482094, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 21, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 2, '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': '2552272', 'n_tokens_mistral': 2618, 'n_tokens_neox': 2567, 'n_words': 628}