//设置package包名称以及导入依赖的类
package org.argus.jc.incremental.jawa
package remote

import java.net.{ConnectException, InetAddress, UnknownHostException}

import org.argus.jc.incremental.jawa.data.{CompilationData, CompilerData}
import org.jetbrains.jps.incremental.ModuleLevelBuilder.ExitCode


class RemoteServer(val address: InetAddress, val port: Int) extends Server with RemoteResourceOwner {
  def compile(compilerData: CompilerData, compilationData: CompilationData, client: Client): ExitCode = {
    val arguments = Arguments(compilerData, compilationData).asStrings

    try {
      send(serverAlias, arguments, client)
      ExitCode.OK
    } catch {
      case e: ConnectException =>
        val firstLine = s"Cannot connect to compile server at ${address.toString}:$port"
        val secondLine = "Trying to compile without it"
        val message = s"$firstLine\n$secondLine"
        client.warning(message)
        client.debug(s"$firstLine\n${e.toString}\n${e.getStackTrace.mkString("\n")}")
        JawaBuilder.localServer.compile(compilerData, compilationData, client)
      case e: UnknownHostException =>
        val message = "Unknown IP address of compile server host: " + address.toString
        client.error(message)
        client.debug(s"$message\n${e.toString}\n${e.getStackTrace.mkString("\n")}")
        ExitCode.ABORT
    }
  }
} 

//设置package包名称以及导入依赖的类
package eu.inn.kafka.mimic

import java.net.{InetAddress, InetSocketAddress}
import java.util.concurrent.TimeUnit
import java.util.{Timer, TimerTask}

import com.codahale.metrics.graphite.{Graphite, GraphiteReporter}
import nl.grons.metrics.scala.{InstrumentedBuilder, MetricName}

object Metrics extends Logging {
  val metricRegistry = new com.codahale.metrics.MetricRegistry()

  def startReporter(host: String, port: Int, prefix: String, reportPeriod: Long) = {

    val graphite = new Graphite(new InetSocketAddress(host, port))

    val reporter = GraphiteReporter.forRegistry(metricRegistry)
      .prefixedWith(prefix + "." + InetAddress.getLocalHost.getHostName.replaceAll("\\.", "-"))
      .build(graphite)

    new Timer("graphite-reporter-timer").schedule(
      new TimerTask { def run() = reporter.report() },
      reportPeriod,
      reportPeriod
    )
  }
}

trait Metrics extends InstrumentedBuilder {
  override lazy val metricBaseName = MetricName("kafka-mimic")

  lazy val metricRegistry = Metrics.metricRegistry
}

trait MetricsComponent extends Metrics with Logging {

  this: ConfigComponent =>

  def startMetricsReporting() = {
    if (config.hasPath("enabled") && config.getBoolean("enabled")) {
      val host = config.getString("host")
      val port = config.getInt("port")
      log.info(s"Starting graphite reporter for $host:$port")
      Metrics.startReporter(
        host,
        port,
        config.getString("prefix"),
        config.getDuration("report-period", TimeUnit.MILLISECONDS)
      )
    }

    metrics.gauge("heartbeat") { 1 }
  }
} 

//设置package包名称以及导入依赖的类
package com.karasiq.bittorrent.protocol.extensions

import java.net.{InetAddress, InetSocketAddress}
import java.nio.ByteBuffer

import akka.util.ByteString
import com.karasiq.bittorrent.format.{BEncode, BEncodedDictionary, BEncodedString}
import com.karasiq.bittorrent.protocol.{BitTorrentTcpProtocol, TcpMessageProtocol}

trait PeerExchange extends PeerExchangeMessages with PeerExchangeTcp

trait PeerExchangeMessages {
  case class PeerExchangeList(addresses: Seq[InetSocketAddress])
}

