1回答
在中尝试使用时间戳时，我发现了一些奇怪的东西。在从生产者发送了几条消息之后，我运行了kafka-console-consumer.sh并验证了这些消息是否在主题中。我叫住制片人，等了一分钟。当我重新运行kafka-console-consumer.sh时，它没有显示我之前生成的消息。我还添加了producer.flush()和producer.close()，但结果仍然相同。现在，当我停止使用时间戳字段时，一切都正常工作，这让我相信使用时间戳的
浏览 19提问于2019-11-23得票数 1
我正在阅读来自kafka主题的消息，它有3个副本和70个分区，使用的是storm-kafka api，我使用的kafka喷口有15个执行器，因为我没有70个执行器，所以有可能丢失消息吗？我如何确保卡夫卡中的所有信息都被消费者接收？
谢谢
浏览 1提问于2017-05-29得票数 0
回答已采纳
在我的spring boot应用程序中，我正在读取来自kafka主题的消息，并将消息保存到HBase中。 如果数据库关闭，主题中的消息被消费，我应该如何确保消息不会丢失。
浏览 10提问于2020-01-02得票数 0
我有一个用例，我想在kafka之上实现同步请求/响应。例如，当用户发送HTTP请求时，我希望在特定的kafka输入主题上生成一条消息，该消息触发数据流，最终生成输出主题上的响应。工作流程是: HTTP请求->在输入主题上生成消息-> (从输入主题消费消息->应用程序逻辑->生成输出主题消息) ->消费<em
浏览 4提问于2022-11-15得票数 0
我们正在使用低级处理器API开发一个Kafka Streams应用程序。 根据Kafka上的文档，所有的线程和并行性都是由Stream线程和流任务处理的。使用主题上的分区，并行性也是可扩展的。这意味着使用Kafka Streams API主要用于消费主题中的数据，而不是用于实际处理。实际处理将在Kafka流线程的初始数据消耗之后调用的新线程中进行。拓扑中的
浏览 15提问于2019-05-29得票数 2
2回答
在我们的场景中，我们有一组微服务，它们通过发送事件消息与其他服务交互。在高峰时刻，我们预计每天会收到数百万条信息。每条消息都针对一个或多个侦听器类型。我们的要求如下：
将每条消息发送到JMS主队列，然后该队列的侦听器将消息根据消息内容路由到特定队列
浏览 1提问于2018-12-31得票数 0
回答已采纳
我正在使用KakfkaStreams (2.3.0)来使用流和Ktable (压缩主题)之间的连接来丰富价值。问题是压缩后的主题被写入到使用流主题的不同分区器中，因此联接无法按预期工作(一些键不匹配，因为它们位于不同的分区中)。
代理版本为0.10.2。我开始研究如何重新划分压缩的主题，方法是使用一个groupBy()，后面跟着一个()，但是当它开始读取创建的重新分区主题时，它开始抛出一个带有<em
浏览 1提问于2019-08-28得票数 0
回答已采纳
1回答
我将为消息实现一个断路器模式。基本要求是，如果微服务无法将消息发布到发布主题，则应停止接受来自其他Kafka主题的消息。当发布主题可用时，微服务应该开始接受来自其他Kafka主题的消息。有没有办法在Spring boot Kafka Streams中实现这一点？
浏览 33提问于2020-11-30得票数 1
2回答
在使用Kafka的星火流中，我面临数据丢失的问题，我的用例如下：
火花流(DirectStream)应用程序从Kafka主题读取消息并进行处理。在处理消息的基础上，应用程序将处理后的消息写入不同的Kafka主题，例如，如果消息是协调的，那么将消息写到协调的
浏览 4提问于2017-05-30得票数 8
我有一个单节点kafka代理和简单的流应用程序。我创建了两个主题(topic1和topic2)。Produced on topic1 - processed message - write to topic2我只发了一条信息。有时一条产生的消息有2条写到目标主题，有时写到3条。我不明白为什么会发生这种情况。我的意思是，即使是在代理重新启动后产
浏览 0提问于2018-06-07得票数 2
在spring运行时，可以让kafka源模块作为处理器模块工作吗？有代码样本吗？我正试图实现这样的目标: http (xd源)、xd源(xd处理器)、xd使用者(xd接收器)stream create kafkaSourceTest --definition "http --outputType=application/json | kafka --zkconnect=localh
浏览 3提问于2015-05-12得票数 0
回答已采纳
默认情况下，kafka stream使用快速本地KV存储RocksDB作为其状态存储。我的问题是，如果节点崩溃，并且那些要重放状态的主题消息由于kafka保留而被丢弃(所以它不能从头开始重放消息)，kafka流是否会丢失状态？这是不是意味着那些旧的状态在这种情况下丢失了？谢谢!
浏览 0提问于2018-03-18得票数 1
1回答
我使用Spark作为批处理来处理来自kafka的日志。在每个周期中，我的代码应该得到到达kafka消费者的任何东西。然而，我想对每个周期从kafka获得的数据量进行重新划分。while True: offset = OffsetRange(topic, 0, fromOffset, untilOffset)
kafka_content= Kaf
浏览 7提问于2017-01-28得票数 0
回答已采纳
1回答
我正在查看Logstash的插件，我注意到它为kafka提供了一个输入和输出插件。对于单个主题，Logstash是否可以将kafka主题消息转发到另一个kafka服务器，或者我是否应该预期会有一些数据丢失？
浏览 0提问于2019-10-14得票数 0
我使用Kafka Rest代理来生成和使用Azure APIM从Kafka主题消费，因此使用Rest代理的原因。我可以使用auto.offset.reset =最早的线性方式使用消息，所以每次调用Kafka Rest都会给我最新的消息，这是很棒的，A1 is data requestor
B2，而B2使用卡夫卡主题C3提供的消息给A1，但由于某些
浏览 2提问于2022-07-21得票数 0
我们有kafka流应用程序。Producer在将kafka消息发送到Kafka流媒体应用之前，会在消息中添加头部。在Kafka流媒体应用中，我们使用AbstractProcessor和context.forward(null, Optional.of(event));将消息转发到另一个主题。 但是报头正在丢失。我希望标题是从输入消息到输出主题。 ProcessorContext接口。headers(
浏览 5提问于2020-06-10得票数 0
回答已采纳
当hdfs不可用时，有没有办法确保数据的安全性？场景是: kafka-source，flume memory-channel，hdfs-sink。如果flume服务宕机，是否可以存储topic分区的偏移量，并在恢复后从正确的位置消费？
浏览 3提问于2016-06-12得票数 0
作为我的客户端站点设计决策的一部分，http请求-响应流中涉及的组件(微服务)允许生成关于kafka主题()的消息，但不允许使用来自kafka主题的消息。这类组件(微服务)可以读写数据库，可以与其他组件对话，可以生成主题消息，却不能使用来自卡夫卡主题的消息。
相反，设计建议编
浏览 2提问于2021-01-29得票数 0
云服务器
ICP备案
云直播
腾讯会议
实时音视频
腾讯云开发者
