text:
package kafka.zk

import java.nio.charset.StandardCharsets.UTF_8
import java.util.Properties

import kafka.api.LeaderAndIsr
import kafka.cluster.Broker
import kafka.controller.LeaderIsrAndControllerEpoch
import kafka.log.LogConfig
import kafka.security.auth.SimpleAclAuthorizer.VersionedAcls
import kafka.security.auth.{Acl, Resource, ResourceType}

import kafka.server.ConfigType
import kafka.utils._
import kafka.zookeeper._
import org.apache.kafka.common.TopicPartition
import org.apache.zookeeper.KeeperException.Code
import org.apache.zookeeper.data.{ACL, Stat}
import org.apache.zookeeper.{CreateMode, KeeperException}

import scala.collection.mutable.ArrayBuffer
import scala.collection.{Seq, mutable}

/**
 * Provides higher level Kafka-specific operations on top of the pipelined [[kafka.zookeeper.ZooKeeperClient]].
 *
 * This performs better than [[kafka.utils.ZkUtils]] and should replace it completely, eventually.
 *
 * Implementation note: this class includes methods for various components (Controller, Configs, Old Consumer, etc.)
 * and returns instances of classes from the calling packages in some cases. This is not ideal, but it makes it
 * easier to quickly migrate away from `ZkUtils`. We should revisit this once the migration is completed and tests are
 * in place. We should also consider whether a monolithic [[kafka.zk.ZkData]] is the way to go.
 */
class KafkaZkClient(zooKeeperClient: ZooKeeperClient, isSecure: Boolean) extends Logging {
  import KafkaZkClient._

  def createSequentialPersistentPath(path: String, data: String = ""): String = {
    val createRequest = CreateRequest(path, data.getBytes("UTF-8"), acls(path), CreateMode.PERSISTENT_SEQUENTIAL)
    val createResponse = retryRequestUntilConnected(createRequest)
    createResponse.resultException.foreach(e => throw e)
    createResponse.path
  }

  /**
   * Gets topic partition states for the given partitions.
   * @param partitions the partitions for which we want ot get states.
   * @return sequence of GetDataResponses whose contexts are the partitions they are associated with.
   */
  def getTopicPartitionStatesRaw(partitions: Seq[TopicPartition]): Seq[GetDataResponse] = {
    val getDataRequests = partitions.map { partition =>
      GetDataRequest(TopicPartitionStateZNode.path(partition), ctx = Some(partition))
    }
    retryRequestsUntilConnected(getDataRequests)
  }

  /**
   * Sets topic partition states for the given partitions.
   * @param leaderIsrAndControllerEpochs the partition states of each partition whose state we wish to set.
   * @return sequence of SetDataResponse whose contexts are the partitions they are associated with.
   */
  def setTopicPartitionStatesRaw(leaderIsrAndControllerEpochs: Map[TopicPartition, LeaderIsrAndControllerEpoch]): Seq[SetDataResponse] = {
    val setDataRequests = leaderIsrAndControllerEpochs.map { case (partition, leaderIsrAndControllerEpoch) =>
      val path = TopicPartitionStateZNode.path(partition)
      val data = TopicPartitionStateZNode.encode(leaderIsrAndControllerEpoch)
      SetDataRequest(path, data, leaderIsrAndControllerEpoch.leaderAndIsr.zkVersion, Some(partition))
    }
    retryRequestsUntilConnected(setDataRequests.toSeq)
  }

  /**
   * Creates topic partition state znodes for the given partitions.
   * @param leaderIsrAndControllerEpochs the partition states of each partition whose state we wish to set.
   * @return sequence of CreateResponse whose contexts are the partitions they are associated with.
   */
  def createTopicPartitionStatesRaw(leaderIsrAndControllerEpochs: Map[TopicPartition, LeaderIsrAndControllerEpoch]): Seq[CreateResponse] = {
    createTopicPartitions(leaderIsrAndControllerEpochs.keys.map(_.topic).toSet.toSeq)
    createTopicPartition(leaderIsrAndControllerEpochs.keys.toSeq)
    val createRequests = leaderIsrAndControllerEpochs.map { case (partition, leaderIsrAndControllerEpoch) =>
      val path = TopicPartitionStateZNode.path(partition)
      val data = TopicPartitionStateZNode.encode(leaderIsrAndControllerEpoch)
      CreateRequest(path, data, acls(path), CreateMode.PERSISTENT, Some(partition))
    }
    retryRequestsUntilConnected(createRequests.toSeq)
  }

  /**
   * Sets the controller epoch conditioned on the given epochZkVersion.
   * @param epoch the epoch to set
   * @param epochZkVersion the expected version number of the epoch znode.
   * @return SetDataResponse
   */
  def setControllerEpochRaw(epoch: Int, epochZkVersion: Int): SetDataResponse = {
    val setDataRequest = SetDataRequest(ControllerEpochZNode.path, ControllerEpochZNode.encode(epoch), epochZkVersion)
    retryRequestUntilConnected(setDataRequest)
  }

  /**
   * Creates the controller epoch znode.
   * @param epoch the epoch to set
   * @return CreateResponse
   */
  def createControllerEpochRaw(epoch: Int): CreateResponse = {
    val createRequest = CreateRequest(ControllerEpochZNode.path, ControllerEpochZNode.encode(epoch),
      acls(ControllerEpochZNode.path), CreateMode.PERSISTENT)
    retryRequestUntilConnected(createRequest)
  }

  /**
   * Try to update the partition states of multiple partitions in zookeeper.
   * @param leaderAndIsrs The partition states to update.
   * @param controllerEpoch The current controller epoch.
   * @return UpdateLeaderAndIsrResult instance containing per partition results.
   */
  def updateLeaderAndIsr(leaderAndIsrs: Map[TopicPartition, LeaderAndIsr], controllerEpoch: Int): UpdateLeaderAndIsrResult = {
    val successfulUpdates = mutable.Map.empty[TopicPartition, LeaderAndIsr]
    val updatesToRetry = mutable.Buffer.empty[TopicPartition]
    val failed = mutable.Map.empty[TopicPartition, Exception]
    val leaderIsrAndControllerEpochs = leaderAndIsrs.map { case (partition, leaderAndIsr) => partition -> LeaderIsrAndControllerEpoch(leaderAndIsr, controllerEpoch) }
    val setDataResponses = try {
      setTopicPartitionStatesRaw(leaderIsrAndControllerEpochs)
    } catch {
      case e: Exception =>
        leaderAndIsrs.keys.foreach(partition => failed.put(partition, e))
        return UpdateLeaderAndIsrResult(successfulUpdates.toMap, updatesToRetry, failed.toMap)
    }
    setDataResponses.foreach { setDataResponse =>
      val partition = setDataResponse.ctx.get.asInstanceOf[TopicPartition]
      setDataResponse.resultCode match {
        case Code.OK =>
          val updatedLeaderAndIsr = leaderAndIsrs(partition).withZkVersion(setDataResponse.stat.getVersion)
          successfulUpdates.put(partition, updatedLeaderAndIsr)
        case Code.BADVERSION => updatesToRetry += partition
        case _ => failed.put(partition, setDataResponse.resultException.get)
      }
    }
    UpdateLeaderAndIsrResult(successfulUpdates.toMap, updatesToRetry, failed.toMap)
  }

  /**
   * Get log configs that merge local configs with topic-level configs in zookeeper.
   * @param topics The topics to get log configs for.
   * @param config The local configs.
   * @return A tuple of two values:
   *         1. The successfully gathered log configs
   *         2. Exceptions corresponding to failed log config lookups.
   */
  def getLogConfigs(topics: Seq[String], config: java.util.Map[String, AnyRef]):
  (Map[String, LogConfig], Map[String, Exception]) = {
    val logConfigs = mutable.Map.empty[String, LogConfig]
    val failed = mutable.Map.empty[String, Exception]
    val configResponses = try {
      getTopicConfigs(topics)
    } catch {
      case e: Exception =>
        topics.foreach(topic => failed.put(topic, e))
        return (logConfigs.toMap, failed.toMap)
    }
    configResponses.foreach { configResponse =>
      val topic = configResponse.ctx.get.asInstanceOf[String]
      configResponse.resultCode match {
        case Code.OK =>
          val overrides = ConfigEntityZNode.decode(configResponse.data)
          val logConfig = LogConfig.fromProps(config, overrides)
          logConfigs.put(topic, logConfig)
        case Code.NONODE =>
          val logConfig = LogConfig.fromProps(config, new Properties)
          logConfigs.put(topic, logConfig)
        case _ => failed.put(topic, configResponse.resultException.get)
      }
    }
    (logConfigs.toMap, failed.toMap)
  }

  /**
   * Get entity configs for a given entity name
   * @param rootEntityType entity type
   * @param sanitizedEntityName entity name
   * @return The successfully gathered log configs
   */
  def getEntityConfigs(rootEntityType: String, sanitizedEntityName: String): Properties = {
    val getDataRequest = GetDataRequest(ConfigEntityZNode.path(rootEntityType, sanitizedEntityName))
    val getDataResponse = retryRequestUntilConnected(getDataRequest)

    getDataResponse.resultCode match {
      case Code.OK =>
        ConfigEntityZNode.decode(getDataResponse.data)
      case Code.NONODE => new Properties()
      case _ => throw getDataResponse.resultException.get
    }
  }

  /**
   * Sets or creates the entity znode path with the given configs depending
   * on whether it already exists or not.
   * @param rootEntityType entity type
   * @param sanitizedEntityName entity name
   * @throws KeeperException if there is an error while setting or creating the znode
   */
  def setOrCreateEntityConfigs(rootEntityType: String, sanitizedEntityName: String, config: Properties) = {

    def set(configData: Array[Byte]): SetDataResponse = {
      val setDataRequest = SetDataRequest(ConfigEntityZNode.path(rootEntityType, sanitizedEntityName), ConfigEntityZNode.encode(config), ZkVersion.NoVersion)
      retryRequestUntilConnected(setDataRequest)
    }

    def create(configData: Array[Byte]) = {
      val path = ConfigEntityZNode.path(rootEntityType, sanitizedEntityName)
      createRecursive(path, ConfigEntityZNode.encode(config))
    }

    val configData = ConfigEntityZNode.encode(config)

    val setDataResponse = set(configData)
    setDataResponse.resultCode match {
      case Code.NONODE => create(configData)
      case _ => setDataResponse.resultException.foreach(e => throw e)
    }
  }

  /**
   * Returns all the entities for a given entityType
   * @param entityType entity type
   * @return List of all entity names
   */
  def getAllEntitiesWithConfig(entityType: String): Seq[String] = {
    getChildren(ConfigEntityTypeZNode.path(entityType))
  }

  /**
   * Creates config change notification
   * @param sanitizedEntityPath  sanitizedEntityPath path to write
   * @throws KeeperException if there is an error while setting or creating the znode
   */
  def createConfigChangeNotification(sanitizedEntityPath: String): Unit = {
    val path = ConfigEntityChangeNotificationSequenceZNode.createPath
    val createRequest = CreateRequest(path, ConfigEntityChangeNotificationSequenceZNode.encode(sanitizedEntityPath), acls(path), CreateMode.PERSISTENT_SEQUENTIAL)
    val createResponse = retryRequestUntilConnected(createRequest)
    if (createResponse.resultCode != Code.OK) {
      createResponse.resultException.foreach(e => throw e)
    }
  }

  /**
   * Gets all brokers in the cluster.
   * @return sequence of brokers in the cluster.
   */
  def getAllBrokersInCluster: Seq[Broker] = {
    val brokerIds = getSortedBrokerList
    val getDataRequests = brokerIds.map(brokerId => GetDataRequest(BrokerIdZNode.path(brokerId), ctx = Some(brokerId)))
    val getDataResponses = retryRequestsUntilConnected(getDataRequests)
    getDataResponses.flatMap { getDataResponse =>
      val brokerId = getDataResponse.ctx.get.asInstanceOf[Int]
      getDataResponse.resultCode match {
        case Code.OK =>
          Option(BrokerIdZNode.decode(brokerId, getDataResponse.data))
        case Code.NONODE => None
        case _ => throw getDataResponse.resultException.get
      }
    }
  }

  /**
   * Gets the list of sorted broker Ids
   */
  def getSortedBrokerList(): Seq[Int] =
    getChildren(BrokerIdsZNode.path).map(_.toInt).sorted

  /**
   * Gets all topics in the cluster.
   * @return sequence of topics in the cluster.
   */
  def getAllTopicsInCluster: Seq[String] = {
    val getChildrenResponse = retryRequestUntilConnected(GetChildrenRequest(TopicsZNode.path))
    getChildrenResponse.resultCode match {
      case Code.OK => getChildrenResponse.children
      case Code.NONODE => Seq.empty
      case _ => throw getChildrenResponse.resultException.get
    }

  }

  /**
   * Checks the topic existence
   * @param topicName
   * @return true if topic exists else false
   */
  def topicExists(topicName: String): Boolean = {
    pathExists(TopicZNode.path(topicName))
  }

