Advertisement
Guest User

Untitled

a guest
Feb 7th, 2016
61
0
Never
Not a member of Pastebin yet? Sign Up, it unlocks many cool features!
Scala 0.57 KB | None | 0 0
  1. package com.bestmile
  2. package service
  3.  
  4. import kafka.utils.ZkUtils
  5. import org.I0Itec.zkclient.{ZkClient, ZkConnection}
  6.  
  7. class KafkaAdmin(zkClient: ZkClient) {
  8.   def topics: Seq[String] =  ZkUtils.getAllTopics(zkClient)
  9.   def delete(topic: String): Unit = ZkUtils.createPersistentPath(zkClient, ZkUtils.getDeleteTopicPath(topic))
  10. }
  11.  
  12. object KafkaAdmin {
  13.   def fromConfig(implicit config: Config): KafkaAdmin = {
  14.     import ConfigKeys._
  15.     val addr = config(Zookeeper)
  16.     new KafkaAdmin(new ZkClient(new ZkConnection(addr.getHostString + ":" + addr.getPort)))
  17.   }
  18. }
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement