我终于找到了解决方案，合流kafka python包的文档不足以解决这个问题。但开源的一个好处是，你可以阅读代码并找出答案。因此，要在特定的代理中创建主题，我必须按照下面的方式编写创建主题代码。请注意，我使用了副本分配而不是复制因子。这两者是相互排斥的。如果使用复制因子，分区将由kafka分配，您可以通过副本分配来控制分配。但是，我确信在重新平衡/重新分配分区的情况下，这将被重新分配。但这也可以通过on-u-revoke事件来处理。但现在，这对我很有效。

def createTopic(admin_client,topics):
    #topic_name=topics
    topic_name = ['rajib1_test_xxx_topic']
    replica_assignment = [[262, 261]]
    topic_list = [NewTopic(topic, num_partitions=1, replica_assignment=replica_assignment) for topic in topic_name]
futmap=admin_client.create_topics(topic_list)

# Wait for each operation to finish.

for topic, f in futmap.items():
    try:
        f.result()  # The result itself is None
        print("Topic {} created".format(topic))
    except Exception as e:
        print("Failed to create topic {}: {}".format(topic, e))

# return futmap

你也可以用 kafka-reassign-partitions.sh Kafka提供的工具，用于将一个主题的副本更改为另一个代理。
例如，如果您希望将主题“test”（在本例中为单复制、单分区）放在代理“1”上，可以首先定义一个计划（名为replicachange.json）：

{
"partitions":
  [{"topic": "test", "partition": 0,
    "replicas": [
       1
    ]
  }],
"version":1
}

然后使用以下方法执行：

kafka-reassign-partitions.sh --zookeeper localhost:2181 --execute \
--reassignment-json-file replicachange.json