  /**
   * Sets the topic znode with the given assignment.
   * @param topic the topic whose assignment is being set.
   * @param assignment the partition to replica mapping to set for the given topic
   * @return SetDataResponse
   */
  def setTopicAssignmentRaw(topic: String, assignment: collection.Map[TopicPartition, Seq[Int]]): SetDataResponse = {
    val setDataRequest = SetDataRequest(TopicZNode.path(topic), TopicZNode.encode(assignment), ZkVersion.NoVersion)
    retryRequestUntilConnected(setDataRequest)
  }

  /**
   * Sets the topic znode with the given assignment.
   * @param topic the topic whose assignment is being set.
   * @param assignment the partition to replica mapping to set for the given topic
   * @throws KeeperException if there is an error while setting assignment
   */
  def setTopicAssignment(topic: String, assignment: Map[TopicPartition, Seq[Int]]) = {
    val setDataResponse = setTopicAssignmentRaw(topic, assignment)
    if (setDataResponse.resultCode != Code.OK) {
      setDataResponse.resultException.foreach(e => throw e)
    }
  }

  /**
   * Create the topic znode with the given assignment.
   * @param topic the topic whose assignment is being set.
   * @param assignment the partition to replica mapping to set for the given topic
   * @throws KeeperException if there is an error while creating assignment
   */
  def createTopicAssignment(topic: String, assignment: Map[TopicPartition, Seq[Int]]) = {
    createRecursive(TopicZNode.path(topic), TopicZNode.encode(assignment))
  }

  /**
   * Gets the log dir event notifications as strings. These strings are the znode names and not the absolute znode path.
   * @return sequence of znode names and not the absolute znode path.
   */
  def getAllLogDirEventNotifications: Seq[String] = {
    val getChildrenResponse = retryRequestUntilConnected(GetChildrenRequest(LogDirEventNotificationZNode.path))
    getChildrenResponse.resultCode match {
      case Code.OK => getChildrenResponse.children.map(LogDirEventNotificationSequenceZNode.sequenceNumber)
      case Code.NONODE => Seq.empty
      case _ => throw getChildrenResponse.resultException.get
    }
  }

  /**
   * Reads each of the log dir event notifications associated with the given sequence numbers and extracts the broker ids.
   * @param sequenceNumbers the sequence numbers associated with the log dir event notifications.
   * @return broker ids associated with the given log dir event notifications.
   */
  def getBrokerIdsFromLogDirEvents(sequenceNumbers: Seq[String]): Seq[Int] = {
    val getDataRequests = sequenceNumbers.map { sequenceNumber =>
      GetDataRequest(LogDirEventNotificationSequenceZNode.path(sequenceNumber))
    }
    val getDataResponses = retryRequestsUntilConnected(getDataRequests)
    getDataResponses.flatMap { getDataResponse =>
      getDataResponse.resultCode match {
        case Code.OK => LogDirEventNotificationSequenceZNode.decode(getDataResponse.data)
        case Code.NONODE => None
        case _ => throw getDataResponse.resultException.get
      }
    }
  }

  /**
   * Deletes all log dir event notifications.
   */
  def deleteLogDirEventNotifications(): Unit = {
    val getChildrenResponse = retryRequestUntilConnected(GetChildrenRequest(LogDirEventNotificationZNode.path))
    if (getChildrenResponse.resultCode == Code.OK) {
      deleteLogDirEventNotifications(getChildrenResponse.children)
    } else if (getChildrenResponse.resultCode != Code.NONODE) {
      getChildrenResponse.resultException.foreach(e => throw e)
    }
  }

  /**
   * Deletes the log dir event notifications associated with the given sequence numbers.
   * @param sequenceNumbers the sequence numbers associated with the log dir event notifications to be deleted.
   */
  def deleteLogDirEventNotifications(sequenceNumbers: Seq[String]): Unit = {
    val deleteRequests = sequenceNumbers.map { sequenceNumber =>
      DeleteRequest(LogDirEventNotificationSequenceZNode.path(sequenceNumber), ZkVersion.NoVersion)
    }
    retryRequestsUntilConnected(deleteRequests)
  }

  /**
   * Gets the assignments for the given topics.
   * @param topics the topics whose partitions we wish to get the assignments for.
   * @return the replica assignment for each partition from the given topics.
   */
  def getReplicaAssignmentForTopics(topics: Set[String]): Map[TopicPartition, Seq[Int]] = {
    val getDataRequests = topics.map(topic => GetDataRequest(TopicZNode.path(topic), ctx = Some(topic)))
    val getDataResponses = retryRequestsUntilConnected(getDataRequests.toSeq)
    getDataResponses.flatMap { getDataResponse =>
      val topic = getDataResponse.ctx.get.asInstanceOf[String]
      getDataResponse.resultCode match {
        case Code.OK => TopicZNode.decode(topic, getDataResponse.data)
        case Code.NONODE => Map.empty[TopicPartition, Seq[Int]]
        case _ => throw getDataResponse.resultException.get
      }
    }.toMap
  }

  /**
   * Gets partition the assignments for the given topics.
   * @param topics the topics whose partitions we wish to get the assignments for.
   * @return the partition assignment for each partition from the given topics.
   */
  def getPartitionAssignmentForTopics(topics: Set[String]):  Map[String, Map[Int, Seq[Int]]] = {
    val getDataRequests = topics.map(topic => GetDataRequest(TopicZNode.path(topic), ctx = Some(topic)))
    val getDataResponses = retryRequestsUntilConnected(getDataRequests.toSeq)
    getDataResponses.flatMap { getDataResponse =>
      val topic = getDataResponse.ctx.get.asInstanceOf[String]
       if (getDataResponse.resultCode == Code.OK) {
        val partitionMap = TopicZNode.decode(topic, getDataResponse.data).map { case (k, v) => (k.partition, v) }
        Map(topic -> partitionMap)
      } else if (getDataResponse.resultCode == Code.NONODE) {
        Map.empty[String, Map[Int, Seq[Int]]]
      } else {
        throw getDataResponse.resultException.get
      }
    }.toMap
  }

  /**
   * Gets the partition numbers for the given topics
   * @param topics the topics whose partitions we wish to get.
   * @return the partition array for each topic from the given topics.
   */
  def getPartitionsForTopics(topics: Set[String]): Map[String, Seq[Int]] = {
    getPartitionAssignmentForTopics(topics).map { topicAndPartitionMap =>
      val topic = topicAndPartitionMap._1
      val partitionMap = topicAndPartitionMap._2
      topic -> partitionMap.keys.toSeq.sortWith((s, t) => s < t)
    }
  }

  /**
   * Gets the partition count for a given topic
   * @param topic The topic to get partition count for.
   * @return  optional integer that is Some if the topic exists and None otherwise.
   */
  def getTopicPartitionCount(topic: String): Option[Int] = {
    val topicData = getReplicaAssignmentForTopics(Set(topic))
    if (topicData.nonEmpty)
      Some(topicData.size)
    else
      None
  }

  /**
   * Gets the assigned replicas for a specific topic and partition
   * @param topicPartition TopicAndPartition to get assigned replicas for .
   * @return List of assigned replicas
   */
  def getReplicasForPartition(topicPartition: TopicPartition): Seq[Int] = {
    val topicData = getReplicaAssignmentForTopics(Set(topicPartition.topic))
    topicData.getOrElse(topicPartition, Seq.empty)
  }

  /**
   * Gets the data and version at the given zk path
   * @param path zk node path
   * @return A tuple of 2 elements, where first element is zk node data as string
   *         and second element is zk node version.
   *         returns (None, ZkVersion.NoVersion) if node doesn't exists and throws exception for any error
   */
  def getDataAndVersion(path: String): (Option[String], Int) = {
    val (data, stat) = getDataAndStat(path)
    stat match {
      case ZkStat.NoStat => (data, ZkVersion.NoVersion)
      case _ => (data, stat.getVersion)
    }
  }

  /**
   * Gets the data and Stat at the given zk path
   * @param path zk node path
   * @return A tuple of 2 elements, where first element is zk node data as string
   *         and second element is zk node stats.
   *         returns (None, ZkStat.NoStat) if node doesn't exists and throws exception for any error
   */
  def getDataAndStat(path: String): (Option[String], Stat) = {
    val getDataRequest = GetDataRequest(path)
    val getDataResponse = retryRequestUntilConnected(getDataRequest)

    getDataResponse.resultCode match {
      case Code.OK =>
        if (getDataResponse.data == null)
          (None, getDataResponse.stat)
        else {
          val data = Option(getDataResponse.data).map(new String(_, UTF_8))
          (data, getDataResponse.stat)
        }
      case Code.NONODE => (None, ZkStat.NoStat)
      case _ => throw getDataResponse.resultException.get
    }
  }

  /**
   * Gets all the child nodes at a given zk node path
   * @param path
   * @return list of child node names
   */
  def getChildren(path : String): Seq[String] = {
    val getChildrenResponse = retryRequestUntilConnected(GetChildrenRequest(path))
    getChildrenResponse.resultCode match {
      case Code.OK => getChildrenResponse.children
      case Code.NONODE => Seq.empty
      case _ => throw getChildrenResponse.resultException.get
    }
  }

  /**
   * Conditional update the persistent path data, return (true, newVersion) if it succeeds, otherwise (the path doesn't
   * exist, the current version is not the expected version, etc.) return (false, ZkVersion.NoVersion)
   *
   * When there is a ConnectionLossException during the conditional update, ZookeeperClient will retry the update and may fail
   * since the previous update may have succeeded (but the stored zkVersion no longer matches the expected one).
   * In this case, we will run the optionalChecker to further check if the previous write did indeed succeeded.
   */
  def conditionalUpdatePath(path: String, data: String, expectVersion: Int,
                            optionalChecker:Option[(KafkaZkClient, String, String) => (Boolean,Int)] = None): (Boolean, Int) = {

    val setDataRequest = SetDataRequest(path, data.getBytes(UTF_8), expectVersion)
    val setDataResponse = retryRequestUntilConnected(setDataRequest)

    setDataResponse.resultCode match {
      case Code.OK =>
        debug("Conditional update of path %s with value %s and expected version %d succeeded, returning the new version: %d"
          .format(path, data, expectVersion, setDataResponse.stat.getVersion))
        (true, setDataResponse.stat.getVersion)

      case Code.BADVERSION =>
        optionalChecker match {
          case Some(checker) => checker(this, path, data)
          case _ =>
            debug("Checker method is not passed skipping zkData match")
            debug("Conditional update of path %s with data %s and expected version %d failed due to %s"
              .format(path, data, expectVersion, setDataResponse.resultException.get.getMessage))
            (false, ZkVersion.NoVersion)
        }

      case Code.NONODE =>
        debug("Conditional update of path %s with data %s and expected version %d failed due to %s".format(path, data,
          expectVersion, setDataResponse.resultException.get.getMessage))
        (false, ZkVersion.NoVersion)

      case _ =>
        debug("Conditional update of path %s with data %s and expected version %d failed due to %s".format(path, data,
          expectVersion, setDataResponse.resultException.get.getMessage))
        throw setDataResponse.resultException.get
    }
  }

  /**
   * Creates the delete topic znode.
   * @param topicName topic name
   * @throws KeeperException if there is an error while setting or creating the znode
   */
  def createDeleteTopicPath(topicName: String): Unit = {
    createRecursive(DeleteTopicsTopicZNode.path(topicName))
  }

  /**
   * Checks if topic is marked for deletion
   * @param topic
   * @return true if topic is marked for deletion, else false
   */
  def isTopicMarkedForDeletion(topic: String): Boolean = {
    pathExists(DeleteTopicsTopicZNode.path(topic))
  }

  /**
   * Get all topics marked for deletion.
   * @return sequence of topics marked for deletion.
   */
  def getTopicDeletions: Seq[String] = {
    val getChildrenResponse = retryRequestUntilConnected(GetChildrenRequest(DeleteTopicsZNode.path))
    getChildrenResponse.resultCode match {
      case Code.OK => getChildrenResponse.children
      case Code.NONODE => Seq.empty
      case _ => throw getChildrenResponse.resultException.get
    }
  }

  /**
   * Remove the given topics from the topics marked for deletion.
   * @param topics the topics to remove.
   */
  def deleteTopicDeletions(topics: Seq[String]): Unit = {
    val deleteRequests = topics.map(topic => DeleteRequest(DeleteTopicsTopicZNode.path(topic), ZkVersion.NoVersion))
    retryRequestsUntilConnected(deleteRequests)
  }

  /**
   * Returns all reassignments.
   * @return the reassignments for each partition.
   */
  def getPartitionReassignment: Map[TopicPartition, Seq[Int]] = {
    val getDataRequest = GetDataRequest(ReassignPartitionsZNode.path)
    val getDataResponse = retryRequestUntilConnected(getDataRequest)
    getDataResponse.resultCode match {
      case  Code.OK => ReassignPartitionsZNode.decode(getDataResponse.data)
      case Code.NONODE => Map.empty
      case _ => throw getDataResponse.resultException.get
    }
  }

