我有一个在多VM环境中运行的Kafka代理(私有云-带有我们自己的库伯内特斯集群-4节点集群)。
我创建了一个spring-boot应用程序,它有一个发布者,必须在Kafka代理中向Kafka主题发布消息。我有两个集装箱(Kafka经纪人
我无法通过在发布者的bootstrap.servers中提供Kafka的服务名称:端口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
看起来您需要先创建test
主题,然后才能发布任何内容。
您可以查看该主题在 Kafka 集群中是否可用。例如,来自Kafka经纪人播客之一:
bin/kafka-topics.sh --bootstrap-server localhost:9092 --zookeeper zookeeper-svc:2181 --topic test --describe
或者,您可以在Spring启动应用程序上创建它:
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();
}
}
✌️