msg = kafka_consumer.poll(timeout=2.0)
如何通过Python confluent-kafka api使用
浏览 2提问于2016-07-07得票数 0
我有一个springboot应用程序，监听卡夫卡消息，并将它们转换为对象 @KafkaListener(topics = "test", groupId = "group_id")
public void
浏览 25提问于2020-03-30得票数 1
回答已采纳
当我使用使用反序列化器的使用者类使用一个主题时，我会遇到一个错误。定制的Serde类有序列化程序和反序列化器。但是，我试图理解为什么使用自定义serde来反序列化，然后使用者API就会产生错误，因为它不是org.apache.kafka.common.serialization.Deserializer的一个实例可以使用KTable与Serdes.Integer()序列化<
浏览 0提问于2019-05-24得票数 4
回答已采纳
1回答
用于业务逻辑类的Apache
、、、、
我正在使用Apache和Kakfa，在这里我能够正确地使用来自主题的消息并反序列化它。我正在使用flink datastream Map函数来处理收到的kafka消息，但是flink期望我的所有业务对象都是可序列化的。下面是伪代码KafkaConsumer<KafkaT
浏览 2提问于2022-01-30得票数 0
我有一个SchemaRegistry和一个KafkaBroker，我使用Avrov1.8.1从它们中提取数据。对于反序列化，我一直在使用Confluent的。现在，我打算重构我的代码，以便使用Alpakka提供的，但不幸的是，这破坏了反序列化，因为它导致了NullPointerExceptions：
线程“主”org.apache.kafka.common.errors.SerializationException:错误中的异常，在
浏览 1提问于2019-05-31得票数 1
回答已采纳
3回答
我一直在想，在使用静态类型化语言(如Java )使用Kafka主题的消息时，是否需要使用Avro模式注册表。我正在使用来自Kafka主题设置的信息，如下所示： props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG<byte[], FooClass> kafkaConsumer</em
浏览 5提问于2020-04-20得票数 1
回答已采纳
我正在尝试使用来自kafka主题的消息，其中json的内容没有序列化。这个主题是用实际的JSON生成的，没有序列化，如下所示。case_id": "500g000000Tw5ggAAB", "state_of_origin": "CA"现在，当我尝试使用下面的配置和侦听器代码使用来自主题的消息</
浏览 13提问于2022-08-08得票数 0
2回答
我是Python的新手，从Kafka开始。我有一个需要发送和使用json消息的需求。为此，我使用kafka-python与卡夫卡进行交流。#Producer.pyimport jsonconsumer = KafkaConsumer(bootstr
浏览 30提问于2019-04-05得票数 3
回答已采纳
2回答
我使用wurstmeister/kafka码头形象和Apache消费者。当我向某个主题发送“意外”消息时，我感到很不爽。在发送预期的消息时，我没有这个问题(不要在反序列化时抛出异常)。以下是我对kafka的docker-compose配置： image: wurstmeister/kafka - zookeeper@Ove
浏览 4提问于2017-09-25得票数 0
回答已采纳
1回答
我有一个JAVA springboot项目，在kafka中推送avro消息。在我的开发过程中，使用windows PC，我所有的kafka消息都以所有特殊字符(é，à，...)显示。使用以下序列化程序：
key-serializer: "org.apache.kafka.common.serialization.StringSerializer"
浏览 41提问于2021-05-19得票数 0
回答已采纳
我创建了一个Spring引导应用程序，它将消息发送到Kafka队列(使用spring集成kafka支持)。我想发送任意的json序列化对象。有没有办法在我的spring boot应用程序中获得/注入json-de-/序列化程序？或者如何即席反/序列化一个对象？
应用序列化有哪些好的实践？
浏览 1提问于2018-01-29得票数 0
1回答
为了更好地了解平台，我从几天以来一直在玩卡夫卡的合流版本。对于发送到一个主题的一些格式错误的avro消息，我得到了一些序列化异常。>意图:非常简单，生产者正在发送Avro记录，并且使用者应该使用所有记录而不存在任何问题(它可以将所有与模式不兼容的消息保留在模式注册表中(我们在模式注册表中启用了一个选项，这样您就可以向topi
浏览 2提问于2017-05-12得票数 5
回答已采纳
我正在建立Kafka( kafka _2.11-0.10.0.0)和OSB 12c(12.2.1.2)之间的连接，用于消费来自kafka的OSB消息。我遵循了在OSB中安装Kafka传输所需的步骤，这是成功的，但当我尝试使用Kafka的消息时，我得到的错误如下：
com.bea.wli.sb.transports.TransportException(KafkaConsumer.jav
浏览 0提问于2019-07-08得票数 0
我已经使用以下序列化程序序列化了avro数据。，我使用了以下反序列化器。<init>(KafkaConsumer.java:713) ~[kafka-clients-2.7.0.jar:na]
at org.apache.kafka.clients.consumer.KafkaConsumer<String, GenericRecord> kafkaConsumer = new <em
浏览 4提问于2021-03-12得票数 0
回答已采纳
我正在将来自卡夫卡主题的消息存储在KeyValueStore中，以便以后查询它们。StreamListener public void process(@Input("input") KTable<String,MyMessage> myMessages) {更新的反序列化程序包
spring.cloud.stream.kafka.streams.bindings.input
浏览 1提问于2019-07-03得票数 1
回答已采纳
1回答
我遵循confluent link来发布关于mysql表更改的kafka消息。当我试图从一个springboot应用程序中使用这个消息时，我得到了下面的异常。我怎样才能解决这个问题，这样我就可以阅读消息了。有时我能够读懂消息，但如果像consumed message is key: null value : 2foo��䶰Z��䶰Z一样，我会得到一个序列化的版本 Caused by: java.lang.ClassCastException(
浏览 22提问于2019-03-16得票数 0
我可以通过KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)创建一个KafkaConsumer。有没有办法从ListConsumerGroupsResult获取KafkaConsumer？通过我的IDE和在线搜索，我没有找到这样的方法，但我想知道社区中是否有人这样做过。
浏览 4提问于2018-10-18得票数 1
1回答
我有一个Kafka使用者从生产者那里读取数据，但有时在从表中删除记录时，使用者会获取空值，从而导致值序列化程序中的NoneType错误消息。注意，这是使用postgresql数据库中的Kafka流。consumer = KafkaConsumer( bootstrap_servers='localhost:9092',
浏览 4提问于2022-10-12得票数 0
回答已采纳
从IIBv10.0.0.7开始，我可以使用KafkaConsumer节点来接收发布在Kafka主题上的消息。我需要一些客户端，这将能够接收来自Kafka的消息，并将其放在IBM MQ中，并从IBM MQ获得消息，并发布到Kafka主题中。我已经有IIB和IBM MQ了。Kafka是集成系统之一的消息系统。是否可以使用KafkaConsumer节点
浏览 0提问于2017-10-01得票数 1
我尝试使用以下命令在kafka consumer中读取消息：这里我们可以读取大约4天的旧消息，因为我们在kafka服务器配置文件中设置了保留时间为7天。但是，当我们尝试使用kaka-python客户端库的Kaf
浏览 2提问于2015-09-01得票数 0
云服务器
ICP备案
腾讯会议
云直播
对象存储
腾讯云开发者