  /**
   * Sets or creates the partition reassignment znode with the given reassignment depending on whether it already
   * exists or not.
   *
   * @param reassignment the reassignment to set on the reassignment znode
   * @throws KeeperException if there is an error while setting or creating the znode
   */
  def setOrCreatePartitionReassignment(reassignment: collection.Map[TopicPartition, Seq[Int]]): Unit = {

    def set(reassignmentData: Array[Byte]): SetDataResponse = {
      val setDataRequest = SetDataRequest(ReassignPartitionsZNode.path, reassignmentData, ZkVersion.NoVersion)
      retryRequestUntilConnected(setDataRequest)
    }

    def create(reassignmentData: Array[Byte]): CreateResponse = {
      val createRequest = CreateRequest(ReassignPartitionsZNode.path, reassignmentData, acls(ReassignPartitionsZNode.path),
        CreateMode.PERSISTENT)
      retryRequestUntilConnected(createRequest)
    }

    val reassignmentData = ReassignPartitionsZNode.encode(reassignment)
    val setDataResponse = set(reassignmentData)
    setDataResponse.resultCode match {
      case Code.NONODE =>
        val createDataResponse = create(reassignmentData)
        createDataResponse.resultException.foreach(e => throw e)
      case _ => setDataResponse.resultException.foreach(e => throw e)
    }
  }

  /**
   * Creates the partition reassignment znode with the given reassignment.
   * @param reassignment the reassignment to set on the reassignment znode.
   * @throws KeeperException if there is an error while setting or creating the znode
   */
  def createPartitionReassignment(reassignment: Map[TopicPartition, Seq[Int]])  = {
    val createRequest = CreateRequest(ReassignPartitionsZNode.path, ReassignPartitionsZNode.encode(reassignment),
      acls(ReassignPartitionsZNode.path), CreateMode.PERSISTENT)
    val createResponse = retryRequestUntilConnected(createRequest)

    if (createResponse.resultCode != Code.OK) {
      throw createResponse.resultException.get
    }
  }

  /**
   * Deletes the partition reassignment znode.
   */
  def deletePartitionReassignment(): Unit = {
    val deleteRequest = DeleteRequest(ReassignPartitionsZNode.path, ZkVersion.NoVersion)
    retryRequestUntilConnected(deleteRequest)
  }

  /**
   * Checks if reassign partitions is in progress
   * @return true if reassign partitions is in progress, else false
   */
  def reassignPartitionsInProgress(): Boolean = {
    pathExists(ReassignPartitionsZNode.path)
  }

  /**
   * Gets topic partition states for the given partitions.
   * @param partitions the partitions for which we want to get states.
   * @return map containing LeaderIsrAndControllerEpoch of each partition for we were able to lookup the partition state.
   */
  def getTopicPartitionStates(partitions: Seq[TopicPartition]): Map[TopicPartition, LeaderIsrAndControllerEpoch] = {
    val getDataResponses = getTopicPartitionStatesRaw(partitions)
    getDataResponses.flatMap { getDataResponse =>
      val partition = getDataResponse.ctx.get.asInstanceOf[TopicPartition]
      getDataResponse.resultCode match {
        case Code.OK => TopicPartitionStateZNode.decode(getDataResponse.data, getDataResponse.stat).map(partition -> _)
        case Code.NONODE => None
        case _ => throw getDataResponse.resultException.get
      }
    }.toMap
  }

  /**
   * Gets topic partition state for the given partition.
   * @param partition the partition for which we want to get state.
   * @return LeaderIsrAndControllerEpoch of the partition state if exists, else None
   */
  def getTopicPartitionState(partition: TopicPartition): Option[LeaderIsrAndControllerEpoch] = {
    val getDataResponse = getTopicPartitionStatesRaw(Seq(partition)).head
    if (getDataResponse.resultCode == Code.OK) {
      TopicPartitionStateZNode.decode(getDataResponse.data, getDataResponse.stat)
    } else if (getDataResponse.resultCode == Code.NONODE) {
      None
    } else {
      throw getDataResponse.resultException.get
    }
  }

  /**
   * Gets the leader for a given partition
   * @param partition
   * @return optional integer if the leader exists and None otherwise.
   */
  def getLeaderForPartition(partition: TopicPartition): Option[Int] = {
    val leaderIsrEpoch = getTopicPartitionState(partition)
    if (leaderIsrEpoch.isDefined)
      Option(leaderIsrEpoch.get.leaderAndIsr.leader)
    else
      None
  }

  /**
   * Gets the isr change notifications as strings. These strings are the znode names and not the absolute znode path.
   * @return sequence of znode names and not the absolute znode path.
   */
  def getAllIsrChangeNotifications: Seq[String] = {
    val getChildrenResponse = retryRequestUntilConnected(GetChildrenRequest(IsrChangeNotificationZNode.path))
    getChildrenResponse.resultCode match {
      case Code.OK => getChildrenResponse.children.map(IsrChangeNotificationSequenceZNode.sequenceNumber)
      case Code.NONODE => Seq.empty
      case _ => throw getChildrenResponse.resultException.get
    }
  }

  /**
   * Reads each of the isr change notifications associated with the given sequence numbers and extracts the partitions.
   * @param sequenceNumbers the sequence numbers associated with the isr change notifications.
   * @return partitions associated with the given isr change notifications.
   */
  def getPartitionsFromIsrChangeNotifications(sequenceNumbers: Seq[String]): Seq[TopicPartition] = {
    val getDataRequests = sequenceNumbers.map { sequenceNumber =>
      GetDataRequest(IsrChangeNotificationSequenceZNode.path(sequenceNumber))
    }
    val getDataResponses = retryRequestsUntilConnected(getDataRequests)
    getDataResponses.flatMap { getDataResponse =>
      getDataResponse.resultCode match {
        case Code.OK => IsrChangeNotificationSequenceZNode.decode(getDataResponse.data)
        case Code.NONODE => None
        case _ => throw getDataResponse.resultException.get
      }
    }
  }

  /**
   * Deletes all isr change notifications.
   */
  def deleteIsrChangeNotifications(): Unit = {
    val getChildrenResponse = retryRequestUntilConnected(GetChildrenRequest(IsrChangeNotificationZNode.path))
    if (getChildrenResponse.resultCode == Code.OK) {
      deleteIsrChangeNotifications(getChildrenResponse.children.map(IsrChangeNotificationSequenceZNode.sequenceNumber))
    } else if (getChildrenResponse.resultCode != Code.NONODE) {
      getChildrenResponse.resultException.foreach(e => throw e)
    }
  }

  /**
   * Deletes the isr change notifications associated with the given sequence numbers.
   * @param sequenceNumbers the sequence numbers associated with the isr change notifications to be deleted.
   */
  def deleteIsrChangeNotifications(sequenceNumbers: Seq[String]): Unit = {
    val deleteRequests = sequenceNumbers.map { sequenceNumber =>
      DeleteRequest(IsrChangeNotificationSequenceZNode.path(sequenceNumber), ZkVersion.NoVersion)
    }
    retryRequestsUntilConnected(deleteRequests)
  }

  /**
   * Gets the partitions marked for preferred replica election.
   * @return sequence of partitions.
   */
  def getPreferredReplicaElection: Set[TopicPartition] = {
    val getDataRequest = GetDataRequest(PreferredReplicaElectionZNode.path)
    val getDataResponse = retryRequestUntilConnected(getDataRequest)
    getDataResponse.resultCode match {
      case Code.OK => PreferredReplicaElectionZNode.decode(getDataResponse.data)
      case Code.NONODE => Set.empty
      case _ => throw getDataResponse.resultException.get
    }
  }

  /**
   * Deletes the preferred replica election znode.
   */
  def deletePreferredReplicaElection(): Unit = {
    val deleteRequest = DeleteRequest(PreferredReplicaElectionZNode.path, ZkVersion.NoVersion)
    retryRequestUntilConnected(deleteRequest)
  }

  /**
   * Gets the controller id.
   * @return optional integer that is Some if the controller znode exists and can be parsed and None otherwise.
   */
  def getControllerId: Option[Int] = {
    val getDataRequest = GetDataRequest(ControllerZNode.path)
    val getDataResponse = retryRequestUntilConnected(getDataRequest)
    getDataResponse.resultCode match {
      case Code.OK => ControllerZNode.decode(getDataResponse.data)
      case Code.NONODE => None
      case _ => throw getDataResponse.resultException.get
    }
  }

  /**
   * Deletes the controller znode.
   */
  def deleteController(): Unit = {
    val deleteRequest = DeleteRequest(ControllerZNode.path, ZkVersion.NoVersion)
    retryRequestUntilConnected(deleteRequest)
  }

  /**
   * Gets the controller epoch.
   * @return optional (Int, Stat) that is Some if the controller epoch path exists and None otherwise.
   */
  def getControllerEpoch: Option[(Int, Stat)] = {
    val getDataRequest = GetDataRequest(ControllerEpochZNode.path)
    val getDataResponse = retryRequestUntilConnected(getDataRequest)
    getDataResponse.resultCode match {
      case Code.OK =>
        val epoch = ControllerEpochZNode.decode(getDataResponse.data)
        Option(epoch, getDataResponse.stat)
      case Code.NONODE => None
      case _ => throw getDataResponse.resultException.get
    }
  }

  /**
   * Recursively deletes the topic znode.
   * @param topic the topic whose topic znode we wish to delete.
   */
  def deleteTopicZNode(topic: String): Unit = {
    deleteRecursive(TopicZNode.path(topic))
  }

  /**
   * Deletes the topic configs for the given topics.
   * @param topics the topics whose configs we wish to delete.
   */
  def deleteTopicConfigs(topics: Seq[String]): Unit = {
    val deleteRequests = topics.map(topic => DeleteRequest(ConfigEntityZNode.path(ConfigType.Topic, topic), ZkVersion.NoVersion))
    retryRequestsUntilConnected(deleteRequests)
  }

  //Acl management methods

  /**
   * Creates the required zk nodes for Acl storage
   */
  def createAclPaths(): Unit = {
    createRecursive(AclZNode.path, throwIfPathExists = false)
    createRecursive(AclChangeNotificationZNode.path, throwIfPathExists = false)
    ResourceType.values.foreach(resource => createRecursive(ResourceTypeZNode.path(resource.name), throwIfPathExists = false))
  }

  /**
   * Gets VersionedAcls for a given Resource
   * @param resource Resource to get VersionedAcls for
   * @return  VersionedAcls
   */
  def getVersionedAclsForResource(resource: Resource): VersionedAcls = {
    val getDataRequest = GetDataRequest(ResourceZNode.path(resource))
    val getDataResponse = retryRequestUntilConnected(getDataRequest)
    getDataResponse.resultCode match {
      case Code.OK => ResourceZNode.decode(getDataResponse.data, getDataResponse.stat)
      case Code.NONODE => VersionedAcls(Set(), -1)
      case _ => throw getDataResponse.resultException.get
    }
  }

  /**
   * Sets or creates the resource znode path with the given acls and expected zk version depending
   * on whether it already exists or not.
   * @param resource
   * @param aclsSet
   * @param expectedVersion
   * @return true if the update was successful and the new version
   */
  def conditionalSetOrCreateAclsForResource(resource: Resource, aclsSet: Set[Acl], expectedVersion: Int): (Boolean, Int) = {
    def set(aclData: Array[Byte],  expectedVersion: Int): SetDataResponse = {
      val setDataRequest = SetDataRequest(ResourceZNode.path(resource), aclData, expectedVersion)
      retryRequestUntilConnected(setDataRequest)
    }

    def create(aclData: Array[Byte]): CreateResponse = {
      val path = ResourceZNode.path(resource)
      val createRequest = CreateRequest(path, aclData, acls(path), CreateMode.PERSISTENT)
      retryRequestUntilConnected(createRequest)
    }

    val aclData = ResourceZNode.encode(aclsSet)

    val setDataResponse = set(aclData, expectedVersion)
    setDataResponse.resultCode match {
      case Code.OK => (true, setDataResponse.stat.getVersion)
      case Code.NONODE => {
        val createResponse = create(aclData)
        createResponse.resultCode match {
          case Code.OK => (true, 0)
          case Code.NODEEXISTS => (false, 0)
          case _ => throw createResponse.resultException.get
        }
      }
      case Code.BADVERSION => (false, 0)
      case _ => throw setDataResponse.resultException.get
    }
  }

  /**
   * Creates Acl change notification message
   * @param resourceName resource name
   */
  def createAclChangeNotification(resourceName: String): Unit = {
    val path = AclChangeNotificationSequenceZNode.createPath
    val createRequest = CreateRequest(path, AclChangeNotificationSequenceZNode.encode(resourceName), acls(path), CreateMode.PERSISTENT_SEQUENTIAL)
    val createResponse = retryRequestUntilConnected(createRequest)
    createResponse.resultException.foreach(e => throw e)
  }

  def propagateLogDirEvent(brokerId: Int) {
    val logDirEventNotificationPath: String = createSequentialPersistentPath(
      LogDirEventNotificationZNode.path + "/" + LogDirEventNotificationSequenceZNode.SequenceNumberPrefix,
      new String(LogDirEventNotificationSequenceZNode.encode(brokerId), UTF_8))
    debug(s"Added $logDirEventNotificationPath for broker $brokerId")
  }