trait PeerExchangeTcp { self: PeerExchangeMessages ?
  implicit object PeerExchangeListTcpProtocol extends TcpMessageProtocol[PeerExchangeList] {
    private val ipv4Length: Int = 4
    private val ipv6Length: Int = 16
    private val portLength: Int = 2

    override def toBytes(value: PeerExchangeList): ByteString = {
      val (ipv4, ipv6) = value.addresses.partition(_.getAddress.getAddress.length == ipv4Length)
      def packAddress(address: InetSocketAddress): ByteString = {
        val port = ByteBuffer.allocate(portLength)
        port.putShort(address.getPort.toShort)
        port.flip()
        ByteString(address.getAddress.getAddress) ++ ByteString(port)
      }
      BEncodedDictionary(Vector(
        "dropped" ? BEncodedString(ByteString.empty),
        "added" ? BEncodedString(ipv4.map(packAddress).fold(ByteString.empty)(_ ++ _)),
        "added.f" ? BEncodedString(ByteString(Array.fill(ipv4.length)(1.toByte))),
        "added6" ? BEncodedString(ipv6.map(packAddress).fold(ByteString.empty)(_ ++ _)),
        "added6.f" ? BEncodedString(ByteString(Array.fill(ipv6.length)(1.toByte)))
      )).toBytes
    }

    override def fromBytes(bs: ByteString): Option[PeerExchangeList] = {
      import com.karasiq.bittorrent.format.BEncodeImplicits._
      BEncode.parse(bs.toArray[Byte]).collectFirst {
        case BEncodedDictionary(values) ?
          val map = values.toMap
          val ipv4 = map.byteString("added").fold(Iterator[ByteString]())(_.grouped(ipv4Length + portLength))
          val ipv6 = map.byteString("added6").fold(Iterator[ByteString]())(_.grouped(ipv6Length + portLength))
          val addresses = (ipv4 ++ ipv6).map { bytes ?
            val address = InetAddress.getByAddress(bytes.dropRight(portLength).toArray)
            val port = BitTorrentTcpProtocol.int32FromBytes(bytes.takeRight(portLength))
            new InetSocketAddress(address, port)
          }
          PeerExchangeList(addresses.toVector)
      }
    }
  }
} 

//设置package包名称以及导入依赖的类
package akka.remote.transport.netty

import akka.actor.Address
import akka.remote.transport.AssociationHandle
import akka.remote.transport.AssociationHandle.{ HandleEventListener, InboundPayload }
import akka.remote.transport.Transport.AssociationEventListener
import akka.util.ByteString
import java.net.{ SocketAddress, InetAddress, InetSocketAddress }
import org.jboss.netty.buffer.{ ChannelBuffer, ChannelBuffers }
import org.jboss.netty.channel._
import scala.concurrent.{ Future, Promise }


private[remote] class UdpAssociationHandle(val localAddress: Address,
                                           val remoteAddress: Address,
                                           private val channel: Channel,
                                           private val transport: NettyTransport) extends AssociationHandle {

  override val readHandlerPromise: Promise[HandleEventListener] = Promise()

  override def write(payload: ByteString): Boolean = {
    if (!channel.isConnected)
      channel.connect(new InetSocketAddress(InetAddress.getByName(remoteAddress.host.get), remoteAddress.port.get))

    if (channel.isWritable && channel.isOpen) {
      channel.write(ChannelBuffers.wrappedBuffer(payload.asByteBuffer))
      true
    } else false
  }

  override def disassociate(): Unit = try channel.close()
  finally transport.udpConnectionTable.remove(transport.addressToSocketAddress(remoteAddress))

} 

//设置package包名称以及导入依赖的类
package akka.remote.testconductor

import akka.testkit.AkkaSpec
import akka.actor.{ PoisonPill, Props, AddressFromURIString }
import akka.testkit.ImplicitSender
import akka.remote.testconductor.Controller.NodeInfo
import java.net.InetSocketAddress
import java.net.InetAddress

object ControllerSpec {
  val config = """
    akka.testconductor.barrier-timeout = 5s
    akka.actor.provider = akka.remote.RemoteActorRefProvider
    akka.actor.debug.fsm = on
    akka.actor.debug.lifecycle = on
    """
}

class ControllerSpec extends AkkaSpec(ControllerSpec.config) with ImplicitSender {

