如何在spring-kafka中检测kafka连接问题并切换到不同的集群？
在Spring Kafka中检测Kafka连接问题并切换到不同的集群，可以通过以下步骤实现：
- 配置多个Kafka集群的连接信息：在Spring Kafka的配置文件中，配置多个Kafka集群的连接信息，包括bootstrap.servers、key.serializer、value.serializer等参数。每个集群的连接信息应该分别配置。
- 实现自定义的Kafka连接检测逻辑：可以通过编写自定义的HealthIndicator来检测Kafka连接是否正常。HealthIndicator是Spring Boot提供的一种用于检测应用程序各种组件健康状态的机制。在自定义的HealthIndicator中，可以使用Kafka的AdminClient来检测Kafka连接是否正常。
- 实现切换逻辑：当检测到某个Kafka集群的连接出现问题时，可以通过修改Spring Kafka的配置，将消息发送到另一个可用的Kafka集群。可以通过动态修改配置的方式实现切换，或者使用Spring Kafka提供的KafkaTemplate的setProducerFactory方法来切换Producer的配置。
- 监控和报警：可以使用Spring Boot Actuator来监控Kafka连接状态，并通过集成的监控系统（如Prometheus、Grafana等）进行实时监控和报警。
总结： 在Spring Kafka中检测Kafka连接问题并切换到不同的集群，需要配置多个Kafka集群的连接信息，实现自定义的Kafka连接检测逻辑，并在检测到连接问题时进行切换。同时，可以使用监控系统进行实时监控和报警。以下是腾讯云相关产品和产品介绍链接地址：
- 腾讯云消息队列 CKafka：腾讯云提供的高可用、高可靠、高性能的消息队列服务，适用于大规模数据流转、日志处理、实时计算等场景。详情请参考：https://cloud.tencent.com/product/ckafka
请注意，以上答案仅供参考，具体实现方式可能因实际情况而异，建议根据具体需求和环境进行调整。
相关·内容
我想从一个kafka集群消费消息并发布到另一个kafka集群。想知道如何使用spring-kafka进行配置吗？
浏览 72提问于2021-01-08得票数 2
回答已采纳
我们目前正在尝试创建一个SCS应用程序，该应用程序可以生成特定的主题消息，但是在运行该应用程序时，我们遇到了错误： The following method did not exist:
org.springframework.kafka.core.DefaultKafkaProducerFactory.setBeanName(Ljava/lang/String;)V
The method's class, org.springframework.kafka.core.DefaultKafkaProducerFactory, is available from the f
浏览 109提问于2020-07-22得票数 1
回答已采纳
在远程kafka云集群中，kafka代理将被更新为新版本(5.1)，因此将应用新的kafka协议。
现在我应该更新我的卡夫卡客户，使之能够连接。现在，我在我的spring引导应用程序中使用了以下kafka相关依赖项：
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-parent</artifactId>
<version>1.3.8.RELEASE</version>
<groupId>org.springframewor
浏览 0提问于2019-04-15得票数 4
回答已采纳
1回答
有没有人知道Spring Boot与Kafka Connect的集成？有一个spring-kafka项目，我认为它与Kafka客户端进行了很好的集成，但没有连接和流API。
浏览 2提问于2017-08-09得票数 7
我使用的是spring-kafka，我的@KafkaListener方法有一个用@org.springframework.messaging.handler.annotation.Payload注释的参数，如下所示：
@KafkaListener(topics = "my-topic")
public void consumeTrigger(@Payload InputDTO InputDTO) {
...
}
输入是json格式的，我使用org.springframework.kafka.support.serializer.JsonDeserializer作为值反序列
浏览 18提问于2020-05-08得票数 0
我构造了一个spring boot应用程序来在Kafka中生成数据，但现在我需要将该应用程序连接到旧的Kafka (0.11v)。我尝试只更改bootstrap-server，但收到连接错误。
我再次检查了这个Kafka实例是否工作正常，以及其他应用程序是否正在连接。
我的问题是，像v2.5.4.RELEASE这样的spring-Kafka的新版本是否可以连接到旧版本的Kafka。旧版Kafka实例有什么配置可以添加吗？我构造了一个spring boot应用程序来在Kafka中生成数据，但现在我需要将该应用程序连接到旧的Kafka (0.11v)。我尝试只更改bootstrap-server，
浏览 3提问于2020-08-28得票数 0
我们今天遇到了一个问题，一个Kakfa消费者无法连接到任何代理，日志中没有错误消息。当它通过SSL连接并且指向错误的信任库时，它未能连接。但是，虽然在这种情况下，这是由于SSL失败，但我也曾见过其他情况，即使用者无法连接，日志中没有错误消息。
是否有一种方法来配置卡夫卡使用者，以便在使用者无法连接到代理？时记录错误消息--我正在寻找一个通用解决方案，而不是特定于SSL连接问题的解决方案。
(我们使用的是：org.apache.kafka:kafka-clients:0.9.0.1和org.springframework.kafka:spring-kafka:1.0.4.RELEASE)
浏览 1提问于2018-11-07得票数 3
我有一个Spring Boot (2.3.3)服务，它使用spring-kafka来访问专用的Kafka/Zookeeper配置。我一直在使用服务器设置spring.kafka.bootstrap- application.properties =localhost:9092来访问我的开发/测试Apache Kafka服务。 然而，在生产中，我们在Zookeeper中配置了一个Kafka Brokers集群(在许多服务器上)，并且我被要求修改我的服务来查询Zookeeper，以获得代理列表并使用该列表而不是引导服务器配置。原因是，众所周知，我们的DevOps人员会重新配置服务器/节点和Kaf
浏览 25提问于2020-09-15得票数 1
我正在使用一个启用SSL的kafka集群来连接到使用者并发布消息。下面是技术栈。
spring-kafka : 2.6.6
spring-boot : 2.4.3
卡夫卡性质
kafka:
bootstrap-servers: ${BOOTSTRAP-SERVERS-HOST}
subscription-topic: TEST
properties:
security.protocol: SSL
ssl.truststore.location: ${SUBSCRIPTION_TRUSTSTORE_PATH}
ssl.truststore.password: ${S
浏览 28提问于2022-01-14得票数 0
希望使用homogeneous kafka consumers在来自Spring应用程序的不同集群上创建spring-kafka。
也就是说，希望为已经定义的类创建一个，该对象侦听动态定义的多个集群。
例如:假设是包含template for kafkaconsumer的Spring应用程序S。有三个卡夫卡星系团custer1，cluster2，cluster3。应用程序S充当从每个集群生成的数据的聚合器。这里的解决方案是三个相同模板的consumers将侦听parallel中的单个集群。
上面的场景是否有可能使用spring？
浏览 0提问于2018-10-09得票数 1
我正在尝试reactor-kafka消费消息。其他一切都很好，但我想为失败的消息添加一个retry(2)。默认情况下，spring-kafka已经重试了3次失败的记录，我想用reactor-kafka来实现同样的效果。 我使用spring-kafka作为reactive-kafka的包装器。下面是我的消费者模板： reactiveKafkaConsumerTemplate
.receiveAutoAck()
.map(ConsumerRecord::value)
.flatMap(this::co
浏览 38提问于2021-10-01得票数 3
回答已采纳
我正在使用Spring-Kakfa连接kakfa集群，并且我需要在分配完所有主题的所有主题分区后执行一段代码。以下面的代码为例 @Component
public class MyKafkaMessageListener1 {
@KafkaListener(topics = "topic1" , groupId = "cg1")
public void handleMessage(String message){
System.out.println("Ms
浏览 37提问于2021-08-06得票数 0
2回答
一个简单的spring kafka，它使用的是网络集群上的一个主题：
错误：
无法建立与节点-1 (localhost/127.0.0.1:9092)断开连接的引导代理本地主机:9092 (id：-1齿条: null)。可能找不到经纪人。
谜题：
配置的代理是而不是本地的，它是BROKER_1.FOO.NET:9094，并且是可用的。
pom.xml
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</
浏览 7提问于2021-06-08得票数 1
回答已采纳
在中，它说：
Kafka流应用程序只能与此配置值指定的单个Kafka集群通信。未来版本的Kafka流将支持连接到不同的Kafka集群来读取输入流和写入输出流。
这是否意味着我的整个应用程序只能连接到一个卡夫卡集群，或者每个实例只能连接到一个集群？
我是否可以创建多个KafkaStreams实例，它们具有连接到不同集群的不同属性？
浏览 10提问于2017-08-23得票数 13
有没有在spring-kafka中配置SpecificAvroSerdes (或任何基于模式注册表的serdes - JsonSchema和Protobuf)的示例，允许利用一些自动配置(基于yaml或properties文件)。 在SO like How to use Spring-Kafka to read AVRO message with Confluent Schema registry?中有几个类似的问题 但我想详细介绍一下Kafka Streams Serdes和serdes的声明性配置。 谢谢
浏览 29提问于2020-11-13得票数 1
我使用SpringBoot1.5.6 Kerberos 来使用Kerberos身份验证连接到Kafka 0.11。这些是我为Kafka使用的依赖关系：
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-kafka</artifactId>
<version>3.0.0.RELEASE</ver
浏览 4提问于2019-02-20得票数 3
回答已采纳
1回答
我设计了一个组件来收集应用程序运行时数据，这些数据通过Kafka发送到分析服务器。在大多数情况下，应用程序会集成Kafka。为了避免连接到同一个Kafka两次，我需要确定应用程序是否使用Kafka。如果应用程序使用Kafka，我会直接重用连接。
那么，我如何预测应用程序使用kafka？
如果app集成了spring-kafka，我应该怎么做？
浏览 1提问于2021-05-20得票数 0
1回答
Kafka JDBC connect兼容Spring-Kafka库吗？
我确实关注了，但仍然有一些困惑。
浏览 4提问于2020-07-04得票数 0
我使用spring-kafka来使用SpringBoot1.5.16实现流应用程序。我们使用的spring-kafka版本是1.3.8 version。
我正在搜索一个方法来关闭引导应用程序，以防错误终止所有与withKafka流相关的线程。我发现在中有可能注册一个句柄来处理不明确的异常。方法为。
我看到这个方法是在spring-kafka中以KStreamBuilderFactoryBean类型公开的。
我的问题如下。是否有一种简单的方法将UncaughtExceptionHandler注册为bean并让Spring正确地注入到工厂bean中？或者我应该创建自己的KStreamBuilder
浏览 4提问于2020-01-17得票数 1
回答已采纳
我们有使用Kafka 2.1.1的Kafka集群，我们正在使用Flink内置的Kafka连接器(客户端)，即使用0.10。
我想问一下，Kafka客户端版本0.10是否可以与Kafka服务器版本2.1.1一起工作？
谢谢!
浏览 2提问于2019-10-09得票数 0
回答已采纳
我正在编写一个Kafka消费者，为了学习的目的，这一次我考虑使用Spring-Kafka实现。到目前为止，我一直使用Java API来编写使用者。 我想手动管理偏移量，所以我在Spring-Kafka包中搜索类似于ConsumerRebalanceListener的东西。我的成功之处在于，我在Spring中遇到了可以用来代替ConsumerRebalanceListener的ConsumerAwareRebalanceListener。 但是当我查看ConsumerAwareRebalanceListener接口时，可以看到两个方法- onPartitionsRevokedBeforeCom
浏览 49提问于2019-04-10得票数 0
我已经配置了3个kafka集群，我正在尝试使用spring-kafka。但是当我杀死kafka时，我不能发送其他消息到队列中。 Kafka版本2.0.0 spring-kafka版本2.0.1 kafka-topics.sh --describe -zookeeper=zoo1:2181打印 KAFKA_SWARM_TEST PartitionCount:1 ReplicationFactor:2 Configs:
Topic: KAFKA_SWARM_TEST Partition: 0 Leader: 2 Replicas: 1,2
浏览 21提问于2019-08-26得票数 3
这是我第一次使用卡夫卡。我有一个spring引导应用程序，我正在使用kafka主题的消息，并将消息存储在DB中。我有一个处理DB fail的要求，如果DB关闭，则不应该提交该消息，并在一段时间后暂停消费消息，然后监听器可以再次开始使用消息。做这件事的更好的方法是什么。
i am using spring-kafka:2.2.8.RELEASE which is internally using kafka 2.0.1
浏览 9提问于2020-06-01得票数 1
目前我正在使用Spring boot 2.4.0和spring-kafka。 我想使用消费者和生产者，而不是kafka admin。 我试图通过将"bootstrap.servers“设置为null来覆盖KafkaAdmin类，但不起作用。它仍然“尝试”连接以生成KafkaAdmin并记录错误。这不是致命的，但有没有办法完全禁用kafka管理？
浏览 50提问于2021-01-05得票数 0
回答已采纳
我有spring引导应用程序，它连接到我的kafka集群。应用程序(如kafka客户端)使用SASL身份验证，在初始化kafka生产者和使用者之前，我通过System.setProperty()指定了JAAS配置。它很好地工作在单卡夫卡集群设置。
kafka_client_jaas.conf
KafkaClient {
org.apache.kafka.common.security.plain.PlainLoginModule required
username="myClusterUser"
password="user-secret";
};
浏览 3提问于2020-07-22得票数 2
回答已采纳
我使用的是spring-kafka 2.2.8，并使用以下设置编写了一个简单的异步生成器：
producer config key : acks and value is : 1
producer config key : max.in.flight.requests.per.connection and value is : 5
producer config key : compression.type and value is : lz4
producer config key : linger.ms and value is : 10
producer config key :
浏览 0提问于2020-08-08得票数 0
1回答
我正在尝试让多个不同的都连接到一个Kafka/Zookeeper集群，所有这些都在Kubernetes内部。微服务公司正在使用org.springframework.kafka:spring-kafka -作为事件的消费者和生产者。
所有的服务都连接到kafka，好的，主题是创建的，但是每个服务的使用者是非常不一致的。
例如，当服务一次启动时，所有使用者都将侦听消息并调用该函数。但是，当我重新启动一切(包括卡夫卡和动物园管理员)时，它要么不能工作，要么一些不同服务的消费者会工作等等。
下面是我的一些配置--我没有任何基于Java的配置--就在我的application.yml中，如下所示：
s
浏览 5提问于2020-11-28得票数 1
回答已采纳
1回答
我们试图在AWS上将db2-IIDR (IBM )的数据发布到Kafka。由于以下错误，订阅失败-
An error occurred during the conversation with Kafka.
Error: org.apache.kafka.common.errors.TimeoutException: Batch containing 1 record(s) expired due to timeout while requesting metadata from brokers for cdckafka-subsname-commitstream-0
在aws上发
浏览 1提问于2019-01-25得票数 0
回答已采纳
1回答
我已经看到了一个类似的问题，但没有清楚地回答我的疑虑。
我们有不同的卡夫卡集群和大量的开发运作习惯围绕它。我们有办法启动/停止集群，许多帮助维护集群的利用脚本等等。
现在我们想使用Kafka连接连接器来满足新的需求，但据我所见，Kafka连接非常耦合到汇流-集线器。这就像我甚至不能使用连接器，除非安装一个完全运行的汇流集线器。
这使得我们很难使用卡夫卡连接连接器，我知道汇合式集线器可能是一个帮助运行这些连接器的框架，但是我们甚至不能使用分离的Kafka集群(一个没有被汇合-集线器利用的集群)。
但也许我错过了什么..。
你知道在已经存在的Kafka集群(完全独立于汇合式集线器)上是否有正确使用
浏览 0提问于2019-09-06得票数 0
回答已采纳
我有一个使用Spring-Kafka的Spring Boot应用程序，我使用docker中的landoop's fast-data-dev作为我的kafka集群。我使用kafka模板、avro和confluent的模式注册表生成到一些主题的消息。但是，当我进入快速数据开发ui时，我可以看到我的模式已注册，我的消息仍然显示为Data type: binary，而不是'avro‘，因此它们的值在ui中不可读。我尝试使用kafka- avro -console-producer发送一些消息到不同的主题，这些消息没有这样的问题-它们在ui中是可见的和可读的。然而，当我使用kafka-
浏览 33提问于2019-05-06得票数 0
我在spring boot中使用cron表达式安排了一个作业。当我添加maven spring-kafka依赖项时，我的工作没有启动。这是我的pom.xml中的依赖项： <dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency> 在我的类中，我使用以下注释
浏览 24提问于2020-09-28得票数 0
我希望将多个数据库的更改聚合到一个数据库中，因此我想在每个数据库旁边运行一个Debezium连接器和一个Kafka服务器/broker，并使用Kafka接收器连接器从所有的Kafka中使用，将其写入一个数据库。
问题是，我能否同时使用Kafka接收器连接器的一个实例，使用多个独立的Kafka代理(而不是集群)。
浏览 0提问于2021-08-29得票数 0
回答已采纳
我在分布式模式下设置了kafka连接集群，我希望获得与多个kafka集群的连接，而不仅仅是多个代理。可以使用bootstrap.servers在connect-distributed.properties中设置目标代理。所以，首先，我将broker1设置为kafka-集群-A，如下所示：
bootstrap.servers=broker1:9092
当然，效果很好。
然后，我添加了卡夫卡-集群-B的broker2，如下所示：
bootstrap.servers=broker1:9092,broker2:9092
因此，这两个经纪人是在不同的集群中。但这根本不管用。没有任何错误，它只是被卡住了，
浏览 3提问于2021-05-07得票数 2
回答已采纳
我需要识别数据库连接失败异常与其他异常，并在我的kafka侦听器错误处理程序中进行相应的处理。由于Spring Data JPA中的所有内容都被转换为DataAcccessException，我只是想知道当应用程序运行时发生连接故障时会抛出什么异常。工作环境: Spring boot 2.x、Spring data jpa、spring-kafka、Java 8
浏览 2提问于2019-02-20得票数 1
我们在生产环境中使用Apache kafka 2.0，现在我们计划将kafka版本从2.0升级到2.6 我们在基于三个代理的集群设置中运行 我有以下问题。 1) kafka是否可以从一个版本升级到更高版本？ 2)升级过程中是否有数据丢失？ 3)是否可以在集群运行时执行？ 4)如果出现问题，如何回滚到向下版本？ 你能分享你对这个问题的宝贵想法吗？ 这将有助于设置..
浏览 12提问于2021-04-08得票数 0
我有一个卡夫卡流应用程序运行(0.10.2.1)。当我关闭Kafka集群时，streams应用程序继续等待下一条消息，当集群被恢复时，它将恢复使用消息。在集群关闭期间，应用程序看起来运行良好。我已经测试了45分钟了。
我希望卡夫卡会抛出一个例外或者停止。我已经将StateListener配置为在KafkaStreams关闭时进行日志记录，但是它从未被调用过。
kafkaStreams.setStateListener((newState, _) => {
if (newState == KafkaStreams.State.NOT_RUNNING) {
Log.error(&
浏览 2提问于2017-06-29得票数 6
我有一个spring引导应用程序，用于Apache，用于向Kafka集群上的主题发送消息。我为自动设置了一个属性，在应用程序启动并配置了生产者之后，应用程序日志显示没有设置该属性的值，并给出了警告："WARN [task-scheduler-1] org.apache.kafka.clients.producer.ProducerConfig [] --- The configuration 'delivery.timeout.ms' was supplied but isn't a known config"
我已经检查了我的POM并确认我使用了以下
浏览 1提问于2019-10-03得票数 0
回答已采纳
2回答
我正在使用带有strimzi操作符的Kafka，我创建了一个Kafka集群，并使用yml文件部署了Kafka连接。但在这之后，我完全不知道下一步该做什么。我了解到Kafka connect用于将数据从源复制到Kafka集群，或者从Kafka集群复制到另一个目标。我想使用Kafka connect将文件中的数据复制到Kafka集群的any主题。谁能帮助我，我怎么做，我正在分享的yml文件，我用来创建我的Kafka连接集群。
apiVersion: kafka.strimzi.io/v1beta1
kind: KafkaConnect
metadata:
name: my-connect
浏览 4提问于2020-11-20得票数 2
我尝试了几种方法将burrow应用程序从我的EC2实例连接到我的kafka集群，以获得消费者滞后指标。我可以通过控制台产生和使用来自该实例的消息，但是当我想要连接burrow时，它会在日志中抛出这个错误。
"name":"kafkatestingcluster","error":"kafka: client has run out of available brokers to talk to (Is your cluster reachable?)"
我已经检查了bootstrap服务器两次，还有zookeeper，它们都是正常
浏览 0提问于2019-10-21得票数 0
我是卡夫卡的新手。我正在开发Spring-Kafka POC。我们的KAFKA服务器是Kerberized的。有了所有必需的配置，我们就能够访问Kerberized化的Kafka服务器了。现在我们有了另一个需求，我们必须使用来自非Kerberized(简单Kafka消费者) Kafka服务器的主题。我们可以在单个应用程序中通过创建另一个具有自己的侦听器的KafkaConsumer来做到这一点吗？
浏览 10提问于2017-08-07得票数 1
我在一个spring boot项目中工作，在这个项目中，我使用spring-kafka依赖项消费来自kafka主题的消息。这是我的监听器，它运行良好：
@KafkaListener(topics = "${mytopic.name}", containerFactory = "kafkaEventListenerObjectContainerFactory")
public void listenTopic(ConsumerRecord<String, KafkaEvent> cr,
浏览 16提问于2021-07-26得票数 0
1回答
我有Istiov1.6.4的k8s集群。默认情况下，侧插入是禁用的。我让卡夫卡集群在这个k8s上运行，安装了strimzi操作符。卡夫卡集群的工作没有任何问题时，卡夫卡以及客户端吊舱没有注入Istio代理。我的问题是:当我用kafka客户端和Istio代理注入创建一个吊舱时，我无法连接到Kafka集群。客户端的日志：
java.io.IOException: Connection reset by peer
在服务器端：org.apache.kafka.common.network.InvalidReceiveException: Invalid receive (size = 3692956
浏览 4提问于2020-07-09得票数 7
我的项目是用spring.now实现的，我想使用kafkatemplate。因此，我在pom.xml中添加了下面的依赖项，但导入org.springframework.kafka.core.KafkaTemplate在com.kafka.config包中显示错误，因为“无法解析类型.class。它是从所需的.class文件间接引用的。”
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>sprin
浏览 1提问于2018-12-18得票数 0
我正在开发一个使用spring-kafka的Kafka监听器的应用程序。我面临的问题是，当Kafka listener没有打开(各种原因，比如Kafka服务器没有打开或关闭)时，spring上下文初始化会失败。如何确保我的应用程序是独立的。
有谁能帮帮忙。
浏览 44提问于2017-09-10得票数 4
回答已采纳
我使用的是Axon版本(3.3)，它在SpringBoot主类中使用
@SpringBootApplication(exclude = KafkaAutoConfiguration.class)
在我们的用例中，命令端微服务需要从kafka主题中挑选消息，而不是将其公开为Rest api。它会将事件存储在事件存储中，然后将其移动到另一个kafka主题，供查询端微服务消费。
由于KafkaAutoCOnfiguration被禁用，我不能使用spring-kafka配置来编写消费者。如何在Axon中使用普通消息？
我试着写了一个普通的Kafka spring Consumer，但由于Kafka A
浏览 18提问于2019-07-12得票数 0
我想从Kafka源集群镜像到Kafka目标集群。如果我的源集群和目标集群都在同一个版本(比如0.10.0.0)上，那么一切都可以正常工作，但是当我的源集群在0.8.2.1上，而目标集群在0.10.0.0上时，它就不能工作了。
有谁知道吗？
提前谢谢。
浏览 3提问于2016-06-17得票数 1
1回答
分布式模式下Kafka Connect集群的Offset Management的行为是什么，即运行多个Connector并监听同一组主题(或一个主题)？ 因此，在分布式模式下，Kafka Connect将偏移量信息存储在Kafka中，该偏移量将被集群中的workers读取并提交。如果在该Kafka Connect集群中有多个Connector在监听同一主题，会发生什么情况？分区的偏移量是所有连接器的偏移量都相同，还是每个连接器在分区上的偏移量不同？
浏览 15提问于2019-07-01得票数 2
我在使用kafka-streams和spring-kafka应用程序时遇到了同样的问题。第一个是使用kafka-clients:1.0.0库，另一个是版本1.0.2
kubernetes (KAFKA_ADVERTISED_LISTENERS="PLAINTEXT://${POD_IP}:9092").中只有一个代理实例在运行它是有状态的设置，它是通过无头服务内部端点从客户端应用程序访问的(尽管我尝试过集群ip，问题也是一样的)
一旦我删除了这个kafka pod并重新创建它，我的客户端应用程序就无法重新连接。Pod确实是用另一个ip地址重新创建的，但是因为我是通过服务内部端点
浏览 0提问于2018-10-02得票数 3
这是关于升级生产中的现有代码库，这些代码库使用来自kafka-clients，kafka-streams，spring-kafka 2.4.0到2.6.x的窗口，并将spring-boot-starter-parent2.2.2.RELEASE升级到2.3.x，因为2.2与kafka-streams 2.6不兼容。
现有的代码在下面的旧版本中提到了这些bean(2.4.0，2.2春季版本)：
@Bean("DataCompressionCustomTopology")
public Topology customTopology(@Qualifier("CustomFa
浏览 20提问于2021-05-15得票数 2
我目前正在使用Google Cloud Dataflow和Apache Beam来消费来自两个不同Kafka集群中的Kafka主题的消息，这两个集群包含相同的主题名称，但主题中的数据不同。Kafka集群是分开的，因为它们包含来自不同区域的数据。
我只是想知道，是否可以通过在单个KafkaIO.read数据流管道步骤中列出两个群集的所有引导服务器来使用这两个群集的数据？
.withBootstrapServers("CLUSTER1_SERVER:PORT,CLUSTER2_SERVER:PORT");
我正在阅读关于Kafka bootstrap服务器的文档，我不清楚在连接到b
浏览 23提问于2020-05-13得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
实时音视频
即时通信 IM活动推荐
腾讯云开发者
