Impossible de créer des sujets internes - exception Stream fil

Question

Je suis en train d'exécuter une simple application de flux de Wordcount mais je face à l'erreur « Impossible de créer des sujets internes - exception Stream thread »

J'ai vu un fil similaire mais cela semble être plus un problème de réseau.

Aucune sécurité n'est activée sur le courtier kafka.

Un seul courtier est configuré et toujours ce problème.

Quelqu'un peut-il me faire savoir comment résoudre ce problème?

Answer 1

Voici la trace de la pile d'erreurs ci-dessous.

org.apache.kafka.streams.processor.internals.InternalTopicManager) [01/07/2017 22: 31: 33132] WARN Impossible de créer des sujets internes: Impossible de créer le sujet: wordcount-lambda-exemple-Counts1 -repartition due to Ce n'est pas le contrôleur correct pour ce cluster. Retry # 3 (org.apache.kafka.streams.processor.internals.InternalTopicManager) [2017-07-01 22: 31: 33,441] WARN Impossible de créer des sujets internes: Impossible de créer le sujet: wordcount-lambda-example-Counts1 -repartition due to Ce n'est pas le contrôleur correct pour ce cluster. Retry # 4 (org.apache.kafka.streams.processor.internals.InternalTopicManager) [2017-07-01 22: 31: 33,567] WARN flux-fil [StreamThread-1] Transition d'état inattendue de PARTITIONS_REVOKED à NOT_RUNNING. (org.apache.kafka.streams.processor.internals.StreamThread) Exception dans le fil "StreamThread-1" org.apache.kafka.streams.errors.StreamsException: Impossible de créer des sujets internes. à org.apache.kafka.streams.processor.internals.InternalTopicManager.makeReady (InternalTopicManager.java:81) à org.apache.kafka.streams.processor.internals.StreamPartitionAssignor.prepareTopic (StreamPartitionAssignor.java:628) à org.apache.kafka.streams.processor.internals.StreamPartitionAssignor.assign (StreamPartitionAssignor.java:382) à org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.performAssignment (ConsumerCoordinator.java:343) à org. apache.kafka.clients.consumer.internals.AbstractCoordinator.onJoinLeader (AbstractCoordinator.java:501)

Peut-être que je devrais essayer de redémarrer Zookeeper et courtier?

Answer 2

Nettoyez vos files d'attente kafka temporaires. Exécutez la commande --list sur kafka pour voir toutes les files d'attente commençant par vos noms et se terminant par -changelog & -repartition et de les supprimer manuellement. Celui-ci a fonctionné pour moi.

De même, vérifiez vos paramètres sur delete.topic.enable pour que la suppression se fasse réellement. Ce n'était pas le paramètre par défaut jusqu'à 1.0.0 - voir https://issues.apache.org/jira/browse/KAFKA-5384