  val A = RoleName("a")
  val B = RoleName("b")

  "A Controller" must {

    "publish its nodes" in {
      val c = system.actorOf(Props(classOf[Controller], 1, new InetSocketAddress(InetAddress.getLocalHost, 0)))
      c ! NodeInfo(A, AddressFromURIString("akka://sys"), testActor)
      expectMsg(ToClient(Done))
      c ! NodeInfo(B, AddressFromURIString("akka://sys"), testActor)
      expectMsg(ToClient(Done))
      c ! Controller.GetNodes
      expectMsgType[Iterable[RoleName]].toSet should be(Set(A, B))
      c ! PoisonPill // clean up so network connections don't accumulate during test run
    }

  }

} 

//设置package包名称以及导入依赖的类
package com.box.castle

import java.net.InetAddress

import akka.actor.ActorSystem
import akka.pattern.gracefulStop
import com.box.castle.committer.api.{CommitterFactory, TopicFilter}
import com.box.castle.consumer.offsetmetadatamanager.ZookeeperOffsetMetadataManagerFactory
import com.box.castle.consumer.{CastleSimpleConsumerFactory, ClientId, ConsumerId}
import com.box.castle.core.{CuratorFactory, util}
import com.box.castle.core.committer.CommitterActorFactory
import com.box.castle.core.committer.manager.CommitterManagerActorFactory
import com.box.castle.core.common.GracefulShutdown
import com.box.castle.core.config.CastleConfig
import com.box.castle.core.leader.{LeaderActorFactory, LeaderFactory, TaskManager}
import com.box.castle.core.supervisor.CastleSupervisorFactory
import com.box.castle.core.worker.{WorkerActorFactory, WorkerFactory}
import com.box.castle.metrics.MetricsLogger
import com.box.castle.router.RouterFactory
import com.box.castle.router.kafkadispatcher.KafkaDispatcherFactory
import com.box.castle.router.proxy.KafkaDispatcherProxyPoolFactory
import org.slf4s.Logging
import java.nio.file.{Path, Paths}

import scala.concurrent.Await


  def stop(): Unit = {
    val timeout = castleConfig.gracefulShutdownTimeout
    val supervisor = Await.result(system.actorSelection("/user/castle-supervisor").resolveOne(timeout), timeout)
    try {
      Await.result(gracefulStop(supervisor, timeout, GracefulShutdown), timeout)
      log.info("CastleSupervisor has shutdown gracefully, proceeding with shut down of the Akka System")
    } catch {
      case e @ (_: akka.pattern.AskTimeoutException | _: java.util.concurrent.TimeoutException) => {
        log.warn(s"Failed to gracefully shut down CastleSupervisor within the specified grace period of: $timeout")
      }
    }
    system.shutdown()
  }
}

object Castle {
  def apply(castleConfig: CastleConfig, metricsLogger: MetricsLogger = MetricsLogger.defaultLogger, clientId: Option[ClientId] = None): Castle = {
    new Castle(castleConfig,
      metricsLogger, clientId.getOrElse(ClientId(castleConfig.namespace + "-" + InetAddress.getLocalHost.toString)))
  }
} 

//设置package包名称以及导入依赖的类
package com.github.shadowsocks.acl

import java.net.InetAddress
import java.util.Objects

import com.github.shadowsocks.utils.Utils


@throws[IllegalArgumentException]
class Subnet(val address: InetAddress, val prefixSize: Int) extends Comparable[Subnet] {
  private def addressLength = address.getAddress.length << 3
  if (prefixSize < 0 || prefixSize > addressLength) throw new IllegalArgumentException

  override def toString: String =
    if (prefixSize == addressLength) address.getHostAddress else address.getHostAddress + '/' + prefixSize

  override def compareTo(that: Subnet): Int = {
    val addrThis = address.getAddress
    val addrThat = that.address.getAddress
    var result = addrThis lengthCompare addrThat.length // IPv4 address goes first
    if (result != 0) return result
    for ((x, y) <- addrThis zip addrThat) {
      result = (x & 0xFF) compare (y & 0xFF)  // undo sign extension of signed byte
      if (result != 0) return result
    }
    prefixSize compare that.prefixSize
  }

