无法从运行在多vm k8s集群的另一个pod中的spring引导服务访问运行在pod中的kafka代理

0s7z1bwu  于 2021-06-04  发布在  Kafka
关注(0)|答案(2)|浏览(455)

我有一个kafka代理在多vm环境中运行(私有云-有我们自己的kubernetes集群-4节点集群)。
我创建了一个spring启动应用程序,它有一个发布者,发布者必须将消息发布到kafka代理中的kafka主题。我在同一个kubernetes集群中运行了两个容器(kafka broker和spring boot app)。
我无法通过在发布服务器的bootstrap.servers中提供kafka的服务名称:port id来访问kafka代理(在同一k8s集群中运行)
spring boot的发布服务器配置:

props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-svc:9092");- -> not working

群集信息:
kafka broker yaml文件:

apiVersion: extensions/v1beta1
    kind: Deployment
    metadata:
      name: kafka-deploy
    spec:
      replicas: 1
      selector:
        matchLabels:
          app: kafka
      template:
        metadata:
          labels:
            app: kafka
        spec:
          containers:
          - env:
            - name: KAFKA_PORT
              value: "9092"
            - name: KAFKA_ADVERTISED_PORT
              value: "9092"
            - name: KAFKA_ADVERTISED_HOST_NAME
              value: kafka-svc
            - name: KAFKA_CREATE_TOPICS
              value: "test:1:1"
            - name: KAFKA_ZOOKEEPER_CONNECT
              value: zookeeper-svc:2181
            - name: KAFKA_BROKER_ID
              value: "1"
            name: kafka
            image: wurstmeister/kafka
            ports:
            - containerPort: 9092
    ------
       apiVersion: v1
       kind: Service
       metadata:
         name: kafka-svc
       spec:
         type: NodePort
         ports:
         - port: 9092
           targetPort: 9092
           name: http
           protocol: TCP
         selector:
           app: kafka
           kafka-broker-id: "1"

当我试图将字符串消息推送到kafka主题时,我得到以下异常:

"2020-08-18 13:28:04.525  INFO 1 --- [nio-8080-exec-1] o.a.kafka.common.utils.AppInfoParser     : Kafka startTimeMs: 1597757284523
2020-08-18 13:29:04.539 ERROR 1 --- [nio-8080-exec-1] o.s.k.support.LoggingProducerListener    : Exception thrown when sending a message with key='null' and payload='TEST' to topic test:

org.apache.kafka.common.errors.TimeoutException: Topic test not present in metadata after 60000 ms.

2020-08-18 13:29:04.544 ERROR 1 --- [nio-8080-exec-1] o.a.c.c.C.[.[.[/].[dispatcherServlet]    : Servlet.service() for servlet [dispatcherServlet] in context with path [] threw exception [Request processing failed; nested exception is org.springframework.kafka.KafkaException: Send failed; nested exception is org.apache.kafka.common.errors.TimeoutException: Topic test not present in metadata after 60000 ms.] with root cause
k75qkfdt

k75qkfdt1#

谢谢。我使用环境变量选项创建了主题测试,同时旋转kafka代理pod本身。因此,我们已经在kafka代理系统中进行了主题测试。我怀疑我们在kubernetes中用来连接到kafka代理的kafka服务配置有问题。
问题是kafka服务,我们使用它(从spring-boot应用程序的发布者)连接到kafka代理的主题
当我尝试使用kafka服务连接到kafka代理时,出现了以下异常
bash-4.4#kafka-topics.sh—引导服务器kafka-svc:9092 --topic test——描述执行topic命令时的错误:org.apache.kafka.common.errors.timeoutexception:call(callname=listtopics,deadlinems=1597833365003)在尝试一次后在1597833365004超时[2020-08-19 10:36:05,027]错误java.util.concurrent.executionexception:org.apache.kafka.common.errors.timeoutexception:call(callname=listtopics,deadlinems=1597833365003)在org.apache.kafka.common.internals.kafkafutureimpl.wrapandthrow(kafkafutureimpl)上尝试1次后在1597833365004超时。java:45)访问org.apache.kafka.common.internals.kafkafutureinpl.access$000(kafkafutureinpl。java:32)在org.apache.kafka.common.internals.kafkafutureinpl$singlewaiter.await(kafkafutureinpl。java:89) 在org.apache.kafka.common.internals.kafkafutureinpl.get(kafkafutureinpl。java:260)在kafka.admin.topiccommand$adminclienttopicservice.gettopics(topiccommand。scala:333)在kafka.admin.topiccommand$adminclienttopicservice.describetopic(topiccommand。scala:288)在kafka.admin.topiccommand$.main(topiccommand。scala:68)在kafka.admin.topiccommand.main(topiccommand.scala)原因:org.apache.kafka.common.errors.timeoutexception:调用(callname=listtopics,deadlinems=1597833365003)在尝试1次后在1597833365004超时原因:org.apache.kafka.common.errors.timeoutexception:等待节点分配时超时(kafka.admin.topiccommand$)bash-4.4#
但是,当我尝试通过zookeeper服务连接时,我可以得到如下所示的预期结果


***bash-4.4# kafka-topics.sh --zookeeper zookeeper-svc:2181 --topic test --describe

Topic: test   PartitionCount: 1       ReplicationFactor: 1    Configs:
        Topic: test   Partition: 0    Leader: 1       Replicas: 1     Isr: 1
bash-4.4#***
3ks5zfa0

3ks5zfa02#

看起来你需要创造 test 在你发布任何东西之前。
您可以在kafka集群中查看该主题是否可用。例如,从Kafka的一个经纪人豆荚中:

bin/kafka-topics.sh --bootstrap-server localhost:9092 --zookeeper zookeeper-svc:2181 --topic test --describe

也可以在spring boot应用程序上创建:

package io.stockgeeks.kafka.config;

import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.TopicBuilder;

@Configuration
public class KafkaTopicConfiguration {

  @Bean
  public NewTopic topicExample() {
    return TopicBuilder.name("test")
      .partitions(6)
      .replicas(3)
      .build();
  }
}

✌️

相关问题