  /**
   * Deletes all Acl change notifications.
   * @throws KeeperException if there is an error while deleting Acl change notifications
   */
  def deleteAclChangeNotifications(): Unit = {
    val getChildrenResponse = retryRequestUntilConnected(GetChildrenRequest(AclChangeNotificationZNode.path))
    if (getChildrenResponse.resultCode == Code.OK) {
      deleteAclChangeNotifications(getChildrenResponse.children)
    } else if (getChildrenResponse.resultCode != Code.NONODE) {
      getChildrenResponse.resultException.foreach(e => throw e)
    }
  }

  /**
   * Deletes the Acl change notifications associated with the given sequence nodes
   * @param sequenceNodes
   */
  private def deleteAclChangeNotifications(sequenceNodes: Seq[String]): Unit = {
    val deleteRequests = sequenceNodes.map { sequenceNode =>
      DeleteRequest(AclChangeNotificationSequenceZNode.deletePath(sequenceNode), ZkVersion.NoVersion)
    }

    val deleteResponses = retryRequestsUntilConnected(deleteRequests)
    deleteResponses.foreach { deleteResponse =>
      if (deleteResponse.resultCode != Code.OK && deleteResponse.resultCode != Code.NONODE) {
        deleteResponse.resultException.foreach(e => throw e)
      }
    }
  }

  /**
   * Gets the resource types
   * @return list of resource type names
   */
  def getResourceTypes(): Seq[String] = {
    getChildren(AclZNode.path)
  }

  /**
   * Gets the resource names for a give resource type
   * @param resourceType
   * @return list of resource names
   */
  def getResourceNames(resourceType: String): Seq[String] = {
    getChildren(ResourceTypeZNode.path(resourceType))
  }

  /**
   * Deletes the given Resource node
   * @param resource
   * @return delete status
   */
  def deleteResource(resource: Resource): Boolean = {
    deleteRecursive(ResourceZNode.path(resource))
  }

  /**
   * checks the resource existence
   * @param resource
   * @return existence status
   */
  def resourceExists(resource: Resource): Boolean = {
    pathExists(ResourceZNode.path(resource))
  }

  /**
   * Conditional delete the resource node
   * @param resource
   * @param expectedVersion
   * @return return true if it succeeds, false otherwise (the current version is not the expected version)
   */
  def conditionalDelete(resource: Resource, expectedVersion: Int): Boolean = {
    val deleteRequest = DeleteRequest(ResourceZNode.path(resource), expectedVersion)
    val deleteResponse = retryRequestUntilConnected(deleteRequest)
    deleteResponse.resultCode match {
      case Code.OK | Code.NONODE => true
      case Code.BADVERSION => false
      case _ => throw deleteResponse.resultException.get
    }
  }

  /**
   * Deletes the zk node recursively
   * @param path
   * @return  return true if it succeeds, false otherwise
   */
  def deletePath(path: String): Boolean = {
    deleteRecursive(path)
  }

  /**
   * This registers a ZNodeChangeHandler and attempts to register a watcher with an ExistsRequest, which allows data
   * watcher registrations on paths which might not even exist.
   *
   * @param zNodeChangeHandler
   * @return `true` if the path exists or `false` if it does not
   * @throws KeeperException if an error is returned by ZooKeeper
   */
  def registerZNodeChangeHandlerAndCheckExistence(zNodeChangeHandler: ZNodeChangeHandler): Boolean = {
    zooKeeperClient.registerZNodeChangeHandler(zNodeChangeHandler)
    val existsResponse = retryRequestUntilConnected(ExistsRequest(zNodeChangeHandler.path))
    existsResponse.resultCode match {
      case Code.OK => true
      case Code.NONODE => false
      case _ => throw existsResponse.resultException.get
    }
  }

  /**
   * See ZooKeeperClient.registerZNodeChangeHandler
   * @param zNodeChangeHandler
   */
  def registerZNodeChangeHandler(zNodeChangeHandler: ZNodeChangeHandler): Unit = {
    zooKeeperClient.registerZNodeChangeHandler(zNodeChangeHandler)
  }

  /**
   * See ZooKeeperClient.unregisterZNodeChangeHandler
   * @param path
   */
  def unregisterZNodeChangeHandler(path: String): Unit = {
    zooKeeperClient.unregisterZNodeChangeHandler(path)
  }

  /**
   * See ZooKeeperClient.registerZNodeChildChangeHandler
   * @param zNodeChildChangeHandler
   */
  def registerZNodeChildChangeHandler(zNodeChildChangeHandler: ZNodeChildChangeHandler): Unit = {
    zooKeeperClient.registerZNodeChildChangeHandler(zNodeChildChangeHandler)
  }

  /**
   * See ZooKeeperClient.unregisterZNodeChildChangeHandler
   * @param path
   */
  def unregisterZNodeChildChangeHandler(path: String): Unit = {
    zooKeeperClient.unregisterZNodeChildChangeHandler(path)
  }

  /**
   *
   * @param stateChangeHandler
   */
  def registerStateChangeHandler(stateChangeHandler: StateChangeHandler): Unit = {
    zooKeeperClient.registerStateChangeHandler(stateChangeHandler)
  }

  /**
   *
   * @param name
   */
  def unregisterStateChangeHandler(name: String): Unit = {
    zooKeeperClient.unregisterStateChangeHandler(name)
  }

  /**
   * Close the underlying ZooKeeperClient.
   */
  def close(): Unit = {
    zooKeeperClient.close()
  }

  /**
   * Get the committed offset for a topic partition and group
   * @param group the group we wish to get offset for
   * @param topicPartition the topic partition we wish to get the offset for
   * @return optional long that is Some if there was an offset committed for topic partition, group and None otherwise.
   */
  def getConsumerOffset(group: String, topicPartition: TopicPartition): Option[Long] = {
    val getDataRequest = GetDataRequest(ConsumerOffset.path(group, topicPartition.topic, topicPartition.partition))
    val getDataResponse = retryRequestUntilConnected(getDataRequest)
    getDataResponse.resultCode match {
      case Code.OK => ConsumerOffset.decode(getDataResponse.data)
      case Code.NONODE => None
      case _ => throw getDataResponse.resultException.get
    }
  }

  /**
   * Set the committed offset for a topic partition and group
   * @param group the group whose offset is being set
   * @param topicPartition the topic partition whose offset is being set
   * @param offset the offset value
   */
  def setOrCreateConsumerOffset(group: String, topicPartition: TopicPartition, offset: Long): Unit = {
    val setDataResponse = setConsumerOffset(group, topicPartition, offset)
    if (setDataResponse.resultCode == Code.NONODE) {
      createConsumerOffset(group, topicPartition, offset)
    } else {
      setDataResponse.resultException.foreach(e => throw e)
    }
  }

  private def setConsumerOffset(group: String, topicPartition: TopicPartition, offset: Long): SetDataResponse = {
    val setDataRequest = SetDataRequest(ConsumerOffset.path(group, topicPartition.topic, topicPartition.partition),
      ConsumerOffset.encode(offset), ZkVersion.NoVersion)
    retryRequestUntilConnected(setDataRequest)
  }

  private def createConsumerOffset(group: String, topicPartition: TopicPartition, offset: Long) = {
    val path = ConsumerOffset.path(group, topicPartition.topic, topicPartition.partition)
    createRecursive(path, ConsumerOffset.encode(offset))
  }

  /**
   * Deletes the given zk path recursively
   * @param path
   * @return true if path gets deleted successfully, false if root path doesn't exist
   * @throws KeeperException if there is an error while deleting the znodes
   */
  private[zk] def deleteRecursive(path: String): Boolean = {
    val getChildrenResponse = retryRequestUntilConnected(GetChildrenRequest(path))
    getChildrenResponse.resultCode match {
      case Code.OK =>
        getChildrenResponse.children.foreach(child => deleteRecursive(s"$path/$child"))
        val deleteResponse = retryRequestUntilConnected(DeleteRequest(path, ZkVersion.NoVersion))
        if (deleteResponse.resultCode != Code.OK && deleteResponse.resultCode != Code.NONODE) {
          throw deleteResponse.resultException.get
        }
        true
      case Code.NONODE => false
      case _ => throw getChildrenResponse.resultException.get
    }
  }

  private[zk] def pathExists(path: String): Boolean = {
    val existsRequest = ExistsRequest(path)
    val existsResponse = retryRequestUntilConnected(existsRequest)
    existsResponse.resultCode match {
      case Code.OK => true
      case Code.NONODE => false
      case _ => throw existsResponse.resultException.get
    }
  }

  private[zk] def createRecursive(path: String, data: Array[Byte] = null, throwIfPathExists: Boolean = true) = {

    def parentPath(path: String): String = {
      val indexOfLastSlash = path.lastIndexOf("/")
      if (indexOfLastSlash == -1) throw new IllegalArgumentException(s"Invalid path ${path}")
      path.substring(0, indexOfLastSlash)
    }

    def createRecursive0(path: String): Unit = {
      val createRequest = CreateRequest(path, null, acls(path), CreateMode.PERSISTENT)
      var createResponse = retryRequestUntilConnected(createRequest)
      if (createResponse.resultCode == Code.NONODE) {
        createRecursive0(parentPath(path))
        createResponse = retryRequestUntilConnected(createRequest)
        if (createResponse.resultCode != Code.OK && createResponse.resultCode != Code.NODEEXISTS) {
          throw createResponse.resultException.get
        }
      } else if (createResponse.resultCode != Code.OK && createResponse.resultCode != Code.NODEEXISTS) {
        throw createResponse.resultException.get
      }
    }

    val createRequest = CreateRequest(path, data, acls(path), CreateMode.PERSISTENT)
    var createResponse = retryRequestUntilConnected(createRequest)

    if (throwIfPathExists && createResponse.resultCode == Code.NODEEXISTS) {
      createResponse.resultException.foreach(e => throw e)
    } else if (createResponse.resultCode == Code.NONODE) {
      createRecursive0(parentPath(path))
      createResponse = retryRequestUntilConnected(createRequest)
      createResponse.resultException.foreach(e => throw e)
    } else if (createResponse.resultCode != Code.NODEEXISTS)
      createResponse.resultException.foreach(e => throw e)

  }

  private def createTopicPartition(partitions: Seq[TopicPartition]): Seq[CreateResponse] = {
    val createRequests = partitions.map { partition =>
      val path = TopicPartitionZNode.path(partition)
      CreateRequest(path, null, acls(path), CreateMode.PERSISTENT, Some(partition))
    }
    retryRequestsUntilConnected(createRequests)
  }

  private def createTopicPartitions(topics: Seq[String]): Seq[CreateResponse] = {
    val createRequests = topics.map { topic =>
      val path = TopicPartitionsZNode.path(topic)
      CreateRequest(path, null, acls(path), CreateMode.PERSISTENT, Some(topic))
    }
    retryRequestsUntilConnected(createRequests)
  }

  private def getTopicConfigs(topics: Seq[String]): Seq[GetDataResponse] = {
    val getDataRequests = topics.map { topic =>
      GetDataRequest(ConfigEntityZNode.path(ConfigType.Topic, topic), ctx = Some(topic))
    }
    retryRequestsUntilConnected(getDataRequests)
  }

  private def acls(path: String): Seq[ACL] = {
    import scala.collection.JavaConverters._
    ZkUtils.defaultAcls(isSecure, path).asScala
  }

  private def retryRequestUntilConnected[Req <: AsyncRequest](request: Req): Req#Response = {
    retryRequestsUntilConnected(Seq(request)).head
  }

  private def retryRequestsUntilConnected[Req <: AsyncRequest](requests: Seq[Req]): Seq[Req#Response] = {
    val remainingRequests = ArrayBuffer(requests: _*)
    val responses = new ArrayBuffer[Req#Response]
    while (remainingRequests.nonEmpty) {
      val batchResponses = zooKeeperClient.handleRequests(remainingRequests)

      // Only execute slow path if we find a response with CONNECTIONLOSS
      if (batchResponses.exists(_.resultCode == Code.CONNECTIONLOSS)) {
        val requestResponsePairs = remainingRequests.zip(batchResponses)

        remainingRequests.clear()
        requestResponsePairs.foreach { case (request, response) =>
          if (response.resultCode == Code.CONNECTIONLOSS)
            remainingRequests += request
          else
            responses += response
        }

        if (remainingRequests.nonEmpty)
          zooKeeperClient.waitUntilConnected()
      } else {
        remainingRequests.clear()
        responses ++= batchResponses
      }
    }
    responses
  }

  def checkedEphemeralCreate(path: String, data: Array[Byte]): Unit = {
    val checkedEphemeral = new CheckedEphemeral(path, data)
    info(s"Creating $path (is it secure? $isSecure)")
    val code = checkedEphemeral.create()
    info(s"Result of znode creation at $path is: $code")
    if (code != Code.OK)
      throw KeeperException.create(code)
  }

  private class CheckedEphemeral(path: String, data: Array[Byte]) extends Logging {
    def create(): Code = {
      val createRequest = CreateRequest(path, data, acls(path), CreateMode.EPHEMERAL)
      val createResponse = retryRequestUntilConnected(createRequest)
      createResponse.resultCode match {
        case code@ Code.OK => code
        case Code.NODEEXISTS => getAfterNodeExists()
        case code =>
          error(s"Error while creating ephemeral at $path with return code: $code")
          code
      }
    }

    private def getAfterNodeExists(): Code = {
      val getDataRequest = GetDataRequest(path)
      val getDataResponse = retryRequestUntilConnected(getDataRequest)
      getDataResponse.resultCode match {
        case Code.OK if getDataResponse.stat.getEphemeralOwner != zooKeeperClient.sessionId =>
          error(s"Error while creating ephemeral at $path, node already exists and owner " +
            s"'${getDataResponse.stat.getEphemeralOwner}' does not match current session '${zooKeeperClient.sessionId}'")
          Code.NODEEXISTS
        case code@ Code.OK => code
        case Code.NONODE =>
          info(s"The ephemeral node at $path went away while reading it, attempting create() again")
          create()
        case code =>
          error(s"Error while creating ephemeral at $path as it already exists and error getting the node data due to $code")
          code
      }
    }
  }
}

