阅读来自KAFKA Consumer的消息时使用org.springframework.messaging.converter.MessageConversionException
org.springframework.messaging.converter.MessageConversionException是Spring框架中的一个异常类，用于表示消息转换过程中的异常情况。在阅读来自KAFKA Consumer的消息时，如果发生了消息转换异常，就会抛出该异常。
该异常的主要作用是提供了一个统一的异常类型，用于处理消息转换过程中可能出现的各种异常情况，例如消息格式不正确、消息类型不匹配等。通过捕获该异常，开发人员可以针对不同的异常情况进行相应的处理，保证系统的稳定性和可靠性。
在处理该异常时，可以采取以下几种方式：
- 检查消息格式：首先，可以检查消息的格式是否符合预期。如果消息格式不正确，可以尝试修复消息格式或者丢弃该消息。
- 检查消息类型：其次，可以检查消息的类型是否与预期的类型匹配。如果消息类型不匹配，可以尝试进行类型转换或者丢弃该消息。
- 异常处理：最后，可以根据具体的业务需求，选择合适的异常处理策略。例如，可以记录日志、发送告警通知、进行重试等。
推荐的腾讯云相关产品和产品介绍链接地址：
- 腾讯云消息队列 CMQ：腾讯云提供的消息队列服务，支持高可靠、高并发的消息传递。链接地址：https://cloud.tencent.com/product/cmq
- 腾讯云云原生数据库 TDSQL-C：腾讯云提供的云原生数据库，支持分布式、高可用的数据库服务。链接地址：https://cloud.tencent.com/product/tdsqlc
- 腾讯云云服务器 CVM：腾讯云提供的云服务器服务，支持弹性扩展、高性能的计算资源。链接地址：https://cloud.tencent.com/product/cvm
请注意，以上推荐的腾讯云产品仅供参考，具体选择应根据实际需求进行评估和决策。
相关·内容
我正在发送一个avro对象用户并在侦听器中接收它。这里有一些秘密
@Bean
public ProducerFactory<String, User> userProducerFactory() {
Map<String, Object> config = new HashMap<>();
config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
config.put(ProducerConfig.KEY_SERIALIZER_CLA
浏览 4提问于2020-03-25得票数 2
回答已采纳
2回答
我遇到了一些问题，试图使用Spring将Kafka消息反序列化为POJO。我希望使用消息的键和值部分来构造POJO。
Kafka消息键是一个字符串，消息值是JSON。
我尝试通过在和上学习教程来完成消息的值部分。除了我还希望在POJO中有键值之外，java应用程序也不会生成消息。
如何使java应用程序适当地将kafka消息反序列化为POJO？
例如：
key = "test"
{
"value1": "1st value"
"value2": "2nd value"
}
我正在尝试的一个可复制的例子可以
浏览 3提问于2020-03-05得票数 2
回答已采纳
我尝试了官方的示例(消费者/生产者)与我本地安装的confluent 4.0.0，它可以在发送post请求和在侦听器中接收时发送“传感器”avro消息，但当我使用confluent 4.0.0附带的kafka- avro -console-consumer工具查看发送的avro消息时，该工具引发了以下错误(a)。我也尝试使用kafka- avro -console-producer工具发送的avro消息(该消息可以在kafka-avro-console-consumer工具中正确显示)，它报告了以下错误(b)。给定的示例不支持confluent 4.0.0吗？非常感谢！
(a) kafka-a
浏览 3提问于2018-08-26得票数 1
我使用spring 2.3.2.RELEASE与spring -> 2.5.4 clients -> 2.5.0
我有以下简单的侦听器
@Slf4j
@Component
class SampleKafkaListener {
private final ObjectMapper objectMapper;
@Autowired
SampleKafkaListener(ObjectMapper objectMapper) {
this.objectMapper = objectMapper;
}
@KafkaListene
浏览 9提问于2020-08-25得票数 0
1回答
我试图使用Spring与合流模式注册表和Kafka反序列化器。然而，在“卡夫卡春书”中却没有这样的东西。
以下是我为我的春季卡夫卡消费者做的事情：
@Configuration
@EnableKafka
public class KafkaConfiguration {
@Bean
ConcurrentKafkaListenerContainerFactory<String, Customer> kafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<String,
浏览 0提问于2019-07-08得票数 2
我正在尝试使用Kafka-Spring (spring-boot)实现一个Kafka消费者。我使用生产者发送了一条消息，但是Spring抛出了一个异常，并且由于这个异常，我的侦听器方法没有被调用。
下面是我的监听器方法：
@Service
public class KafkaExchangeApi {
@KafkaListener(topics = ORDER_TOPIC, groupId = "group_id")
public void handleOrderResponse(@Payload GatewayOrder response) {
浏览 0提问于2020-09-10得票数 1
当用spring-kafa启动Spring 2.4.5时，value.deserializer值显示为StringDeserializer而不是KafkaAvroDeserializer
2021-05-12 13:46:05.313 INFO 12632 --- [ main] o.a.k.clients.consumer.ConsumerConfig : ConsumerConfig values:
... Elided for brevity ...
key.deserializer = class org.apache.kafka.common.se
浏览 18提问于2021-05-12得票数 2
回答已采纳
我有一个Kafka消费者应用程序，用Spring 2.0.2编写。当我在侦听器中收到消息时，会收到以下错误：
Caused by: org.springframework.messaging.converter.MessageConversionException: failed to resolve class name. Class not found [com.test.demo.domain.Account]; nested exception is java.lang.ClassNotFoundException: com.test.demo.domain.Account
生产者中对
浏览 0提问于2018-05-23得票数 7
回答已采纳
因此，我使用spring-boot2.1.6并集成kafka consumer来使用在该主题上发布的任何类型的消息。作为参考，我正在关注
所以我在我的pom中有依赖：
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
我在application.yml中进行配置
spring:
kafka:
bo
浏览 0提问于2020-05-23得票数 0
我正在开发Spring和Apache --尝试使用用户定义的配置-
org.springframework.kafka.listener.ListenerExecutionFailedException: Listener method could not be invoked with the incoming message
Endpoint handler details:
Method [public void com.example.demo.Consumer.consume(com.example.demo.User) throws java.io.IOException]
Bea
浏览 10提问于2021-02-19得票数 2
1回答
卡夫卡不承认作为一个论据
、、、、
我们正试图在java spring项目中实现Kafka确认。在没有确认的情况下，我们成功地接收和读取了消息，但是当我们在方法中添加应答时，我们收到了这个错误：
org.springframework.kafka.listener.ListenerExecutionFailedException: invokeHandler Failed; nested exception is java.lang.IllegalStateException:
No Acknowledgment available as an argument, the listener container must hav
浏览 2提问于2021-01-14得票数 0
1回答
我有一个带有apache kafka的SpringBoot项目(一个开源的流处理软件)，我有这个侦听器。
@KafkaListener(topics = "test")
public String consume(Hostel hostel) throws IOException {
}
这个序列化器
public class HostelSerializer implements Deserializer<Hostel> {
private final ObjectMapper objectMapper;
public InputRequest
浏览 3提问于2020-04-09得票数 1
回答已采纳
我有一个Springboot应用程序。使用此配置：
kafka:
consumer:
listener:
ack-mode: MANUAL_IMMEDIATE
广告给这个消费者：
@KafkaListener(topics = "test", groupId = "group_id")
public void consume(String message, Acknowledgment ack) throws IOException {
ack.acknowledge();
}
但是
浏览 6提问于2020-03-30得票数 3
回答已采纳
我正在编写一个使用Avro消息的spring云流接收器应用程序。我试图让应用程序使用这些消息，但是我得到了以下错误。
org.springframework.messaging.converter.MessageConversionException: Cannot convert from [[B] to [org.apache.avro.generic.GenericRecord] for GenericMessage [payload=byte[304], headers={kafka_offset=42898134, kafka_consumer=org.apache.kafka.cl
浏览 1提问于2018-09-19得票数 1
1回答
我用了卡夫卡，我想要跑投法
卡夫卡是带动物园管理员的码头集装箱，我有一个题目叫updateRate。
这是方法是放方法运行。
@PutMapping("v1/branches/{branchId}/rateUpdate")
public ResponseEntity<Void> rateBranchUpdate(@RequestParam("rate") Double rate,@PathVariable Long branchId) throws JsonProcessingException {
Map<String,String&
浏览 3提问于2021-03-09得票数 0
回答已采纳
为了表明我的意思我做了一个简单的项目。
依赖关系：
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.kafka&l
浏览 6提问于2020-11-01得票数 1
回答已采纳
1回答
所有人。
package kitchen;
import entity.Order;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.spr
浏览 0提问于2020-12-17得票数 0
回答已采纳
当我将org.springframework.kafka.listener.ListenerExecutionFailedException:设置为false并尝试使用基于注释的enable.auto.commit -kafka @KafkaListener手动提交偏移时，无法使用传入的消息调用enable.auto.commit侦听器方法。
我有一个非常简单的代码如下：
@KafkaListener(id = "someid", topics = "${demo.topic}", containerFactory = "someContainerFac
浏览 5提问于2017-06-27得票数 11
回答已采纳
1回答
我们使用，我们的应用程序使用一个主题并处理消息，然后将它们输出到不同的主题。
这些主题也在同一个应用程序中使用，并输出到最后一个主题。
我们注意到，每当第一个使用者使用新消息而这些线程仍然处于活动状态时，就会创建大量的生产者线程。
以下是我的简化配置：
cloud:
stream:
function:
definition: schedulingConsumer;consumerSearch1;consumerSearch2
default:
group: ${kafka.group}
contentType:
浏览 3提问于2022-06-27得票数 1
回答已采纳
1回答
我在使用卡夫卡时遇到了一些问题。任何帮助都是非常感谢的！我有动物园管理员和卡夫卡集群3个节点在码头群。您可以看到下面的Kafka broker配置。
KAFKA_DEFAULT_REPLICATION_FACTOR: 3
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3
KAFKA_MIN_INSYNC_REPLICAS: 2
KAFKA_NUM_PARTITIONS: 8
KAFKA_REPLICA_SOCKET_TIMEOUT_MS: 30000
KAFKA_REQUEST_TIMEOUT_MS: 30000
KAFKA_COM
浏览 0提问于2019-02-18得票数 1
3回答
我有一个Spring应用程序，它听了Spring的一个Kafka主题。在使用消息之后，将使用消息中的信息执行几次web/rest服务调用，以收集其他一些数据，这一过程需要一些时间才能完成。因此，我使用一个大小为20的线程池来创建并行消息处理。
这个系统通常运行良好，但很少有大量的消息(大约200 K)在短时间内(1秒)向Kafka主题发送/生成消息。在这种情况下，消费者会立即使用消息，但是消息处理机制还不够快。因此，当等待线程和应用程序获得OutOfMemoryError时，所有消耗的消息都保留在内存中。
将线程池大小增加到某个点可能是一种改进，但它不是解决此问题的永久解决方案。我希望在一段时
浏览 4提问于2021-01-02得票数 0
from kafka import KafkaConsumer,KafkaProducer
consumer = KafkaConsumer(bootstrap_servers=['IP:9092'],auto_offset_reset='earliest',
consumer_timeout_ms=3000,enable_auto_commit=True,
group_id='group39assignment')
浏览 2提问于2022-08-15得票数 0
我试图从同一个Kafka主题反序列化不同的JSON有效负载。这里提出的其他问题，引导我进行了第一次尝试，但我无法使它运行。
正如加里提到的()，有一些提示(JsonSerializer.ADD_TYPE_INFO_HEADERS)，但是当我发送和接收这两条消息时，我会得到一个异常。
org.springframework.kafka.listener.ListenerExecutionFailedException: Listener method could not be invoked with the incoming message
Endpoint handler details:
浏览 0提问于2019-01-24得票数 5
回答已采纳
1回答
弹簧靴卡夫卡手册失败
、、、
我有一个Spring消费者的下面的配置。我在尝试手动确认而不是自动提交。在手动确认后，我开始出错。Spring版本为2.7.2。
kafka.consumer.groupId=mcs-ccp-event
message.topic.name=mcs_ccp_test
kafka.bootstrapAddress=kafka-dev-app-a1.com:9092
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.listener.ack-mode=MANUAL_IMMEDIATE
和这种使用者配置
@EnableKafk
浏览 26提问于2022-08-06得票数 0
回答已采纳
1回答
应用程序A写入用户对象(Json)下面的Kafka主题：
public class UserEvent {
private UUID id;
private Object payload; // contains User fields name etc.
}
应用程序B试图使用下面的User.java(多个绑定程序)使用这个用户对象(User.java驻留在应用程序B项目中)：
binders:
azureEventHub:
type: kafka
environment:
spring:
浏览 5提问于2022-05-25得票数 0
1回答
我在活动上用“kafka -node”创建了kafka消费者。
consumer.on('message' ()=>{
connecting to mongodb and inserting to a collection.
})
用于创建与mongo的连接并返回对象的mongo.js文件
const MongoClient = require('mongodb').MongoClient, assert = require('assert');
const url = 'mongodb://root:****@ds031257.
浏览 0提问于2019-04-03得票数 0
回答已采纳
我有一个int-kafka:outbound-channel-adapter，用于向kafka发送消息，然后使用int-kafka:inbound-channel-adapter接收消息。沟通似乎很好，我可以发送和接收信息，但格式有点奇怪。我分别向出站适配器发送单独的消息，但是当我收到消息时，我会得到一条消息，并将所有消息聚合到该消息的有效负载中。
当我收到消息时，这就是消息有效负载的样子。
[ headers={id=3934de02-1f42-ab90-6aa5-9c15f3cd0b6e，timestamp=1439260669762} [payload={mytopic={0=stri
浏览 1提问于2015-08-11得票数 1
回答已采纳
使用Spring云流的Kafka绑定的消息生成器
@Component
public static class PageViewEventSource implements ApplicationRunner {
private final MessageChannel pageViewsOut;
private final Log log = LogFactory.getLog(getClass());
public PageViewEventSource(AnalyticsBinding binding) {
this.pageViewsOut = binding.pageVi
浏览 2提问于2019-03-19得票数 3
回答已采纳
我有一个Streaming应用程序，它从Kafka主题中读取数据并将其放在HDFS上。我希望应用程序存储抵消读到__consumer_offsets主题的消息，以便在应用程序失败时开始阅读。这个应用程序运行良好(我可以看到HDFS上的数据)，但我看不到它对__consumer_offsets的提交。
这是我的KafkaParams：
val kafkaParams = Map(
"metadata.broker.list" -> "xx.xxx.x.xx:6667",
"enable.auto.commit" -&g
浏览 0提问于2018-11-08得票数 0
使用该配置
使用spring-integration-kafka扩展和以下配置：
<int-kafka:zookeeper-connect id="zookeeperConnect"
zk-connect="#{kafkaConfig['zooKeeperUrl']}" zk-connection-timeout="10000"
zk-session-timeout="10000" zk-sync-time="2000" />
<int-kafka:consume
浏览 0提问于2016-01-29得票数 0
我正面临一个奇怪的问题，我的镜像生成器能够镜像目标集群中的主题名称，但当我尝试使用来自它的消息时，我看不到任何消息。下面是我的配置文件-
producer.config
bootstrap.servers=dest_broker_ip:9092,dest_broker_ip:9092
acks=1
batch.size=100
max.in.flight.requests.per.connection=1
retries=1000000
client.id=mirror_maker_producer
consumer.config
bootstrap.servers=source_broker
浏览 0提问于2018-05-22得票数 1
1回答
目的是消费卡夫卡的记录，并重定向到Azure服务巴士。可以配置kafka消费者，并在调试时确认其在每次轮询kafka主题时消耗数千条消息。
另一方面，Azure服务总线将始终发布一条消息。因此，当集成运行时，kafka日志将显示接收到数千条消息，然后Azure服务总线日志将迭代每个消息，每次发送一个消息到队列。这需要几分钟的迭代，大大减缓了进程的速度。
组件写到，批量发送的设置是默认的，但是对于如何实现这个设置还不太清楚。
public class SampleKafkaConsumer extends RouteBuilder {
@Override
public void conf
浏览 11提问于2022-06-26得票数 0
我正在运行Python3.6和python-kafka 1.3.5
我有一个kafka主题，分区0-15，它们目前都有消息在里面。我希望我的使用者从所有分区中读取和提取消息。这就是我目前正在尝试做的事情
import kafka
consumer = kafka.KafkaConsumer(bootstrap_servers='broker1:9092,broker2:9092,broker3:9092', sasl_mechanism='PLAIN', enable_auto_commit=False, group_id="my_test_group&
浏览 3提问于2021-02-04得票数 0
我正在尝试使用spring-integration-kafka抽象来实现Kafka consumer。有必要同步处理消息。
org.springframework.integration.kafka.listener.AcknowledgingMessageListener.java是否同步处理消息？
浏览 0提问于2015-10-23得票数 1
我以前是一个学习卡夫卡的传统ActiveMQ用户。我有个问题。
使用Active MQ，您可以这样做：
将100条消息提交到队列中，等待多久，就会消耗该队列中的100条消息。保证消息的单个使用者.
我试着用卡夫卡做同样的事
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import ja
浏览 1提问于2022-07-18得票数 0
回答已采纳
在使用消息时，我使用kafka-consumer-groups --describe --group test --bootstrap-server device2:9092来显示最新的CURRENT-OFFSET (大约10条消息/秒)。
输出如下所示：
TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
comment 0
浏览 2提问于2020-02-13得票数 1
回答已采纳
我在1.1.0版中有一个Kafka控制台消费者，我用它从kafka获得消息。当我使用kafka-控制台--带有选项的消费. it脚本--最大的消息时，它似乎是在背诵错误的偏移。
我创建了一个主题和一个消费小组，并阅读了一些信息：
/kafka_2.11-1.1.0/bin/kafka-consumer-groups.sh --bootstrap-server 192.168.1.23:9092 --describe --group my-consumer-group
TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG
浏览 1提问于2020-02-28得票数 2
回答已采纳
在RHEL v6.9上使用Kafka v2.1.0
当一位卡夫卡经纪人倒闭时，消费者无法接收数据。
执行的步骤：
1. Start zookeeper
2. Start Kafka-Server0 (localhost:9092, kafkalogs1)
3. Start Kafka-Server1 (localhost:9094, kafkalog2)
4. Create topic "test1", num of partitions = 1, replication factor = 2
5. Run producer for topic "test1
浏览 1提问于2018-12-13得票数 1
回答已采纳
1回答
我正在使用多个使用Kafka进行异步通信的应用程序。这些应用程序由多个部门管理，合同测试适用于确保通信期间使用的消息遵循预期的模式，并将根据合同规范进行演变。
听起来python的pact库非常合适，因为它有助于为HTTP和消息集成创建契约测试。
我想要做的是发送一个HTTP请求，然后立即收听适当的、专门的Kafka主题。但是似乎测试迫使我指定HTTP代码，即使我期望的是没有HTTP状态代码的队列中的消息。此外，HTTP请求似乎是在使用者监听之前发送的。下面是一些示例代码。
from pact.consumer import Consumer as p_Consumer
from pact.p
浏览 5提问于2021-11-26得票数 0
我有一个@StreamListener，它接受@Payload字符串。为了测试这个侦听器类，我使用嵌入式Kafka编写了一个Junit类。我在运行测试类时出错
错误
org.springframework.messaging.converter.MessageConversionException:不能从[B到java.lang.String for GenericMessage )转换
如果我将@Payload的dataType从String更改为byte[]，则侦听器类将选择消息。
谁能帮我知道这是什么问题吗？我想这是云流配置的一部分。
@ExtendWith(SpringExten
浏览 16提问于2019-09-27得票数 1
我对阿帕奇卡夫卡很陌生。我读了很多书，我认为理解了分区和使用者组机制。但我有个问题。
我有两个春季启动项目。它们都实现了apache用户。对于这两个项目，kafka参数都在每个项目的application.properties文件中定义。以下是共同参数：
spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.consumer.auto-offset-reset=latest
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.St
浏览 5提问于2022-10-22得票数 0
要求:期望接收/处理inMessageHandler中的inMessage和emailMessageHandler中的emailMessage。
问题：(如果消费者配置的消费者组id不同)消息一直流动，直到消费者服务处理程序的transform方法在这两种情况下都是如此，但在emailMessageHandler中的emailMessage的情况下却没有得到它。(但在inMessageHandler中是针对inMessage的)。
问题：(如果消费者配置的消费者组id相同)消息从consumer inMessageHandler的transform方法流向emailMessageHandler
浏览 4提问于2015-09-21得票数 1
尝试使用Kafka绑定从云流读取CloudEvents的批处理消息。如果我在自定义序列化程序/反序列化器中使用任何自定义类，那么它可以正常工作，但是对于cloudevents，消息不会出现。
spring
cloud:
function.definition: consumer
stream:
bindings:
producer-out-0:
destination: audit
group: audit-producer
producer:
useNative
浏览 14提问于2022-10-26得票数 0
回答已采纳
1回答
有时，kafka-node消费者从偏移量0开始消费，而它的默认行为是只消费较新的消息。则它将不会切换回其默认行为。你知道如何解决这个问题吗?知道发生了什么，它的行为突然改变了吗？代码非常简单，而且不需要修改代码就可以实现。
var kafka = require("kafka-node") ;
Consumer = kafka.Consumer;
client = new kafka.KafkaClient();
consumer = new Consumer(client, [{ topic: "Topic_23", partition: 0}
浏览 5提问于2020-01-31得票数 3
我正试图返回一个数组。
我可以将消息数组打印到控制台，并且可以看到它正在被填充。但是，在“最后”之后的代码似乎是无法到达的。我做错了什么？
def kafka_messages(topic, partition):
messages = []
try:
consumer = SimpleConsumer(kafka, b"consumer-group"
, bytes(topic, "UTF-8")
浏览 2提问于2014-11-03得票数 0
回答已采纳
我正在尝试处理失败，并将消息写入failure-Q，以防消息在db期间由于db停机或任何其他问题而失败。
我们需要每天重试一次failure-Q的消息。因为我们使用属性文件来配置消费者配置：
# KAFKA
spring.kafka.bootstrap-servers=localhost:9092
## CONSUMER
kafka.topic.consumer.boot: tink_boot.j
kafka.consumer.group-id: tink_boot.j-001
### Kafka from 5000 to 60000
spring.kafka.consumer.auto-c
浏览 6提问于2020-03-27得票数 0
1回答
没有收到卡夫卡主题的信息
、、、、
在这个程序中调用poll()时，我没有收到任何消息，但是当我从cmd运行kafka-console-cmd. but时，我不知道到底出了什么问题。
执行从main.py开始
from queue import Queue
from concurrent.futures import ThreadPoolExecutor
import time
import json
from kafka_message_consumer import KafkaMessageConsumer
from kafka_discovery_executor import KafkaDiscoveryExecutor
浏览 1提问于2021-10-24得票数 2
回答已采纳
我查看了这些文档，发现如下：
spring.kafka.producer.properties.spring.json.type.mapping=cat:com.mycat.Cat,hat:com.myhat.Hat
我的制片人的财产：
spring.kafka.producer.bootstrap-server=localhost:9092
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-seriali
浏览 27提问于2022-09-16得票数 0
回答已采纳
我正在尝试编写一个简单的Kafka消费者来读取windows机器上的数据。但是我的消费者不能读取任何数据。生产者产生了20多条消息，但没有一条消息被消费。
下面是我为消费者编写的代码：
Properties properties= new Properties();
properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");
properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG
浏览 0提问于2020-05-13得票数 1
我在Ubuntu服务器上安装了最新的confluent platform Kafka。现在，当我尝试从topic获取消费时，上周五正在工作的所有主题以及我创建的任何新主题都会出现错误。
kafka-topics --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic distance2
topic created distance2
kafka-console-consumer --bootstrap-server localhost:9092 --topic di
浏览 1提问于2020-06-01得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
实时音视频
即时通信 IM活动推荐
腾讯云开发者