  override def equals(other: Any): Boolean = other match {
    case that: Subnet => address == that.address && prefixSize == that.prefixSize
    case _ => false
  }
  override def hashCode: Int = Objects.hash(address, prefixSize: Integer)
}

object Subnet {
  @throws[IllegalArgumentException]
  def fromString(value: String): Subnet = {
    val parts = value.split("/")
    val addr = Utils.parseNumericAddress(parts(0))
    parts.length match {
      case 1 => new Subnet(addr, addr.getAddress.length << 3)
      case 2 => new Subnet(addr, parts(1).toInt)
      case _ => throw new IllegalArgumentException()
    }
  }
} 

//设置package包名称以及导入依赖的类
package conf.connection

import java.net.InetAddress

import com.datastax.driver.core.Cluster
import com.typesafe.config.ConfigFactory
import com.websudos.phantom.connectors.{KeySpace, SessionProvider}
import com.websudos.phantom.dsl.Session

import scala.collection.JavaConversions._



object Config {
  val config = ConfigFactory.load()

}

trait DefaultsConnector extends SessionProvider {
  val config = ConfigFactory.load()
  implicit val space: KeySpace = DataConnection.keySpace
  val cluster = DataConnection.cluster
  override implicit lazy val session: Session = DataConnection.session
}

object DataConnection {
  val config = ConfigFactory.load()
  val hosts: Seq[String] = Config.config.getStringList("cassandra.host").toList
  val inets = hosts.map(InetAddress.getByName)
  val keySpace: KeySpace = KeySpace(Config.config.getString("cassandra.keyspace"))
  val cluster =
    Cluster.builder()
      .addContactPoints(inets)
      .withClusterName(Config.config.getString("cassandra.cluster"))
//      .withCredentials(config.getString("cassandra.username"), config.getString("cassandra.password"))
      .build()
  val session: Session = cluster.connect(keySpace.name)
} 

//设置package包名称以及导入依赖的类
package edu.uw.at.iroberts.wirefugue.pcap

import org.scalatest.{Matchers, WordSpecLike}


class IPAddressSpec extends WordSpecLike with Matchers {
  val ipString: String = "192.168.42.196"
  val ipBytes: Seq[Byte] = Seq(0xc0, 0xa8, 0x2a, 0xc4).map(_.toByte)
  "an IPAddress" should {
    "be constructable from String and Seq[Byte]" in {
      IPAddress(ipString) shouldEqual IPAddress(ipBytes)
    }

    "produce a string representation" in {
      val a = IPAddress(ipBytes)

      a.toString shouldEqual ipString
    }

    "produce a byte sequence representation" in {
      val a = IPAddress(ipString)

      a.bytes shouldEqual ipBytes
    }

    "provide an equivalent java.net.Inet4Address" in {
      import java.net.{InetAddress, Inet4Address}

      val a = IPAddress(ipString)

      a.inet4Address shouldEqual InetAddress.getByAddress(ipBytes.toArray).asInstanceOf[Inet4Address]
    }

    "reject invalid sequences upon construction" in {
      assertThrows[IllegalArgumentException] {
        IPAddress(Seq[Byte](0x10, 0x20, 0x30, 0x40, 0x50))
      }
    }

    "reject invalid strings upon construction" in {
      assertThrows[IllegalArgumentException] {
        IPAddress("192.168.0.256")
      }
      assertThrows[IllegalArgumentException] {
        IPAddress("10.0.0.0.1")
      }
      assertThrows[IllegalArgumentException] {
        IPAddress("192.168.2")
      }
    }
  }

} 

//设置package包名称以及导入依赖的类
package akka.stream.alpakka.ftp

import akka.NotUsed
import akka.stream.alpakka.ftp.FtpCredentials.AnonFtpCredentials
import akka.stream.alpakka.ftp.scaladsl.Sftp
import akka.stream.IOResult
import akka.stream.scaladsl.{Sink, Source}
import akka.util.ByteString