object KafkaZkClient {

  /**
   * @param successfulPartitions The successfully updated partition states with adjusted znode versions.
   * @param partitionsToRetry The partitions that we should retry due to a zookeeper BADVERSION conflict. Version conflicts
   *                      can occur if the partition leader updated partition state while the controller attempted to
   *                      update partition state.
   * @param failedPartitions Exceptions corresponding to failed partition state updates.
   */
  case class UpdateLeaderAndIsrResult(successfulPartitions: Map[TopicPartition, LeaderAndIsr],
                                      partitionsToRetry: Seq[TopicPartition],
                                      failedPartitions: Map[TopicPartition, Exception])
}


text:
Computes the perspective for a containefied field of view

## Example

```js
// Create a viewport with fovy PI/2
// and update body on window resize

var Viewport = require('css-viewport'),
    v = new Viewport(Math.PI/2);
window.onresize = updatePerspective;
updatePerspective();

function updatePerspective() {
    v.update(document.body);
};
```

## Installation

`npm install css-viewport`

## Contributors

 - Matt-Esch

## MIT Licenced

redpajama_set_name: RedPajamaGithub


text:
#include <bgfx.h>
#include <bx/allocator.h>
#include <bx/fpumath.h>
#include <bx/timer.h>
#include <ocornut-imgui/imgui.h>
#include "imgui.h"
#include "ocornut_imgui.h"
#include <stb/stb_image.c>

#if defined(SCI_NAMESPACE)
#	include "../entry/input.h"
#	include "scintilla.h"
#endif // defined(SCI_NAMESPACE)

#include "vs_ocornut_imgui.bin.h"
#include "fs_ocornut_imgui.bin.h"

struct OcornutImguiContext
{
	static void* memAlloc(size_t _size);
	static void memFree(void* _ptr);
	static void renderDrawLists(ImDrawData* draw_data);

	void render(ImDrawData* draw_data)
	{
		const float width  = ImGui::GetIO().DisplaySize.x;
		const float height = ImGui::GetIO().DisplaySize.y;

		float ortho[16];
		bx::mtxOrtho(ortho, 0.0f, width, height, 0.0f, -1.0f, 1.0f);

		bgfx::setViewTransform(m_viewId, NULL, ortho);

		// Render command lists
		for (int32_t ii = 0; ii < draw_data->CmdListsCount; ++ii)
		{
			bgfx::TransientVertexBuffer tvb;
			bgfx::TransientIndexBuffer tib;

			const ImDrawList* cmd_list   = draw_data->CmdLists[ii];
			uint32_t vtx_size = (uint32_t)cmd_list->VtxBuffer.size();
			uint32_t idx_size = (uint32_t)cmd_list->IdxBuffer.size();

			if (!bgfx::checkAvailTransientVertexBuffer(vtx_size, m_decl) || !bgfx::checkAvailTransientIndexBuffer(idx_size) )
			{
				// not enough space in transient buffer just quit drawing the rest...
				break;
			}

			bgfx::allocTransientVertexBuffer(&tvb, vtx_size, m_decl);
			bgfx::allocTransientIndexBuffer(&tib, idx_size);

			ImDrawVert* verts = (ImDrawVert*)tvb.data;
			memcpy(verts, cmd_list->VtxBuffer.begin(), vtx_size * sizeof(ImDrawVert) );

			ImDrawIdx* indices = (ImDrawIdx*)tib.data;
			memcpy(indices, cmd_list->IdxBuffer.begin(), idx_size * sizeof(ImDrawIdx) );

			uint32_t elem_offset = 0;
			const ImDrawCmd* pcmd_begin = cmd_list->CmdBuffer.begin();
			const ImDrawCmd* pcmd_end   = cmd_list->CmdBuffer.end();
			for (const ImDrawCmd* pcmd = pcmd_begin; pcmd != pcmd_end; pcmd++)
			{
				if (pcmd->UserCallback)
				{
					pcmd->UserCallback(cmd_list, pcmd);
					elem_offset += pcmd->ElemCount;
					continue;
				}
				if (0 == pcmd->ElemCount)
				{
					continue;
				}

				bgfx::setState(0
					| BGFX_STATE_RGB_WRITE
					| BGFX_STATE_ALPHA_WRITE
					| BGFX_STATE_BLEND_FUNC(BGFX_STATE_BLEND_SRC_ALPHA, BGFX_STATE_BLEND_INV_SRC_ALPHA)
					| BGFX_STATE_MSAA
					);
				bgfx::setScissor(uint16_t(bx::fmax(pcmd->ClipRect.x, 0.0f) )
					, uint16_t(bx::fmax(pcmd->ClipRect.y, 0.0f) )
					, uint16_t(bx::fmin(pcmd->ClipRect.z, 65535.0f)-bx::fmax(pcmd->ClipRect.x, 0.0f) )
					, uint16_t(bx::fmin(pcmd->ClipRect.w, 65535.0f)-bx::fmax(pcmd->ClipRect.y, 0.0f) )
					);
				union { void* ptr; bgfx::TextureHandle handle; } texture = { pcmd->TextureId };

				bgfx::setTexture(0, s_tex, 0 != texture.handle.idx
					? texture.handle
					: m_texture
					);

				bgfx::setVertexBuffer(&tvb, 0, vtx_size);
				bgfx::setIndexBuffer(&tib, elem_offset, pcmd->ElemCount);
				bgfx::submit(m_viewId, m_program);

				elem_offset += pcmd->ElemCount;
			}
		}
	}

	void create(const void* _data, uint32_t _size, float _fontSize, bx::AllocatorI* _allocator)
	{
		m_viewId = 255;
		m_allocator = _allocator;
		m_lastScroll = 0;
		m_last = bx::getHPCounter();

		ImGuiIO& io = ImGui::GetIO();
		io.RenderDrawListsFn = renderDrawLists;
		if (NULL != m_allocator)
		{
			io.MemAllocFn = memAlloc;
			io.MemFreeFn  = memFree;
		}

		io.DisplaySize = ImVec2(1280.0f, 720.0f);
		io.DeltaTime   = 1.0f / 60.0f;
		io.IniFilename = NULL;

#if defined(SCI_NAMESPACE)
		io.KeyMap[ImGuiKey_Tab]        = (int)entry::Key::Tab;
		io.KeyMap[ImGuiKey_LeftArrow]  = (int)entry::Key::Left;
		io.KeyMap[ImGuiKey_RightArrow] = (int)entry::Key::Right;
		io.KeyMap[ImGuiKey_UpArrow]    = (int)entry::Key::Up;
		io.KeyMap[ImGuiKey_DownArrow]  = (int)entry::Key::Down;
		io.KeyMap[ImGuiKey_Home]       = (int)entry::Key::Home;
		io.KeyMap[ImGuiKey_End]        = (int)entry::Key::End;
		io.KeyMap[ImGuiKey_Delete]     = (int)entry::Key::Delete;
		io.KeyMap[ImGuiKey_Backspace]  = (int)entry::Key::Backspace;
		io.KeyMap[ImGuiKey_Enter]      = (int)entry::Key::Return;
		io.KeyMap[ImGuiKey_Escape]     = (int)entry::Key::Esc;
		io.KeyMap[ImGuiKey_A]          = (int)entry::Key::KeyA;
		io.KeyMap[ImGuiKey_C]          = (int)entry::Key::KeyC;
		io.KeyMap[ImGuiKey_V]          = (int)entry::Key::KeyV;
		io.KeyMap[ImGuiKey_X]          = (int)entry::Key::KeyX;
		io.KeyMap[ImGuiKey_Y]          = (int)entry::Key::KeyY;
		io.KeyMap[ImGuiKey_Z]          = (int)entry::Key::KeyZ;
#endif // defined(SCI_NAMESPACE)

		const bgfx::Memory* vsmem;
		const bgfx::Memory* fsmem;

		switch (bgfx::getRendererType() )
		{
		case bgfx::RendererType::Direct3D9:
			vsmem = bgfx::makeRef(vs_ocornut_imgui_dx9, sizeof(vs_ocornut_imgui_dx9) );
			fsmem = bgfx::makeRef(fs_ocornut_imgui_dx9, sizeof(fs_ocornut_imgui_dx9) );
			break;

		case bgfx::RendererType::Direct3D11:
		case bgfx::RendererType::Direct3D12:
			vsmem = bgfx::makeRef(vs_ocornut_imgui_dx11, sizeof(vs_ocornut_imgui_dx11) );
			fsmem = bgfx::makeRef(fs_ocornut_imgui_dx11, sizeof(fs_ocornut_imgui_dx11) );
			break;

		case bgfx::RendererType::Metal:
			vsmem = bgfx::makeRef(vs_ocornut_imgui_mtl, sizeof(vs_ocornut_imgui_mtl) );
			fsmem = bgfx::makeRef(fs_ocornut_imgui_mtl, sizeof(fs_ocornut_imgui_mtl) );
			break;

		default:
			vsmem = bgfx::makeRef(vs_ocornut_imgui_glsl, sizeof(vs_ocornut_imgui_glsl) );
			fsmem = bgfx::makeRef(fs_ocornut_imgui_glsl, sizeof(fs_ocornut_imgui_glsl) );
			break;
		}

		bgfx::ShaderHandle vsh = bgfx::createShader(vsmem);
		bgfx::ShaderHandle fsh = bgfx::createShader(fsmem);
		m_program = bgfx::createProgram(vsh, fsh, true);

		m_decl
			.begin()
			.add(bgfx::Attrib::Position, 2, bgfx::AttribType::Float)
			.add(bgfx::Attrib::TexCoord0, 2, bgfx::AttribType::Float)
			.add(bgfx::Attrib::Color0, 4, bgfx::AttribType::Uint8, true)
			.end();

		s_tex = bgfx::createUniform("s_tex", bgfx::UniformType::Int1);

		uint8_t* data;
		int32_t width;
		int32_t height;
		void* font = ImGui::MemAlloc(_size);
		memcpy(font, _data, _size);
		io.Fonts->AddFontFromMemoryTTF(font, _size, _fontSize);

		io.Fonts->GetTexDataAsRGBA32(&data, &width, &height);

		m_texture = bgfx::createTexture2D( (uint16_t)width
			, (uint16_t)height
			, 1
			, bgfx::TextureFormat::BGRA8
			, 0
			, bgfx::copy(data, width*height*4)
			);

		ImGuiStyle& style = ImGui::GetStyle();
		style.FrameRounding = 4.0f;
	}

	void destroy()
	{
		ImGui::Shutdown();

		bgfx::destroyUniform(s_tex);
		bgfx::destroyTexture(m_texture);
		bgfx::destroyProgram(m_program);

		m_allocator = NULL;
	}

	void beginFrame(int32_t _mx, int32_t _my, uint8_t _button, int32_t _scroll, int _width, int _height, char _inputChar, uint8_t _viewId)
	{
		m_viewId = _viewId;
		ImGuiIO& io = ImGui::GetIO();
		if (_inputChar < 0x7f)
		{
			io.AddInputCharacter(_inputChar); // ASCII or GTFO! :(
		}

		io.DisplaySize = ImVec2( (float)_width, (float)_height);

		const int64_t now = bx::getHPCounter();
		const int64_t frameTime = now - m_last;
		m_last = now;
		const double freq = double(bx::getHPFrequency() );
		io.DeltaTime = float(frameTime/freq);

		io.MousePos = ImVec2( (float)_mx, (float)_my);
		io.MouseDown[0] = 0 != (_button & IMGUI_MBUT_LEFT);
		io.MouseDown[1] = 0 != (_button & IMGUI_MBUT_RIGHT);
		io.MouseWheel = (float)(_scroll - m_lastScroll);
		m_lastScroll = _scroll;

#if defined(SCI_NAMESPACE)
		uint8_t modifiers = inputGetModifiersState();
		io.KeyShift = 0 != (modifiers & (entry::Modifier::LeftShift | entry::Modifier::RightShift) );
		io.KeyCtrl  = 0 != (modifiers & (entry::Modifier::LeftCtrl  | entry::Modifier::RightCtrl ) );
		io.KeyAlt   = 0 != (modifiers & (entry::Modifier::LeftAlt   | entry::Modifier::RightAlt  ) );
		for (int32_t ii = 0; ii < (int32_t)entry::Key::Count; ++ii)
		{
			io.KeysDown[ii] = inputGetKeyState(entry::Key::Enum(ii) );
		}
#endif // defined(SCI_NAMESPACE)

		ImGui::NewFrame();

		//ImGui::ShowTestWindow(); //Debug only.
		//
#if defined(SCI_NAMESPACE) && 0
		bool opened = true;
		ImGuiScintilla("Scintilla Editor", &opened, ImVec2(640.0f, 480.0f) );
#endif // 0
	}

