kafka add partitions function「建议收藏」

kafka add partitions function「建议收藏」代码功能在java代码中调用scala接口addPartitions.使用场景在kafka中如果需要定制kafka-topic的管理,那么其中一个功能很可能会用到:增加partition数量。但是在kafka-1.0.x之上的版本的AdminUtils中预留了相关的apiaddPartitions,具体功能的实现可以参考下面源码(scala):/***Addparti…

大家好,又见面了,我是你们的朋友全栈君。

代码功能

在 java 代码中调用 scala 接口 addPartitions.

使用场景

在kafka中如果需要定制kafka-topic的管理,那么其中一个功能很可能会用到:增加partition数量

但是在kafka-1.0.x之上的版本的AdminUtils中预留了相关的api addPartitions,具体功能的实现可以参考下面源码(scala):

/** * Add partitions to existing topic with optional replica assignment * * @param zkUtils Zookeeper utilities * @param topic Topic for adding partitions to * @param existingAssignment A map from partition id to its assigned replicas * @param allBrokers All brokers in the cluster * @param numPartitions Number of partitions to be set * @param replicaAssignment Manual replica assignment, or none * @param validateOnly If true, validate the parameters without actually adding the partitions * @return the updated replica assignment */
 @deprecated("This method is deprecated and will be replaced by kafka.zk.AdminZkClient.", "1.1.0")
 def addPartitions(zkUtils: ZkUtils,
                   topic: String,
                   existingAssignment: Map[Int, Seq[Int]],
                   allBrokers: Seq[BrokerMetadata],
                   numPartitions: Int = 1,
                   replicaAssignment: Option[Map[Int, Seq[Int]]] = None,
                   validateOnly: Boolean = false): Map[Int, Seq[Int]] = {
   val existingAssignmentPartition0 = existingAssignment.getOrElse(0,
     throw new AdminOperationException(
       s"Unexpected existing replica assignment for topic '$topic', partition id 0 is missing. " +
         s"Assignment: $existingAssignment"))

   val partitionsToAdd = numPartitions - existingAssignment.size
   if (partitionsToAdd <= 0)
     throw new InvalidPartitionsException(
       s"The number of partitions for a topic can only be increased. " +
         s"Topic $topic currently has ${existingAssignment.size} partitions, " +
         s"$numPartitions would not be an increase.")

   replicaAssignment.foreach { proposedReplicaAssignment =>
     validateReplicaAssignment(proposedReplicaAssignment, existingAssignmentPartition0,
       allBrokers.map(_.id).toSet)
   }

   val proposedAssignmentForNewPartitions = replicaAssignment.getOrElse {
     val startIndex = math.max(0, allBrokers.indexWhere(_.id >= existingAssignmentPartition0.head))
     AdminUtils.assignReplicasToBrokers(allBrokers, partitionsToAdd, existingAssignmentPartition0.size,
       startIndex, existingAssignment.size)
   }
   val proposedAssignment = existingAssignment ++ proposedAssignmentForNewPartitions
   if (!validateOnly) {
     info(s"Creating $partitionsToAdd partitions for '$topic' with the following replica assignment: " +
       s"$proposedAssignmentForNewPartitions.")
     // add the combined new list
     AdminUtils.createOrUpdateTopicPartitionAssignmentPathInZK(zkUtils, topic, proposedAssignment, update = true)
   }
   proposedAssignment

 }

复制代码

虽然 java 和 scala 之间的契合性很强,但是上述的参数列表在 java 中实现并调用该接口的复杂度还是比较高的,本人也是花费了一定的时间才调试通过的。

具体实现

下面部分是在 java 中调用该接口的实现:

public class addPartitions {
   public static boolean addPartitions(String zookeeperUri, String topic, int partitionNum) {
   	boolean succeed = true;
       ZkClient zkClient = createZkClient(zookeeperUri);
       ZkUtils zkUtils = ZkUtils.apply(zkClient, JaasUtils.isZkSecurityEnabled());

       //get existing assignment
       Seq<String> names = JavaConverters.asScalaBufferConverter(Arrays.asList(topic)).asScala();
       scala.collection.mutable.Map<String, scala.collection.Map<Object, Seq<Object>>> assignment = (scala.collection.mutable.Map<String, scala.collection.Map<Object, Seq<Object>>>)zkUtils.getPartitionAssignmentForTopics(names);
       Map<String, scala.collection.Map<Object, Seq<Object>>> partitionaAssigmentMap = JavaConverters.mutableMapAsJavaMapConverter(assignment).asJava();

       //get all brokers metadata in the cluster
       Seq<BrokerMetadata> allBrokerMeta = AdminUtils.getBrokerMetadatas(zkUtils, RackAwareMode.Enforced$.MODULE$, scala.Option.apply(null));

       try {
           AdminUtils.addPartitions(zkUtils, topic, partitionaAssigmentMap.get(topic), allBrokerMeta, partitionNum, Option.empty(), false);
       } catch (InvalidPartitionsException exception) {
           System.out.println("exception", exception);
           succeed = false;
       }
       return succeed;
   }
}
复制代码

此外,推荐 java 和 scala 之间格式转化的神器类 JavaConverters .

版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 举报,一经查实,本站将立刻删除。

发布者:全栈程序员-用户IM,转载请注明出处:https://javaforall.cn/153334.html原文链接:https://javaforall.cn

【正版授权,激活自己账号】: Jetbrains全家桶Ide使用,1年售后保障,每天仅需1毛

【官方授权 正版激活】: 官方授权 正版激活 支持Jetbrains家族下所有IDE 使用个人JB账号...

(0)


相关推荐

  • 贪心算法之背包问题

    贪心算法之背包问题

  • 关于用户路径分析模型_spark用户行为分析

    关于用户路径分析模型_spark用户行为分析一、需求背景在互联网数据化运营实践中,有一类数据分析应用是互联网行业所独有的——路径分析。路径分析应用是对特定页面的上下游进行可视化展示并分析用户在使用产品时的路径分布情况。比如:当用户使用某APP时,是怎样从【首页】进入【详情页】的,用户从【首页】分别进入【详情页】、【播放页】、【下载页】的比例是怎样的,以及可以帮助我们分析用户离开的节点是什么。在场景对应到具体的技术方案设计上,我们将访问数据根据session划分,挖掘出用户频繁访问的路径;功能上允许用户即时查看所选节点相关路径,支持用户自定义设

  • HQL 多表联合查询「建议收藏」

    HQL 多表联合查询「建议收藏」HQL联表查询语句select h from Hytxbz as h,Tgbzk as t where h.hytxbzid=t.hytxbzid and t.bztgid=:bztgid如果想取得对应属性的话,也可以这样写 select h.hytxbzid from Hytxbz as h,Tgbzk as t where h.hytxbzid=t.hytxbzid 

  • encoder和decoder的区别_decode作用

    encoder和decoder的区别_decode作用I’veneverbeensurethatIunderstandthedifferencebetweenstr/unicodedecodeandencode.Iknowthatstr().decode()isforwhenyouhaveastringofbytesthatyouknowhasacertaincharacterenco…

  • IntelliJ IDEA 远程debug调试

    IntelliJ IDEA 远程debug调试远程DEBUG的必要性由于部署环境的差异性,相信很多朋友都碰到过开发环境正常测试过的功能在测试环境甚至生产环境下出现bug的情况。一般情况下,生产环境可以采取的手段比较单一,即通过日志的方式获取运行中的环境上下文,分析日志文件并尝试重现bug。这会带来的问题还是不少的,首先,日志的分析是一项比较耗时的工作;其次,现有的日志记录不一定能反映出问题,你可能需要多次重复这个过程(分析日志->猜测问题->加日志->部署->获取日志)来慢慢逼近问题。倘若是测试环境,我们还多了一项可供选择的手

  • netstat命令详解

    netstat命令详解简介netstat命令用于显示各种网络相关信息,如网络连接,路由表,接口状态(InterfaceStatistics),masquerade连接,多播成员(MulticastMember

发表回复

您的电子邮箱地址不会被公开。

关注全栈程序员社区公众号