import scala.concurrent.Future
import java.net.InetAddress

trait BaseSftpSpec extends SftpSupportImpl with BaseSpec {

  //#create-settings
  val settings = SftpSettings(
    InetAddress.getByName("localhost"),
    getPort,
    AnonFtpCredentials,
    strictHostKeyChecking = false,
    knownHosts = None,
    sftpIdentity = None
  )
  //#create-settings

  protected def listFiles(basePath: String): Source[FtpFile, NotUsed] =
    Sftp.ls(basePath, settings)

  protected def retrieveFromPath(path: String): Source[ByteString, Future[IOResult]] =
    Sftp.fromPath(path, settings)

  protected def storeToPath(path: String, append: Boolean): Sink[ByteString, Future[IOResult]] =
    Sftp.toPath(path, settings, append)
} 

//设置package包名称以及导入依赖的类
package akka.stream.alpakka.ftp

import akka.NotUsed
import akka.stream.alpakka.ftp.FtpCredentials.AnonFtpCredentials
import akka.stream.alpakka.ftp.scaladsl.Ftps
import akka.stream.IOResult
import akka.stream.scaladsl.{Sink, Source}
import akka.util.ByteString

import scala.concurrent.Future
import java.net.InetAddress

trait BaseFtpsSpec extends FtpsSupportImpl with BaseSpec {

  //#create-settings
  val settings = FtpsSettings(
    InetAddress.getByName("localhost"),
    getPort,
    AnonFtpCredentials,
    binary = true,
    passiveMode = true
  )
  //#create-settings

  protected def listFiles(basePath: String): Source[FtpFile, NotUsed] =
    Ftps.ls(basePath, settings)

  protected def retrieveFromPath(path: String): Source[ByteString, Future[IOResult]] =
    Ftps.fromPath(path, settings)

  protected def storeToPath(path: String, append: Boolean): Sink[ByteString, Future[IOResult]] =
    Ftps.toPath(path, settings, append)

} 

//设置package包名称以及导入依赖的类
package com.soteradefense.dga.graphx.louvain

import java.net.InetAddress
import java.nio.ByteBuffer

object IpAddress {
  
  def toString(address: Long) = {
    val byteBuffer = ByteBuffer.allocate(8)
    val addressBytes = byteBuffer.putLong(address)
    // The below is needed because we don't have an unsigned Long, and passing a byte array
    // with more than 4 bytes causes InetAddress to interpret it as a (bad) IPv6 address
    val tmp = new Array[Byte](4)
    Array.copy(addressBytes.array, 4, tmp, 0, 4)
    InetAddress.getByAddress(tmp).getHostAddress()
  }
  
  
  def toLong(_address: String): Long = {
    val address = try {
      InetAddress.getByName(_address)
    } catch {
      case e: Throwable => throw new IllegalArgumentException("Could not parse address: " + e.getMessage)
    }
    val addressBytes = address.getAddress
    val bb = ByteBuffer.allocate(8)
    addressBytes.length match {
      case 4 =>
        bb.put(Array[Byte](0,0,0,0)) // Need a filler
        bb.put(addressBytes)
      case n =>
        throw new IndexOutOfBoundsException("Expected 4 byte address, got " + n)
    }
    bb.getLong(0)
  }

} 

//设置package包名称以及导入依赖的类
package models

import scala.util.Try

import java.net.InetAddress



object Ip {
  def parse(ip: String): Option[InetAddress] = {
    Try(InetAddress.getByName(ip)).toOption
  }

  def removeLastBytes(ip: String): Option[String] = {
    parse(ip).map(_.getHostAddress).map(address => {
      // IPv6 addresses do not have dots
      if(address.contains('.')) {
        address.split('.').take(2).mkString(".")
      }
      else {
        address.split(':').take(3).mkString(":")
      }
    })
  }
} 

//设置package包名称以及导入依赖的类
package cz.alenkacz.db.postgresscala

import java.net.InetAddress
import java.sql.Time
import java.time.Instant
import java.util.UUID

trait DbValue {
  def string: String
  def stringOpt: Option[String]
  def strings: Seq[String]
  def int: Int
  def intOpt: Option[Int]
  def ints: Seq[Int]
  def bigInt: BigInt
  def bigIntOpt: Option[BigInt]
  def bigInts: Seq[BigInt]
  def double: Double
  def doubleOpt: Option[Double]
  def doubles: Seq[Double]
  def float: Float
  def floatOpt: Option[Float]
  def floats: Seq[Float]
  def long: Long
  def longOpt: Option[Long]
  def longs: Seq[Long]
  def bool: Boolean
  def boolOpt: Option[Boolean]
  def bools: Seq[Boolean]
  def short: Short
  def shortOpt: Option[Short]
  def shorts: Seq[Short]
  def inetAddress: InetAddress
  def inetAddresses: Seq[InetAddress]
  def inetAddressOpt: Option[InetAddress]
  def uuid: UUID
  def uuids: Seq[UUID]
  def uuidOpt: Option[UUID]
  def instant: Instant
  def instantOpt: Option[Instant]
  def time: Time
  def timeOpt: Option[Time]
  def bytes: Seq[Byte]
  def bytesOpt: Option[Seq[Byte]]
  def any: Any
} 

//设置package包名称以及导入依赖的类
package hk.edu.polyu.datamining.pamap2

import java.io.{BufferedReader, InputStreamReader}
import java.net.{InetAddress, NetworkInterface, URL, UnknownHostException}

import scala.collection.JavaConversions._
import scala.language.postfixOps

object HostIP {

  
  def load(): Option[String] = {
    val interfaces = NetworkInterface.getNetworkInterfaces
    val interface = interfaces find (_.getName equals "eth0")

    interface flatMap { inet =>
      // the docker adress should be siteLocal
      inet.getInetAddresses find (_ isSiteLocalAddress) map (_ getHostAddress)
    }
  }

  def all(): java.util.List[String] =
    NetworkInterface.getNetworkInterfaces
      .flatMap {
        _.getInetAddresses.map(_.getHostAddress)
      }
      .filterNot(ip => ip.contains("%"))
      .toList

  def hexToString(byte: Byte): String =
    byte match {
      case 10 => "A"
      case 11 => "B"
      case 12 => "C"
      case 13 => "D"
      case 14 => "E"
      case 15 => "F"
      case x: Byte => ("0".toByte + x).toChar + ""
    }

  def +(a: String, b: String) = a + b

  def PublicIP: String = {
    val reader = new BufferedReader(new InputStreamReader(new URL("http://icanhazip.com").openStream()))
    val ip = reader.readLine()
    reader.close()
    ip
  }

  def LocalIP: String = try {
    InetAddress.getLocalHost.getHostAddress
  } catch {
    case e: UnknownHostException => "127.0.0.1"
    case e: Throwable => throw e
  }
} 

//设置package包名称以及导入依赖的类
package korolev.blazeServer

import java.net.InetAddress
import java.security.KeyStore
import javax.net.ssl.{KeyManagerFactory, SSLContext}

import org.http4s.blaze.util.BogusKeystore

import scala.concurrent.ExecutionContextExecutorService


case class BlazeServerConfig(
  port: Int = 8080,
  host: String = InetAddress.getLoopbackAddress.getHostAddress,
  sslContext: Option[SSLContext] = None,
  bufferSize: Int = 8 * 1024,
  maxRequestBodySize: Int = 8 * 1024 * 1024,
  maxRequestHeaderSize: Int = 10 * 1024,
  doNotBlockCurrentThread: Boolean = false
)(
  // Trampoline
  implicit val executionContext: ExecutionContextExecutorService
)

object BlazeServerConfig {

  def bogusSslContext: SSLContext = {
    val ksStream = BogusKeystore.asInputStream()
    assert(ksStream != null)
    val ks = KeyStore.getInstance("JKS")
    ks.load(ksStream, BogusKeystore.getKeyStorePassword)
    val kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm)
    kmf.init(ks, BogusKeystore.getCertificatePassword)
    val context = SSLContext.getInstance("SSL")
    context.init(kmf.getKeyManagers, null, null)
    context
  }
} 