	void endFrame()
	{
		ImGui::Render();
	}

	bx::AllocatorI*     m_allocator;
	bgfx::VertexDecl    m_decl;
	bgfx::ProgramHandle m_program;
	bgfx::TextureHandle m_texture;
	bgfx::UniformHandle s_tex;
	int64_t m_last;
	int32_t m_lastScroll;
	uint8_t m_viewId;
};

static OcornutImguiContext s_ctx;

void* OcornutImguiContext::memAlloc(size_t _size)
{
	return BX_ALLOC(s_ctx.m_allocator, _size);
}

void OcornutImguiContext::memFree(void* _ptr)
{
	BX_FREE(s_ctx.m_allocator, _ptr);
}

void OcornutImguiContext::renderDrawLists(ImDrawData* draw_data)
{
	s_ctx.render(draw_data);
}

void IMGUI_create(const void* _data, uint32_t _size, float _fontSize, bx::AllocatorI* _allocator)
{
	s_ctx.create(_data, _size, _fontSize, _allocator);
}

void IMGUI_destroy()
{
	s_ctx.destroy();
}

void IMGUI_beginFrame(int32_t _mx, int32_t _my, uint8_t _button, int32_t _scroll, int _width, int _height, char _inputChar, uint8_t _viewId)
{
	s_ctx.beginFrame(_mx, _my, _button, _scroll, _width, _height, _inputChar, _viewId);
}

void IMGUI_endFrame()
{
	s_ctx.endFrame();
}


text:
function piepan.onConnect()
    print ("echo loaded!")
end

function piepan.onMessage(message)
    if message.user == nil then
        return
    end
    piepan.me.channel:send(message.text)
end


text:
package handlers

import (
	"net/http"

	"github.com/distribution/distribution/v3"
	"github.com/distribution/distribution/v3/context"
	"github.com/distribution/distribution/v3/registry/api/errcode"
	v2 "github.com/distribution/distribution/v3/registry/api/v2"
	"github.com/gorilla/handlers"
	"github.com/opencontainers/go-digest"
)

// blobDispatcher uses the request context to build a blobHandler.
func blobDispatcher(ctx *Context, r *http.Request) http.Handler {
	dgst, err := getDigest(ctx)
	if err != nil {

		if err == errDigestNotAvailable {
			return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
				ctx.Errors = append(ctx.Errors, v2.ErrorCodeDigestInvalid.WithDetail(err))
			})
		}

		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			ctx.Errors = append(ctx.Errors, v2.ErrorCodeDigestInvalid.WithDetail(err))
		})
	}

	blobHandler := &blobHandler{
		Context: ctx,
		Digest:  dgst,
	}

	mhandler := handlers.MethodHandler{
		"GET":  http.HandlerFunc(blobHandler.GetBlob),
		"HEAD": http.HandlerFunc(blobHandler.GetBlob),
	}

	if !ctx.readOnly {
		mhandler["DELETE"] = http.HandlerFunc(blobHandler.DeleteBlob)
	}

	return mhandler
}

// blobHandler serves http blob requests.
type blobHandler struct {
	*Context

	Digest digest.Digest
}

// GetBlob fetches the binary data from backend storage returns it in the
// response.
func (bh *blobHandler) GetBlob(w http.ResponseWriter, r *http.Request) {
	context.GetLogger(bh).Debug("GetBlob")
	blobs := bh.Repository.Blobs(bh)
	desc, err := blobs.Stat(bh, bh.Digest)
	if err != nil {
		if err == distribution.ErrBlobUnknown {
			bh.Errors = append(bh.Errors, v2.ErrorCodeBlobUnknown.WithDetail(bh.Digest))
		} else {
			bh.Errors = append(bh.Errors, errcode.ErrorCodeUnknown.WithDetail(err))
		}
		return
	}

	if err := blobs.ServeBlob(bh, w, r, desc.Digest); err != nil {
		context.GetLogger(bh).Debugf("unexpected error getting blob HTTP handler: %v", err)
		bh.Errors = append(bh.Errors, errcode.ErrorCodeUnknown.WithDetail(err))
		return
	}
}

// DeleteBlob deletes a layer blob
func (bh *blobHandler) DeleteBlob(w http.ResponseWriter, r *http.Request) {
	context.GetLogger(bh).Debug("DeleteBlob")

	blobs := bh.Repository.Blobs(bh)
	err := blobs.Delete(bh, bh.Digest)
	if err != nil {
		switch err {
		case distribution.ErrUnsupported:
			bh.Errors = append(bh.Errors, errcode.ErrorCodeUnsupported)
			return
		case distribution.ErrBlobUnknown:
			bh.Errors = append(bh.Errors, v2.ErrorCodeBlobUnknown)
			return
		default:
			bh.Errors = append(bh.Errors, err)
			context.GetLogger(bh).Errorf("Unknown error deleting blob: %s", err.Error())
			return
		}
	}

	w.Header().Set("Content-Length", "0")
	w.WriteHeader(http.StatusAccepted)
}


text:
##Code

----

<ClassDocStr>

----

```python
# -*- coding: utf-8 -*-
"""

<DefineSource>
@Date : Fri Nov 14 13:20:38 2014 \n
@Author : Erwan Ledoux \n\n
</DefineSource>

The Packager is an Object that helps to get a module in the SYS framework

"""

#<DefineAugmentation>
import ShareYourSystem as SYS
BaseModuleStr="ShareYourSystem.Standards.Interfacers.Folderer"
DecorationModuleStr="ShareYourSystem.Standards.Classors.Tester"
SYS.setSubModule(globals())
#</DefineAugmentation>

#<ImportSpecificModules>
import sys
import importlib
#</ImportSpecificModules>

#<DefineClass>
@DecorationClass()
class PackagerClass(BaseClass):

	#Definition
	RepresentingKeyStrsList=[
								'PackagingModuleVariable',
								'PackagedModuleStr',
								'PackagedModuleVariable',
								'PackagedInstallFolderPathStr',
								'PackagedLocalFolderPathStr'
							]

	def default_init(self,
						_PackagingModuleVariable=None,
						_PackagedModuleStr="",
						_PackagedModuleVariable=None,
						_PackagedInstallFolderPathStr="",
						_PackagedLocalFolderPathStr="",
						**_KwargVariablesDict
					):

		#Call the parent init method
		BaseClass.__init__(self,**_KwargVariablesDict)

	def do_package(self):

		#debug
		'''
		self.debug(('self.',self,
						[
							'PackagingModuleVariable'
						]))
		'''

		#Check
		if type(self.PackagingModuleVariable) in SYS.StrTypesList:
			self.PackagedModuleStr=self.PackagingModuleVariable
		else:
			self.PackagedModuleVariable=self.PackagingModuleVariable
			self.PackagedModuleStr=self.PackagingModuleVariable.__name__

		#Check for a module
		if self.PackagedModuleVariable==None or self.PackagedModuleStr!=self.PackagedModuleVariable.__name__:

			#Check
			if self.PackagedModuleStr!="":

				#Import the module if not already
				if self.PackagedModuleStr not in sys.modules:
					importlib.import_module(self.PackagedModuleStr)

				#set with sys
				self.PackagedModuleVariable=sys.modules[self.PackagedModuleStr]

		#set
		if self.PackagedModuleVariable!=None:

			#set
			self.PackagedInstallFolderPathStr='/'.join(
				self.PackagedModuleVariable.__file__.split('/')[:-1]
			)+'/'

			#set
			self.PackagedLocalFolderPathStr=SYS.PythonlogyLocalFolderPathStr+self.PackagedModuleVariable.__name__.replace(
				'.','/')+'/'

			#debug
			'''
			self.debug(('self.',self,[
				'PackagedInstallFolderPathStr',
				'PackagedLocalFolderPathStr'
				]))
			'''

			#Hook
			self.folder(self.PackagedLocalFolderPathStr)

		#Return
		#return self

#</DefineClass>

```

View the Packager sources on Github


