Kafka 0.8, est-il possible de créer un sujet avec une partition et une réplication en utilisant du code Java?

Dans Kafka 0.8 beta un sujet peut être créé à l'aide d'une commande comme ci-dessous comme mentionné ici

    bin/kafka-create-topic.sh --zookeeper localhost:2181 --replica 2 --partition 3 --topic test

la commande ci-dessus va créer un topic nommé "test" avec 3 partitions et 2 répliques par partition.

Puis-je faire la même chose à l'aide de Java ?

Pour l'instant ce que j'ai trouvé est à l'aide de Java, nous pouvons créer un producteur comme on le voit ci-dessous

    Producer<String, String> producer = new Producer<String, String>(config);
    producer.send(new KeyedMessage<String, String>("mytopic", msg));

Cela va créer un topic nommé "mytopic" avec le numéro de la partition spécifiée à l'aide de la "num.partitions" attribut " et commencer à produire.

Mais est-il un moyen de définir la partition et la réplication également ? Je ne pouvais pas trouver un tel exemple. Si nous ne pouvons pas alors est-ce à dire, nous avons toujours besoin de créer un sujet avec des partitions et de la réplication (conformément à notre exigence) avant et ensuite utiliser le producteur pour produire de message dans le sujet. Par exemple, il sera possible si je veux créer la "mytopic" de la même manière, mais avec différents nombre de partition (en ignorant le num.partitions attribut) ?

source d'informationauteur Hild

  1. 8

    Note: Ma réponse couvre Kafka 0.8.1+, c'est à dire la dernière version stable disponible en avril 2014.

    Oui, vous pouvez créer un sujet par programmation via le Kafka de l'API. Et oui, vous pouvez spécifier le nombre de partitions souhaité ainsi que le facteur de réplication pour le sujet.

    Noter que la publication récente de Kafka 0.8.1+ offre un peu différente de l'API que Kafka 0.8.0 (qui a été utilisé par Biks dans son lié répondre). J'ai ajouté un exemple de code pour créer un sujet dans Kafka 0.8.1+ à ma réponse à la question Comment Pouvons-nous créer un topic dans Kafka à partir de l'IDE à l'aide de l'API que Biks faisait allusion ci-dessus.

  2. 0

    `

    import kafka.admin.AdminUtils;
import kafka.cluster.Broker;
import kafka.utils.ZKStringSerializer$;
import kafka.utils.ZkUtils;

String zkConnect = "localhost:2181";
ZkClient zkClient = new ZkClient(zkConnect, 10 * 1000, 8 * 1000, ZKStringSerializer$.MODULE$);
ZkUtils zkUtils = new ZkUtils(zkClient, new ZkConnection(zkConnect), false);
    Properties pop = new Properties();
    AdminUtils.createTopic(zkUtils, topic.getTopicName(), topic.getPartitionCount(), topic.getReplicationFactor(),
            pop);
    zkClient.close();`