//设置package包名称以及导入依赖的类
package com.github.jparkie.spark.cassandra.client

import java.net.InetAddress

import com.datastax.spark.connector.cql.{ AuthConf, CassandraConnector }
import com.github.jparkie.spark.cassandra.conf.SparkCassServerConf
import org.apache.spark.Logging

import scala.collection.mutable

private[cassandra] trait SparkCassSSTableLoaderClientManager extends Serializable with Logging {
  case class SessionKey(
    hosts:               Set[InetAddress],
    port:                Int,
    authConf:            AuthConf,
    sparkCassServerConf: SparkCassServerConf
  ) extends Serializable

  @transient
  private[client] val internalClients = mutable.HashMap.empty[SessionKey, SparkCassSSTableLoaderClient]

  private[client] def buildSessionKey(
    cassandraConnector:  CassandraConnector,
    sparkCassServerConf: SparkCassServerConf
  ): SessionKey = {
    SessionKey(cassandraConnector.hosts, cassandraConnector.port, cassandraConnector.authConf, sparkCassServerConf)
  }

  private[client] def buildClient(
    cassandraConnector:  CassandraConnector,
    sparkCassServerConf: SparkCassServerConf
  ): SparkCassSSTableLoaderClient = {
    val newSession = cassandraConnector.openSession()

    logInfo(s"Created SSTableLoaderClient to the following Cassandra nodes: ${cassandraConnector.hosts}")

    val sparkCassSSTableLoaderClient = new SparkCassSSTableLoaderClient(newSession, sparkCassServerConf)

    sys.addShutdownHook {
      logInfo("Closed Cassandra Session for SSTableLoaderClient.")

      sparkCassSSTableLoaderClient.stop()
    }

    sparkCassSSTableLoaderClient
  }

  
  private[cassandra] def evictAll(): Unit = synchronized {
    internalClients.values.foreach(_.stop())
    internalClients.clear()
  }
}

object SparkCassSSTableLoaderClientManager extends SparkCassSSTableLoaderClientManager 

//设置package包名称以及导入依赖的类
package com.github.jparkie.spark.cassandra

import java.net.{ InetAddress, InetSocketAddress }

import com.datastax.driver.core.Session
import com.datastax.spark.connector.cql.CassandraConnector
import org.cassandraunit.utils.EmbeddedCassandraServerHelper
import org.scalatest.{ BeforeAndAfterAll, Suite }

trait CassandraServerSpecLike extends BeforeAndAfterAll { this: Suite =>
  // Remove protected modifier because of SharedSparkContext.
  override def beforeAll(): Unit = {
    super.beforeAll()

    EmbeddedCassandraServerHelper.startEmbeddedCassandra()
  }

  // Remove protected modifier because of SharedSparkContext.
  override def afterAll(): Unit = {
    EmbeddedCassandraServerHelper.cleanEmbeddedCassandra()

    super.afterAll()
  }

  def getClusterName: String = {
    EmbeddedCassandraServerHelper.getClusterName
  }

  def getHosts: Set[InetAddress] = {
    val temporaryAddress =
      new InetSocketAddress(EmbeddedCassandraServerHelper.getHost, EmbeddedCassandraServerHelper.getNativeTransportPort)
        .getAddress

    Set(temporaryAddress)
  }

  def getNativeTransportPort: Int = {
    EmbeddedCassandraServerHelper.getNativeTransportPort
  }

  def getRpcPort: Int = {
    EmbeddedCassandraServerHelper.getRpcPort
  }

  def getCassandraConnector: CassandraConnector = {
    CassandraConnector(hosts = getHosts, port = getNativeTransportPort)
  }

  def createKeyspace(session: Session, keyspace: String): Unit = {
    session.execute(
      s"""CREATE KEYSPACE "$keyspace"
          |WITH REPLICATION = { 'class' : 'SimpleStrategy', 'replication_factor' : 1 };
       """.stripMargin
    )
  }
} 

//设置package包名称以及导入依赖的类
package sinks

import java.net.InetAddress
import java.util

import com.typesafe.config.ConfigFactory
import org.apache.flink.api.common.functions.RuntimeContext
import org.apache.flink.streaming.connectors.elasticsearch.{IndexRequestBuilder, ElasticsearchSink}
import org.elasticsearch.action.index.IndexRequest
import org.elasticsearch.client.Requests
import org.elasticsearch.common.transport.{InetSocketTransportAddress, TransportAddress}
import org.joda.time.DateTime

trait ElasticsearchSinks {

  val appConfiguration = ConfigFactory.load()

  val esConfiguration = new util.HashMap[String, String]
  esConfiguration.put("bulk.flush.max.actions", "1")
  esConfiguration.put("cluster.name", appConfiguration.getString("ElasticSearchCluster"))

  val esTransports = new util.ArrayList[TransportAddress]
  esTransports.add(new InetSocketTransportAddress(InetAddress.getByName(appConfiguration.getString("ElasticSearchHost")), appConfiguration.getInt("ElasticSearchPort")))

  val esPopularLanguagesSink = new ElasticsearchSink(esConfiguration, esTransports, new IndexRequestBuilder[(DateTime, String,Int)] {
    override def createIndexRequest(element: (DateTime,String,Int), ctx: RuntimeContext): IndexRequest = {
      val json = new util.HashMap[String, AnyRef]
      json.put("timestamp", element._1.asInstanceOf[AnyRef])
      json.put("language", element._2.asInstanceOf[AnyRef])
      json.put("cnt", element._3.asInstanceOf[AnyRef])
      Requests.indexRequest.index("twitter").`type`("popular-languages").source(json)
    }
  })

} 

//设置package包名称以及导入依赖的类
import java.net.InetAddress


object Main extends App {


  val ipAddress = InetAddress.getByName("172.17.0.2")
  val port = 2181

  val zooKeeper = new ZooKeeper(ipAddress, port)

  zooKeeper.zooKeeper.getChildren("/",false)

  val root = Root(zooKeeper, "participant", "master")


  val master1 = MasterNode(zooKeeper,root.masterPath)
  val master2 = MasterNode(zooKeeper,root.masterPath)
  val master3 = MasterNode(zooKeeper,root.masterPath)
  val master4 = MasterNode(zooKeeper,root.masterPath)
  val master5 = MasterNode(zooKeeper,root.masterPath)
  val master6 = MasterNode(zooKeeper,root.masterPath)
  root.masterPath.children += master1
  root.masterPath.children += master2
  root.masterPath.children += master3
  root.masterPath.children += master4
  root.masterPath.children += master5
  root.masterPath.children += master6

  val participant = ParticipantNode(zooKeeper,root.participantPath, master1)
  root.participantPath.children += participant


  val agent1 = Agent("192.168.0.1","2222","1"); val dataNode1 = DataNode(zooKeeper,participant, agent1)
  val agent2 = Agent("192.168.0.2","2222","1"); val dataNode2 = DataNode(zooKeeper,participant, agent2)
  val agent3 = Agent("192.168.0.3","2222","1"); val dataNode3 = DataNode(zooKeeper,participant, agent3)
  val agent4 = Agent("192.168.0.4","2222","1"); val dataNode4 = DataNode(zooKeeper,participant, agent4)
  val agent5 = Agent("192.168.0.5","2222","1"); val dataNode5 = DataNode(zooKeeper,participant, agent5)
  val agent6 = Agent("192.168.0.6","2222","1"); val dataNode6 = DataNode(zooKeeper,participant, agent6)

  participant.addChildAndCreate(dataNode1)
  participant.addChildAndCreate(dataNode2)
  participant.addChildAndCreate(dataNode3)
  participant.addChildAndCreate(dataNode4)
  participant.addChildAndCreate(dataNode5)
  participant.addChildAndCreate(dataNode6)

}