java Apache Kafka LEADER_不可用

3 周，2 日 Questions & Answers 17056

我遇到了一个关于阿帕奇·卡夫卡的问题，我不明白。我在我的代理中订阅了一个名为“topicreceived”的主题。代码如下：

protected String readResponse(final String idMessage) {
    if (props != null) {
        kafkaClient = new KafkaConsumer<>(props);
        logger.debug("Subscribed to topic-received");
        kafkaClient.subscribe(Arrays.asList("topic-received"));
        logger.debug("Waiting for reading : topic-received");
        ConsumerRecords<String, String> records =    
                       kafkaClient.poll(kafkaConfig.getRead_timeout());

        if (records != null) {
            for (ConsumerRecord<String, String> record : records) {
                logger.debug("Resultado devuelto : "+record.value());
                return record.value();
            }
        }
    }
    return null;
}

当这发生时，我从另一个点向“topic received”发送一条消息。代码如下所示：

private void sendMessageToKafkaBroker(String idTopic, String value) {
    Producer<String, String> producer = null;
    try {
        producer = new KafkaProducer<String, String>(mapProperties());
        ProducerRecord<String, String> producerRecord = new 
                   ProducerRecord<String, String>("topic-received", value);
        producer.send(producerRecord);
        logger.info("Sended value "+value+" to topic-received");
    } catch (ExceptionInInitializerError eix) {
        eix.printStackTrace();
    } catch (KafkaException ke) {
        ke.printStackTrace();
    } finally {
        if (producer != null) {
            producer.close();
        }
    }
}

我第一次尝试时，当主题为“收到主题”时，我收到了如下警告

"WARN 13164 --- [nio-8085-exec-3] org.apache.kafka.clients.NetworkClient   :  
 Error while fetching metadata with correlation id 1 : {topic-  
 received=LEADER_NOT_AVAILABLE}"

但是如果我再试一次，对于这个主题“topic received”，工作正常，并且不会显示任何警告。无论如何，这对我来说没什么用处，因为我每次都要听一个主题，然后发送到一个新的主题（由字符串标识符引用，例如：…12Erw45-2345Saf-234DASDFasd）

为了寻找谷歌没有的LEADER，一些人谈到了添加到服务器。下一行的属性：

host.name=127.0.0.1
advertised.port=9092
advertised.host.name=127.0.0.1

但这对我不起作用（不知道为什么）

我尝试使用以下代码在整个过程之前创建主题：

 private void createTopic(String idTopic) {
    String zookeeperConnect = "localhost:2181";
    ZkClient zkClient = new ZkClient(zookeeperConnect,10000,10000, 
    ZKStringSerializer$.MODULE$);
    ZkUtils zkUtils = new ZkUtils(zkClient, new 
    ZkConnection(zookeeperConnect),false);
    if(!AdminUtils.topicExists(zkUtils,idTopic)) {
        AdminUtils.createTopic(zkUtils, idTopic, 2, 1, new Properties(), 
    null);
        logger.debug("Created topic "+idTopic+" by super user");
    }
    else{
        logger.debug("topic "+idTopic+" already exists");
    }
  }

没有错误，但它仍在侦听，直到超时

我已经查看了经纪人的属性，以检查是否有任何帮助，但我还没有发现任何足够清楚的内容。我用于阅读的道具有：

    props = new Properties();
    props.put("bootstrap.servers", kafkaConfig.getBootstrap_servers());
    props.put("key.deserializer", kafkaConfig.getKey_deserializer());
    props.put("value.deserializer", kafkaConfig.getValue_deserializer());
    props.put("key.serializer", kafkaConfig.getKey_serializer());
    props.put("value.serializer", kafkaConfig.getValue_serializer());
    props.put("group.id",kafkaConfig.getGroupId());

还有，为了发送

   Properties props = new Properties();
    props.put("bootstrap.servers", kafkaConfig.getHost() + ":" + 
    kafkaConfig.getPort());
    props.put("group.id", kafkaConfig.getGroup_id());
    props.put("enable.auto.commit", kafkaConfig.getEnable_auto_commit());
    props.put("auto.commit.interval.ms", 
    kafkaConfig.getAuto_commit_interval_ms());
    props.put("session.timeout.ms", kafkaConfig.getSession_timeout_ms());
    props.put("key.deserializer", kafkaConfig.getKey_deserializer());
    props.put("value.deserializer", kafkaConfig.getValue_deserializer());
    props.put("key.serializer", kafkaConfig.getKey_serializer());
    props.put("value.serializer", kafkaConfig.getValue_serializer());

有线索吗？为什么，我必须使用来自代理和主题的消息的唯一方法是在错误后重复请求

提前谢谢

Python中文网

有 Java 编程相关的问题?

java Apache Kafka LEADER_不可用

共 (2) 个答案

# 1 楼答案

# 2 楼答案

当试图向不存在的主题生成消息时，会发生这种情况