java - 卡夫卡1.0 : Alter number of partitions for a specific topic using java

Question

Kafka 1.0 中的 AdminUtils.addPartition 调用似乎已更改了 2 版本中的签名，如下所示 - 0.10.2.0 与 1.0.0

============================

卡夫卡0.10.2.0

/**
  * Add partitions to existing topic with optional replica assignment
  *
  * @param zkUtils Zookeeper utilities
  * @param topic Topic for adding partitions to
  * @param numPartitions Number of partitions to be set
  * @param replicaAssignmentStr Manual replica assignment
  * @param checkBrokerAvailable Ignore checking if assigned replica broker             is available. Only used for testing
   */
   def addPartitions(zkUtils: ZkUtils,
                     topic: String,
                     numPartitions: Int = 1,
                     replicaAssignmentStr: String = "",
                     checkBrokerAvailable: Boolean = true,
                     rackAwareMode: RackAwareMode = RackAwareMode.Enforced)

============================

卡夫卡1.0.0

 /**
   * 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
   */
   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]] =

关于如何在 Java 中轻松使用新的 addPartitions 调用而无需指定现有和新的副本分配，有什么想法吗？

Answer 1

试试这个，它对我有用

Seq<String> names = JavaConverters.asScalaBuffer(Arrays.asList(topicname)).toSeq();
        Seq<BrokerMetadata> brokers = AdminUtils.getBrokerMetadatas(zkUtils, Enforced$.MODULE$, Option.empty());

        Map<String, Map<Object, Seq<Object>>> assignment
            = zkUtils.getPartitionAssignmentForTopics(names);
        Option<Map<Object, Seq<Object>>> assgn = assignment.get(topicname);

        AdminUtils.addPartitions(zkUtils, topicname, assgn.get(), brokers, partitions, Option.empty(), false);