1回答
我知道可以通过在kafka-console-consumer中使用--offset从kafka中读取特定偏移量的数据，但这会给出从该特定偏移量到当前偏移量的记录。如何才能将其限制为只有一条记录。
浏览 34提问于2021-06-16得票数 1
回答已采纳
有许多线程描述如何从指定的时间戳开始从Kafka获取记录。所以我认为Kafka‘知道’它存储的每一条记录的时间戳。我需要获取具有指定分区和偏移量的记录的时间戳。有可能吗？信息系统将错误的数据放到Kafka中(客户订单中的产品id错误)，我需要分析日志文件来找出原因。知道该记录的时间戳会容易得多。
浏览 32提问于2021-05-03得票数 0
我有一个java Kafka消费者，在其中我正在批量获取ConsumerRecords进行处理。system which can take random time for different requests or timeout in 5 seconds.我遇到的问题是，如果产生了较晚的记录，但前一条记录仍未超时，则如何提交或提交哪个偏移量。假设我在一批中获得了2条记录，第一条消息的外部调用仍在等待，而第二次调用已完成。如果我等待5秒等待外部响
浏览 2提问于2018-01-05得票数 2
有没有办法从kafka主题中删除一条记录？我知道有脚本kafka- delete -records.sh可以删除指定主题和分区上指定偏移量之前的记录，但我希望能够删除指定偏移量上的记录。这不是在Java上，而是在kafka的裸实例上。
浏览 9提问于2020-02-13得票数 1
回答已采纳
我们正在使用kafka Topology forward向kafka主题发送记录。
我们之前使用了一个单独的生产者来发布消息，并且我们能够获取消息的偏移量和分区。如何使用context.forward获取Kafka Sink处理器发送的记录的偏移量和分区
浏览 0提问于2017-11-01得票数 0
GenericArray<GenericRecord>>> consumerRecords, Acknowledgment acknowledgment) { kafka: key-deserializer: org.apache.kafka.common.serialization.ByteArrayDeserializervalue-deserializer: io.confluent.kafka.serializers.KafkaAvr
浏览 6提问于2022-08-31得票数 1
librdkafka包含函数rd_kafka_position，它获取给定主题-分区的当前偏移量。但评论说：
每个请求分区的\p偏移字段将被设置为上次使用的消息+ 1的偏移量，如果没有先前的消息，则设置为RD_KAFKA_OFFSET_INVALID。在这种情况下，我想知道的是，如果一个消息被消耗，那么下一个消息的偏移量是多少。换句话说，在第一种情况下，它应该是传递给rd_kafka_seek的相同偏移量，在第二种情况下，应该是1加上调用rd_
浏览 6提问于2020-07-24得票数 2
但是我不想自动提交偏移量，因为一旦从Kafka获取记录，我就有一些处理逻辑。
从Kafka获取记录之后，如果处理成功，那么我只想提交偏移量，否则它应该再次从相同的偏移量读取。
浏览 2提问于2019-10-22得票数 3
1回答
我在Kafka上使用手动偏移量提交。当我执行consumer.commitOffsets()时，有没有办法获得已提交的偏移量列表；另外，我使用的是一个高级消费者。
浏览 11提问于2016-08-10得票数 0
回答已采纳
3回答
我在一个消费者组中有两个消费者，他们分配了相同的kafka主题分区。我希望从消费者B内部获得消费者A的最后一次读取偏移量。有什么想法，如何实现这一点？
浏览 2提问于2016-08-01得票数 7
回答已采纳
我想写一个函数，一旦主题中的最后一条消息被读取，它就会调用回调。function getCurrentMessages(kafka, topic, cb_done){ var consumer= new kafka.Consumer(new kafka.Client(), [{topic: topic, offset: 0}], {fromOffset: true});
consumer.on
浏览 3提问于2015-10-03得票数 0
1回答
我是第一次接触kafka和spark structured streaming。我想知道spark in batch模式是如何知道从哪个偏移量读取的？如果我指定"startingOffsets“为”最早的“，我只会得到最新的记录，而不是分区中的所有记录。我在两个不同的集群中运行了相同的代码。集群A(本地机器)获取了6条记录，集群B( TST集群-第一次运行)获取了1条记录。df = spark \ .for
浏览 0提问于2020-10-24得票数 0
有一个消息到达的Kafka主题。我需要阅读一条消息，对其进行处理，然后继续下一条消息。消息处理可能会失败，如果发生这种情况，处理必须重试几次(比方说10次)，然后才能继续处理下一条消息。如果处理失败10次，则需要删除该消息，我们应该继续处理下一条消息。 我们使用reactor-kafka，所有的处理都需要反应式的。这里的问题是，如果在允许的10次尝试后仍然失败，则偏移量不会被提交(当然)，因此下次从Kafka读取相同的偏移量，因此，有
浏览 23提问于2020-04-09得票数 2
我正在用springboot 2，kafk 2.2.0，spring-kafka 2.2.5做项目 我做了kafka exactly once环境，消息的产生和使用都很好。但是kafka-consumer-groups.sh是这样说的。发了一条消息，但是LOG-END-OFFSET加倍了，总是有1个延迟。(在我的java应用程序中，生产和消费工作按预期进行) 我不知道为什么对数末端偏移量会加倍。kafkaTemplate.executeInTransaction(kt -> kt.send(&quo
浏览 30提问于2019-05-17得票数 4
回答已采纳
我在Scala中使用Spark在Kafka消费者应用程序中消费和处理消息。有时候，处理Kafka消息队列中的消息会比平时多花一点时间。properties = readProperties()
在这方面的任何帮助都将非常感谢。谢谢
浏览 3提问于2017-08-31得票数 1
有没有基于offset消费Kafka topic消息的方法？我的意思是，我有一个偏移量id，我以前在一个主题中发布过。现在我需要根据我传递的偏移量Id从topic获取一条消息。
浏览 0提问于2016-12-28得票数 3
当我们准备好时，我们希望消费者将使用偏移量。由于文档的原因，我并不能真正理解seekToEnd代表什么。这是否意味着，它将查找到偏移量(最后一条ACKed消息)，还是将查找到主题的最后一条消息(无论偏移量如何)并等待新的消息？
浏览 0提问于2020-02-12得票数 0
我需要使用confluent-kafka-python检索主题的最后N条消息。请帮帮忙。
浏览 17提问于2022-02-19得票数 2
回答已采纳
3回答
我在客户端使用kafka : kafka_2.12-2.1.0，spring kafka，遇到了一个问题。
我需要通过读取kafka主题中的所有现有消息来加载内存中的映射。为此，我启动了一个新的消费者(具有唯一的消费者组id，并将偏移量设置为earliest)。然后，我遍历消费者(poll方法)以获取所有消息，并在消费者记录为空时停止。但我注意到，当我开始轮询时，前几次迭代返回的消费者记录为空，然后开始返回实际记录。现在，这打破了我的
浏览 1提问于2019-02-11得票数 3
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
云直播活动推荐
腾讯云开发者