text:
A complete environment to create Markdown files with a syntax highlighting that doesn't suck!
This is a fork of @gabrielelana's [vim-markdown](https://github.com/gabrielelana/vim-markdown)
that adds a few features that are useful for notetaking. It is intended to be used in conjunction
with [morgue](https://github.com/ibabushkin/morgue) and is like a drop-in replacement for the
original, i.e, no functionality is lost, just some additions were made.

## Features
* Insert morgue-style timestamps via simple commands
  * `:InsertTimestamp 3mon T` inserts an appointment ( **T**ime mode) on the 3rd monday from now on.
  * `:InsertTimestamp thu D` sets a **D**eadline on next thursday.
  * `:InsertTimestamp 2fri S` sets a **S**chedule on the second friday.
  * `:InsertTimestampExact 2fri 15:00 D` takes an additional time.
* automatic syntax highlighting for valid timestamps (see morgue's readme for the complete format specs).

# Original README

## Features
* Strong support for the Markdown flavor implemented by Github: what you see in Vim is what you get on Github
* Complete syntax implementation: supports proper nesting of all elements in list items, this is the only plugin that is able to do that (and I believe it since it took me a *while* to make it right)
  * A simple example rendered with this plugin. Headers and the fenced code block in list items are correctly highlighted
    ![VimMarkdownRendering](https://github.com/gabrielelana/vim-markdown/raw/master/images/vim_markdown_rendering.png)
  * The same example rendered with the most popular Markdown plugins for Vim
    ![VimMarkdownRenderingComparison1](https://github.com/gabrielelana/vim-markdown/raw/master/images/vim_markdown_rendering_comparison_1.png)
    ![VimMarkdownRenderingComparison2](https://github.com/gabrielelana/vim-markdown/raw/master/images/vim_markdown_rendering_comparison_2.png)
* Code blocks and pieces of Markdown of the current file could be edited in a separate buffer and synchronized back when you finished
  * Inside a ruby fenced code block with `<Leader>e` opens a temporary buffer with the right file type
  * Then select a range in visual mode and with `<Leader>e` opens a temporary buffer with file type `markdown` I call it *Focus Mode* because you can edit a portion of a Markdown file in isolation
  * At last, on an empty line, with `<Leader>e` asks for a file type and then opens a temporary buffer with that file type
  * You see that when you leave the temporary buffer the content syncs back to the main file
  ![EditCodeBlock](https://github.com/gabrielelana/vim-markdown/raw/master/images/vim_markdown_edit_code_block.gif)
* Folding for: headers, code blocks and html blocks
* Format tables automatically (require [`Tabular`](https://github.com/godlygeek/tabular) plugin)
* Automatically detects Jekyll files and adds support for Liquid template engine
* This is a work in progress, more goodies and improvements are coming (see [TODO](#TODO)), stay tuned

## Why?
I wanted a strong support for the Markdown flavor implemented by Github, I wanted a syntax highlight that would mirror the result I would find later on Github, I wanted a syntax highlight that would not break easily, I wanted a syntax highlight that I could rely on (aka rapid feedback), I wanted something more that a mere syntax highlight. The [Markdown Syntax](http://daringfireball.net/projects/markdown/syntax) unfortunately it's so loosely defined that there are *flavors* of Markdown that are subtly incompatible from each other, the [Markdown supported by Github](https://help.github.com/articles/github-flavored-markdown) is one of them.

## Installation
If you use [Vundle](https://github.com/gmarik/vundle), add the following line to your `~/.vimrc`:

    Bundle 'gabrielelana/vim-markdown'

And then execute the following command in your shell:

    $ vim +'PluginInstall! vim-markdown' +qall

You can update to the latest version with the following command in your shell:

    $ vim +PluginUpdate

If you use [NeoBundle](https://github.com/Shougo/neobundle.vim), add the following line to your `~/.vimrc`:

    NeoBundle 'gabrielelana/vim-markdown'

And then execute the following command in your shell:

    $ vim +NeoBundleInstall +qall

You can update to the latest version with the following command in your shell:

    $ vim +NeoBundleInstall! +qall

If you use [Pathogen](https://github.com/tpope/vim-pathogen), execute the following in your shell:

    $ cd ~/.vim/bundle
    $ git clone https://github.com/gabrielelana/vim-markdown.git

If you are not using any package manager, download the [tarball](https://github.com/gabrielelana/vim-markdown/archive/master.tar.gz) and do this:

    $ cp vim-markdown-master.tar.gz ~/.vim
    $ cd ~/.vim
    $ tar --strip-components=1 --overwrite -zxf vim-markdown-master.tar.gz
    $ rm vim-markdown-master.tar.gz

## Self-Promotion
If you like this plugin, then consider to:
* Star the repository on [GitHub](https://github.com/gabrielelana/vim-markdown)
* Follow me on
  * [Twitter](http://twitter.com/gabrielelana)
  * [GitHub](https://github.com/gabrielelana)

## Documentation
I would use this section until I have a proper documentation

### Configuration
* `let g:markdown_include_jekyll_support = 0` to disable support for Jekyll files (enabled by default with: `1`)
* `let g:markdown_enable_folding = 1` to enable the fold expression `markdown#FoldLevelOfLine` to fold markdown files, this is disabled by default because it's a huge performance hit even when folding is disabled with `nofoldenable` option (disabled by default with: `0`)
* `let g:markdown_enable_mappings = 0` to disable default mappings (enabled by default with: `1`)
  * `let g:markdown_enable_insert_mode_mappings = 0` to disable insert mode mappings (enabled by default with: `1`)
  * `let g:markdown_enable_insert_mode_leader_mappings = 1` to enable insert mode leader mappings (disabled by default with: `0`)
* `let g:markdown_enable_spell_checking = 0` to disable spell checking (enabled by default with: `1`)
* `let g:markdown_enable_input_abbreviations = 0` to disable abbreviations for punctuations and emoticons (enabled by default with: `1`)
* `let g:markdown_enable_conceal = 1` to enable conceal for italic, bold, inline-code and link (disabled by default with: `0`)

### Default Mappings (normal and visual mode)
_mappings are local to markdown buffers_
* `<Space>` (`NORMAL_MODE`) switch status of things:
  * Cases
    * A list item `* item` becomes a check list item `* [ ] item`
    * A check list item `* [ ] item` becomes a checked list item `* [x] item`
    * A checked list item `* [x] item` becomes a list item `* item`
  * Can be changed with `g:markdown_mapping_switch_status = '<Leader>s`
* `<Leader>ft` (`NORMAL_MODE`) format the current table
* `<Leader>e` (`NORMAL_MODE`, `VISUAL_MODE`) `:MarkdownEditCodeBlock` edit the current code block in another buffer with a guessed file type. The guess is based on the start of the range for `VISUAL_MODE`. If it's not possible to guess (you are not in a recognizable code block like a fenced code block) then the default is `markdown`. If it's not possible to guess and the current range is a single line and the line is empty then a new code block is created. It's asked to the user the file type of the new code block. The default file type is `markdown`.

### Optional Mappings (insert mode)
_mappings are local to markdown buffers_
* `<Leader>ft` (`INSERT_MODE`) same as `NORMAL_MODE` `<Leader>ft` with an additional mapping for `INSERT_MODE`
* `<Leader>e` (`INSERT_MODE`) same as `NORMAL_MODE` and `VISUAL_MODE` `<leader>e` with an additional mapping for `INSERT_MODE`

### Motions
* `]]` start of the next header
* `[[` start of the previous header

### While Editing in Insert Mode
* `|` in a table triggers the format command
* `<Tab>`/`<S-Tab>` on a list item it will indent/unindent the item
* `<Tab>`/`<S-Tab>` on a blockquote it will increase/decrease the quote level
* `<Enter>` on a list item with no text in it (freshly created) it will delete everything till the column 0

## Development
### Resources
* [Markdown Github Syntax](https://help.github.com/articles/github-flavored-markdown)
* [Markdown Github API](http://developer.github.com/v3/markdown)
* [Markdown Github Quick Preview](http://github-markdown-preview.heroku.com/)

### Syntax Specs
Testing syntax highlight could be tricky, here I use the golden master patter to at least avoid regressions, this is how it works: in `./rspec/features` you will find a bunch of `*.md` files, one for each syntactic element supported, for each of those files there's an html file, this file is created with the `:TOhtml` command and it's the reference (aka golden master) of the syntax highlight of the original file. Running `rspec` you are comparing the current syntax highlight of all the feature's file with the reference syntax highlight. If looking at some of the feature's file you see something wrong you can fix it and after regenerate the golden master files with `GENERATE_GOLDEN_MASTER=1 rspec`

## Known Bugs
* `formatlistpat` doesn't work for ordered lists
* `formatoptions` thinks that `*` in horizontal rules are list items

<a name="TODO" />
## TODO
* Kramdown Syntax
  * Block Inline Attributes
  * Span Inline Attributes
  * Fenced Code Blocks with Inline Attributes
  * Definition Lists
  * Tables
  * Math Blocks
  * Footnotes
  * Abbreviations
  * End of Block Marker
* Code Blocks
  * Edit jekyll front matter code block
  * Explain in this file why I chose to avoid to highlight nested block elements
* Emoji (GFM)
  * Start completion when hitting `:` in insert mode only if preceded by empty spaces or is the beginning of the line
  * Remove duplication between syntax keywords and dictionary completion
  * More `iabbr`
* Lists
  * `i_<BS>` on a list item with no text in it (freshly created) it will delete everything till the column 0
  * `<C-K>` on a list item it will swap it with the item above (if it exists)
  * `<C-J>` on a list item it will swap it with the item below (if it exists)
  * `>` and `<` should properly indent/unindent list items
  * `i_<C-D>` and `i_<C-T>` should properly indent/unindent list items and quote lines
* Define custom text objects for:
  * List items
  * Check list items
  * Paragraph
    * Start of the next paragraph `}`
    * Start of the previous paragraph `{`
  * Other inline elements
* Folding
  * Fold HTML blocks
  * Always consider the first non space character of the line when check for syntax group
  * Always use `synstack` to check the syntax group of a position
  * Write specs for the whole thing
  * Support `foldtext` option


text:
package org.judal.storage.java;

import java.io.Serializable;

/*
 * © Copyright 2016 the original author.
 * This file is licensed under the Apache License version 2.0.
 * 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.
 */

import java.sql.Types;
import java.util.Map;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Set;
import java.util.TreeSet;

import javax.jdo.JDOException;
import javax.jdo.JDOUserException;

import org.judal.metadata.TableDef;
import org.judal.metadata.ViewDef;
import org.judal.metadata.ColumnDef;
import org.judal.serialization.BytesConverter;
import org.judal.storage.ConstraintsChecker;
import org.judal.storage.FieldHelper;
import org.judal.storage.table.TableDataSource;
import org.judal.storage.table.impl.AbstractRecord;
import org.judal.storage.java.internal.CaseInsensitiveValuesMap;

import com.knowgate.debug.DebugFile;

/**
 * Implementation of Record interface using an hash map to hold column values
 * @author Sergio Montoro Ten
 * @version 1.0
 */
public class MapRecord extends AbstractRecord implements JavaRecord {

	private static final long serialVersionUID = 10000l;

	private CaseInsensitiveValuesMap values;
	private HashMap<Integer, String> columnIndexes;

	/**
	 * Constructor
	 */
	public MapRecord(ViewDef tableDefinition) {
		this(tableDefinition,null,null);
	}

	/**
	 * Constructor
	 */
	public MapRecord(ViewDef tableDefinition, FieldHelper fieldHelper) {
		this(tableDefinition, fieldHelper, null);
	}

	/**
	 * Constructor
	 */
	public MapRecord(ViewDef tableDefinition, ConstraintsChecker constraintsChecker) {
		this(tableDefinition, null, constraintsChecker);
	}

	/**
	 * Constructor
	 */
	public MapRecord(ViewDef tableDefinition, FieldHelper fieldHelper, ConstraintsChecker constraintsChecker) {
		super(tableDefinition, fieldHelper, constraintsChecker);
		values = new CaseInsensitiveValuesMap();
		columnIndexes = new HashMap<Integer, String>();
		int i = 0;
		for (ColumnDef cdef: tableDefinition.getColumns())
			columnIndexes.put(++i, cdef.getName());
	}

	/**
	 * Constructor
	 * @param dataSource TableDataSource
	 * @param tableName String
	 * @throws JDOException
	 * @throws JDOUserException if table or view is not found at DataSource metadata
	 */
	public MapRecord(TableDataSource dataSource, String tableName) throws JDOException {
		this(getViewDefForName(dataSource, tableName), dataSource.getFieldHelper());
	}

	/**
	 * Constructor
	 * @param dataSource TableDataSource
	 * @param tableName String
	 * @param fieldHelper FieldHelper
	 * @throws JDOException
	 * @throws JDOUserException if table or view is not found at DataSource metadata
	 */
	public MapRecord(TableDataSource dataSource, String tableName, FieldHelper fieldHelper) throws JDOException {
		this(getViewDefForName(dataSource, tableName), fieldHelper);
	}

	/**
	 * Constructor
	 * @param dataSource TableDataSource
	 * @param tableName String
	 * @param constraintsChecker ConstraintsChecker
	 * @throws JDOException
	 * @throws JDOUserException if table or view is not found at DataSource metadata
	 */
	public MapRecord(TableDataSource dataSource, String tableName, ConstraintsChecker constraintsChecker) throws JDOException {
		this(dataSource, tableName, dataSource.getFieldHelper(), constraintsChecker);
	}

	/**
	 * Constructor
	 * @param dataSource TableDataSource
	 * @param tableName String
	 * @param fieldHelper FieldHelper
	 * @param constraintsChecker ConstraintsChecker
	 * @throws JDOException
	 * @throws JDOUserException if table or view is not found at DataSource metadata
	 */
	public MapRecord(TableDataSource dataSource, String tableName, FieldHelper fieldHelper, ConstraintsChecker constraintsChecker) throws JDOException {
		this(getViewDefForName(dataSource, tableName), fieldHelper, constraintsChecker);
	}

	/**
	 * Constructor
	 * @throws JDOException
	 */
	public MapRecord(TableDataSource dataSource, String tableName, String... columnNames) throws JDOException {
		this(dataSource, tableName, dataSource.getFieldHelper(), null, columnNames);
	}

	/**
	 * Constructor
	 * @throws JDOException
	 */
	public MapRecord(TableDataSource dataSource, String tableName, FieldHelper fieldHelper, String... columnNames) throws JDOException {
		this(dataSource, tableName, fieldHelper, null, columnNames);
	}

	/**
	 * Constructor
	 * @throws JDOException
	 */
	public MapRecord(TableDataSource dataSource, String tableName, ConstraintsChecker constraintsChecker, String... columnNames) throws JDOException {
		this(dataSource, tableName, null, constraintsChecker, columnNames);
	}

	/**
	 * Constructor
	 * @throws JDOException
	 */
	public MapRecord(TableDataSource dataSource, String tableName, FieldHelper fieldHelper, ConstraintsChecker constraintsChecker, String... columnNames) throws JDOException {
		super(new TableDef(tableName, filterColumns(dataSource, tableName, columnAliases(columnNames))), fieldHelper, constraintsChecker);
		values = new CaseInsensitiveValuesMap();
		columnIndexes = new HashMap<Integer, String>();
		int c = 0;
		for (String columnName : columnNames)
			columnIndexes.put(new Integer(++c), getColumnAlias(columnName));
	}

	private static ColumnDef[] filterColumns(TableDataSource dataSource, String tableName, String[] columnNames) {
		ColumnDef[] retval;
		ViewDef vdef = dataSource.getTableOrViewDef(tableName);
		if (null==vdef) {
			retval = new ColumnDef[columnNames.length];
			for (int c=columnNames.length-1; c>=0; c--)
				retval[c] = dataSource.createColumnDef(columnNames[c], c+1, (short) Types.NULL, null);
		} else {
			retval = vdef.filterColumns(columnAliases(columnNames));
		}
		return retval;
	}

	private static String[] columnAliases(String[] columnNames) {
		final int colCount = columnNames.length;
		String[] aliases = new String[colCount];
		for (int c=0; c<colCount; c++)
			aliases[c] = getColumnAlias(columnNames[c]);
		return aliases;
	}

	@Override
	@SuppressWarnings("unchecked")
	public Entry<String,Object>[] asEntries() {
		return values.entrySet().toArray(new Entry[values.size()]);
	}

	@Override
	public Map<String,Object> asMap() {
		return values;
	}

	@Override
	public Set<String> keySet() {
		TreeSet<String> keys = new TreeSet<String>();
		for (ColumnDef c : getTableDef().getColumns())
			keys.add(c.getName());
		return keys;
	}

	@Override
	public Object put(int colpos, Object value) throws ArrayIndexOutOfBoundsException {
		final String columnName = columnIndexes.get(colpos);
		if (null==columnName)
				throw new ArrayIndexOutOfBoundsException("Cannot find column at position " + String.valueOf(colpos)+" of "+String.valueOf(columnIndexes.size())+" columns");
		return values.put(columnName, value);
	}

	/**
	 * Set or replace the value of a column
	 * @param colname String
	 * @param bytes byte[]
	 * @return byte[] Previous value stored at the column
	 */
	@Override
	public Object put(String colname, byte[] bytes) {
		byte[] retval;
		Object former = values.get(colname);
		if (former==null)
			retval = null;
		else if (former instanceof byte[])
			retval = Arrays.copyOf((byte[]) former, ((byte[]) former).length);
		else
			retval = BytesConverter.toBytes(former, Types.JAVA_OBJECT);
		values.put(colname, bytes);
		return retval;
	}

	@Override
	public Object remove(String colname) {
		Object retval = containsKey(colname) ? get(colname) : null;
		values.remove(colname);
		return retval;
	}

	@Override
	public Object remove(Object colname) {
		Object retval = containsKey(colname) ? get(colname) : null;
		values.remove(colname);
		return retval;
	}

	@Override
	public void clear() {
		values.clear();
	}

	@Override
	public boolean containsKey(Object key) {
		return values.containsKey(key);
	}

	@Override
	public boolean containsValue(Object value) {
		return values.containsValue(value);
	}

	@Override
	public Set<java.util.Map.Entry<String, Object>> entrySet() {
		HashMap<String,Object> entries = new HashMap<String,Object>();
		for (Integer columnIndex : columnIndexes.keySet()) {
			String columnName = columnIndexes.get(columnIndex);
			entries.put(columnName, values.get(columnName));
		}
		return Collections.unmodifiableSet(entries.entrySet());
	}

	@Override
	public Object apply(String key) {
		return values.get(key);
	}

	@Override
	public Object get(Object key) {
		return values.get(key);
	}

	@SuppressWarnings("unchecked")
	@Override
	public Map<String,String> getMap(String sKey) throws ClassCastException, InstantiationException, IllegalAccessException, IllegalArgumentException, NoSuchMethodException, SecurityException, ClassNotFoundException {
		return (Map<String,String>) super.getMap(sKey);
	} // getMap

	@Override
	public boolean isEmpty() {
		return values.isEmpty();
	}

	@Override
	public Object put(String key, Object value) {
		return values.put(key, value);
	}

	@Override
	public int size() {
		return values.size();
	}

	@Override
	public Collection<Object> values() {
		return values.values();
	}

	@SuppressWarnings("rawtypes")
	@Override
	public void setValue(Serializable value) throws JDOException {
		if (value instanceof Map) {
			values = new CaseInsensitiveValuesMap();
			columnIndexes = new HashMap<Integer, String>();
			Map valmap = (Map) value;
			Iterator iter = valmap.keySet().iterator();
			int n = 0;
			while (iter.hasNext()) {
				Object key = iter.next();
				values.put(key.toString(), valmap.get(key));
				columnIndexes.put(++n, key.toString());
			}
		} else {
			throw new ClassCastException("MapRecord.setValue() Cannot cast from "+value.getClass().getName()+" to java.util.Map");
		}
	}

	@Override
	public Object getValue() throws JDOException {
		return Collections.unmodifiableMap(values);
	}

	@Override
	public void putAll(Map<? extends String,? extends Object> mValues) {
		for (Entry<? extends String,? extends Object> entry : mValues.entrySet()) {
			put(entry.getKey(), entry.getValue());
		}
	}

	private static ViewDef getViewDefForName(TableDataSource dataSource, String name) {
		final ViewDef vdef = dataSource.getTableOrViewDef(name);
		if (null==vdef) {
			if (DebugFile.trace) {
				DebugFile.writeln("JDOException at MapRecord.getViewDefForName(). Table or View not found " + name);
				DebugFile.writeln("SchemaMetadata contains " + dataSource.getMetaData().tables().size()+" tables");
				DebugFile.writeln("SchemaMetadata contains " + dataSource.getMetaData().views().size()+" views");
				StringBuilder tableList = new StringBuilder();
				tableList.append("JDC tables list = [");
				try {
					boolean first = true;
					tableList.append("SchemaMetaData tables = [");
					first = true;
					for (TableDef t : dataSource.getMetaData().tables()) {
						if (first)
							first = false;
						else
							tableList.append(",");
						tableList.append(t.getName());
					}
					tableList.append("]");
					DebugFile.writeln(tableList.toString());
					tableList.setLength(0);
					tableList.append("SchemaMetaData views = [");
					first = true;
					for (ViewDef t : dataSource.getMetaData().views()) {
						if (first)
							first = false;
						else
							tableList.append(",");
						tableList.append(t.getName());
					}
					tableList.append("]");
					DebugFile.writeln(tableList.toString());
				} catch (Exception e) {
					if (DebugFile.trace) {
						DebugFile.writeln(e.getClass().getName() + " " + e.getMessage());
					}
				}
			}
			throw new JDOUserException("MapRecord constructor. Table or view " + name + " not found");
		}
		return vdef;
	}
}


text:
package com.google.inject;

import static com.google.inject.Asserts.assertContains;

import java.io.IOException;
import java.io.Serializable;
import java.util.List;
import junit.framework.AssertionFailedError;
import junit.framework.TestCase;

/** @author jessewilson@google.com (Jesse Wilson) */
public class SerializationTest extends TestCase {

  public void testAbstractModuleIsSerializable() throws IOException {
    Asserts.reserialize(new MyAbstractModule());
  }

  static class MyAbstractModule extends AbstractModule implements Serializable {
  }

  public void testCreationExceptionIsSerializable() throws IOException {
    CreationException exception = createCreationException();
    CreationException reserialized = Asserts.reserialize(exception);
    assertContains(
        reserialized.getMessage(),
        "1) [Guice/MissingImplementation]: No implementation for List was bound.",
        "at SerializationTest$1.configure");
  }

  private CreationException createCreationException() {
    try {
      Guice.createInjector(
          new AbstractModule() {
            @Override
            protected void configure() {
              bind(List.class);
            }
          });
      throw new AssertionFailedError();
    } catch (CreationException e) {
      return e;
    }
  }

  static class A {
    @Inject B b;
  }

  static class B {}
}


text:
package kafka.server

import java.io.File

import org.apache.kafka.common.protocol.SecurityProtocol
import org.junit.{After, Before, Test}
import kafka.zk.ZooKeeperTestHarness
import kafka.utils.TestUtils
import TestUtils._
import org.apache.kafka.clients.producer.ProducerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.serialization.StringSerializer

class ReplicaFetchTest extends ZooKeeperTestHarness  {
  var brokers: Seq[KafkaServer] = null
  val topic1 = "foo"
  val topic2 = "bar"

  @Before
  override def setUp() {
    super.setUp()
    val props = createBrokerConfigs(2, zkConnect)
    brokers = props.map(KafkaConfig.fromProps).map(TestUtils.createServer(_))
  }

  @After
  override def tearDown() {
    TestUtils.shutdownServers(brokers)
    super.tearDown()
  }

  @Test
  def testReplicaFetcherThread() {
    val partition = 0
    val testMessageList1 = List("test1", "test2", "test3", "test4")
    val testMessageList2 = List("test5", "test6", "test7", "test8")

    // create a topic and partition and await leadership
    for (topic <- List(topic1,topic2)) {
      createTopic(zkUtils, topic, numPartitions = 1, replicationFactor = 2, servers = brokers)
    }

    // send test messages to leader
    val producer = TestUtils.createNewProducer(TestUtils.getBrokerListStrFromServers(brokers),
                                               retries = 5,
                                               keySerializer = new StringSerializer,
                                               valueSerializer = new StringSerializer)
    val records = testMessageList1.map(m => new ProducerRecord(topic1, m, m)) ++
      testMessageList2.map(m => new ProducerRecord(topic2, m, m))
    records.map(producer.send).foreach(_.get)
    producer.close()

    def logsMatch(): Boolean = {
      var result = true
      for (topic <- List(topic1, topic2)) {
        val tp = new TopicPartition(topic, partition)
        val expectedOffset = brokers.head.getLogManager().getLog(tp).get.logEndOffset
        result = result && expectedOffset > 0 && brokers.forall { item =>
          expectedOffset == item.getLogManager().getLog(tp).get.logEndOffset
        }
      }
      result
    }
    waitUntilTrue(logsMatch _, "Broker logs should be identical")
  }
}


text:
Fuck you!

Kick-start your fucking start up at 1871

Welcome to the Chatroom {{ getCurrentUser().name }}

 {{getLanguage}}

Angular Fullstack v2.0.13 |
        @tyhenkel |
         Issues


text:
package net.brabenetz.lib.securedproperties.core;

import net.brabenetz.lib.securedproperties.test.TestUtils;
import org.junit.Assert;
import org.junit.Test;

import static org.hamcrest.CoreMatchers.containsString;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.is;

public class EncryptionTest {

    @Test
    public void testIsAlgorithmSupported() throws Exception {
        Assert.assertTrue(Encryption.isAlgorithmSupported(SupportedAlgorithm.AES_128));
        Assert.assertTrue(Encryption.isAlgorithmSupported(SupportedAlgorithm.DESede_112));
        Assert.assertTrue(Encryption.isAlgorithmSupported(SupportedAlgorithm.DESede_168));
        Assert.assertFalse(Encryption.isAlgorithmSupported(new MockAlgorithm("test", 1)));
        Assert.assertFalse(Encryption.isAlgorithmSupported(new MockAlgorithm("AES", 1000)));
        // By default, Java only supports AES 128 (travis-ci supports AES_256), but you can import it into your jvm
        // Assert.assertFalse(Encryption.isAlgorithmSupported(SupportedAlgorithm.AES_256));
    }

    @Test
    public void testUtilityPattern() {
        // this test is more for test-coverage than logic :D
        Assert.assertTrue(TestUtils.isDefaultConstructorHidden(Encryption.class));
    }

    @Test
    public void testGetFirstSupportedAlgorithm() throws Exception {
        assertThat(Encryption.getFirstSupportedAlgorithm(new MockAlgorithm("test", 1), SupportedAlgorithm.AES_128), is(SupportedAlgorithm.AES_128));
    }

    @Test
    public void testCreateKey_Success() throws Exception {
        Assert.assertNotNull(Encryption.createKey(SupportedAlgorithm.AES_128));
        Assert.assertNotNull(Encryption.createKey(SupportedAlgorithm.AES_192));
        Assert.assertNotNull(Encryption.createKey(SupportedAlgorithm.AES_256));
        Assert.assertNotNull(Encryption.createKey(SupportedAlgorithm.DESede_112));
        Assert.assertNotNull(Encryption.createKey(SupportedAlgorithm.DESede_168));
    }

    @Test
    public void testCreateKey_Failed() throws Exception {
        Exception exc = TestUtils.expectException(() -> Encryption.createKey(new MockAlgorithm("AES", 1000)));
        // exception message: "Wrong keysize: must be equal to 128, 192 or 256"
        assertThat(exc.getMessage(), containsString("128"));
        assertThat(exc.getMessage(), containsString("192"));
        assertThat(exc.getMessage(), containsString("256"));
    }

    @Test
    public void testEncryptDecryptRoundup() throws Exception {
        // an example password with some special chars.
        String password = "abc123!\"§$%&/()=?";

        SupportedAlgorithm[] values = SupportedAlgorithm.values();
        for (SupportedAlgorithm algorithm : values) {
            if (Encryption.isAlgorithmSupported(algorithm)) {
                testEncryptDecryptRoundup(algorithm, password);
            }
        }

    }

    private void testEncryptDecryptRoundup(final SupportedAlgorithm algorithm, final String password) {
        final int saltLength = 11;
        Assert.assertFalse(Encryption.isEncryptedValue(password));
        String secretKeyStr = Encryption.toBase64String(Encryption.createKey(algorithm));
        String encryptPw = Encryption.encrypt(algorithm, Encryption.readSecretKey(algorithm, secretKeyStr), saltLength, password);
        Assert.assertTrue(Encryption.isEncryptedValue(encryptPw));
        String finalPassword = Encryption.decrypt(algorithm, Encryption.readSecretKey(algorithm, secretKeyStr), saltLength, encryptPw);
        assertThat(finalPassword, is(password));
    }

    @Test
        public void testIsEncryptedValue() throws Exception {
            Assert.assertFalse(Encryption.isEncryptedValue(null));
        }
}


text:
%% selectBaseAttributes: Select a given set of attributes to be bases
function [X, y] = selectBaseAttributes(raw_data)
	% Y are the leagues of the players
	y = raw_data(:, 2);
	% Attributes from another kaggle kernel
	% X = [raw_data(:, [14, 12, 13, 5, 6, 7])];
	% All attributes
	X = [raw_data(:, 3:columns(raw_data))];
end


text:
const expect = require('chai').expect;

const {
  getStringHash,
} = require('../../../../packages/workbox-build/build/lib/get-string-hash');

describe(`[workbox-build] lib/get-string-hash.js`, function () {
  it(`should return the expected hashes`, function () {
    const stringsToHashes = new Map([
      ['abc', '900150983cd24fb0d6963f7d28e17f72'],
      ['xyz', 'd16fb36f0911f878998c136191af705e'],
    ]);

    for (const [string, hash] of stringsToHashes) {
      expect(getStringHash(string)).to.eql(hash);
    }
  });
});