如何在Kafka主题中仅消费最新偏移量

文章/答案/技术大牛

发布

2回答

scala、apache-kafka、kafka-consumer-api

我正在开发一个使用kafka的scala应用程序。我的kafka消费者代码如下所示。") props.put(data <- record) return va

浏览 53提问于2020-07-27得票数 1

1回答

我需要创建一个Kafka设置，在每天开始时，必须删除主题中的所有消息，并将其偏移量重置为0。我曾尝试通过将log.retention设置为1天来实现此目的，但似乎不起作用。我还运行了kafka-delete-records。它确实删除了所有的消息，但kafka.tools.GetOffsetShell仍然给了我最新的偏移量。我想让它变成0。另外，我不会显式地设置消费者组。它们是内部的，像_confluent-ksql-default_transien

浏览 17提问于2019-06-25得票数 0

1回答

对Kafka 1.x中的提交和获取消费者偏移量感到困惑

apache-kafka

我使用了下面的Kafka消费者示例代码：ConsumerRecords<String, byte[]> records = kconsumer.poll(timeoutInMS);我有两个问题： (1)偏移量提交的地方是Kafka还是Zookeer(我知道旧的kafka版本是提交给zookeeper的，但是新的kafka版

浏览 4提问于2018-01-12得票数 0

2回答

如何获取Kafka消费者的偏移量？

apache-kafka、spark-streaming、kafka-consumer-api、kafka-producer-api

使用Kafka(v2.11-0.10.1.0)-spark-streaming(v-2.0.1-bin-hadoop2.7). 我有Kafka生产者和Spark-streaming消费者来生产和消费。一切正常，直到我停止消费者(大约2分钟)并重新开始。消费者启动并读取数据，绝对完美。但是，我迷失了2分钟的数据，其中消费者是关闭的。Kafka consumer/server.properties保持不变。, "", docu

浏览 16提问于2017-01-18得票数 0

2回答

Apache nifi和kafka微服务

docker、apache-kafka、microservices、apache-nifi

我是Apache Nifi的新手，但我在尝试连接kafka微服务(与生产者)和Apache nifi消费者时遇到了一些问题。: image: wurstmeister/kafka #:2.13-2.6.0 - 9092:9092 KAFKA_ADVERTISED_HOST_NAME: kafka KAFKA_AUTO_CRE

浏览 37提问于2021-07-14得票数 0

1回答

如何使用KafkaAdminClient (Java)为2.x版本获取卡夫卡主题的最新偏移量/大小

java、apache-kafka

是否有更有效/更简单的方法来使用最新的Java客户端2.4API获取主题/分区的大小/最新偏移量？然后，通过将消费者组的偏移量与主题的大小进行比较，计算消费者群体的滞后.我知道以前的Kafka版本也问过这个问题，也有一种方法可以从Kafka公开的JMX度量中获取这些信息，但是我仍然被一个遗留应用程序困住了，它需要用Java来完成，但我需要使用最新的2.4 Kafka语言据我所知，获取这些

浏览 2提问于2020-02-05得票数 3

回答已采纳

1回答

spring-cloud-stream-kafka应用启动后只消费最新消息

spring-integration、spring-cloud-stream、spring-kafka

在我们的项目中，我们使用spring-cloud-stream-binder-kafka版本1.1.2与kafka集成。最近，我们遇到了这样一种情况:我们的一个服务在启动后使用了某个主题中的旧消息(已经使用过)。该主题有2个分区和2个消费者分组在一个消费者组下。我们不确定偏移量是否正确提交给zookeeper。[-kafka-listener-2] ERROR o.s.k.listener.LoggingErrorHandler.handle - Error while

浏览 14提问于2017-06-28得票数 0

1回答

Kafka Json值反序列化程序

serialization、apache-kafka、deserialization、json-deserialization

我使用的是具有以下属性的kafka消费者： key.deserializer=org.apache.kafka.common.serialization.StringDeserializer value.deserializer=org.apache.kafka.connect.json.JsonDeserializer KafkaProducer(value.serializer=org.apache.kafka.connect.json.JsonSerializer)正在将JSON记录推送到一个主<e

浏览 58提问于2021-08-20得票数 0

回答已采纳

1回答

有没有办法使用camel获取Kafka主题中产生的最后一条消息？

spring、apache-kafka、apache-camel

我有一个带有一个分区的kafka主题。当我的服务器重新启动时，我希望使用主题中最后生成的消息。

浏览 2提问于2021-07-02得票数 0

2回答

Zookeeper如何从__consumer_offsets主题中检索消费者偏移量？

apache-kafka、apache-zookeeper、kafka-consumer-api、kafka-topic

这是一个基于Armando Ballaci提供的答案的"Where do zookeeper store Kafka cluster and related information?“的后续问题。现在很明显，消费者偏移量存储在Kafka集群中的一个名为__consumer_offsets的特殊主题中。很好，我只是想知道这些偏移量的检索是如何工作的。例如，如果数据存储在RDBMS中，下面这样的查询可能会获得某个消费者组的特定消费者的特定主题分区

浏览 40提问于2020-07-05得票数 1

1回答

无法从zookeeper检索主题/组/分区的kafka偏移量

apache-zookeeper、apache-kafka

我们在不同的服务器上运行zookeeper 3.3和kafka 0.8。我们正在使用HL消费者，他们会像预期的那样访问kafka队列中的数据，在重启时，他们会从中断的地方重新开始。因此，消费者的行为符合预期。问题是当我们使用zkCli.sh时，我们在zookeeper中看不到偏移量。现在，使用者被设置为仅针对一个主题在一个分区中运行。因为当消费者停止并再次重启时，消费者的行为如预期的那样(即，它从上次运行时停止的偏移量</em

浏览 2提问于2015-02-04得票数 1

1回答

使用Akka Stream和Kafka* offset提交将事件从Kafka流式传输到Couchbase*

scala、apache-kafka、akka、akka-stream、alpakka

我正在尝试设计一个Akka Stream，使用Alpakka来读取kafka主题中的事件，并将它们放到Couchbase中。，并将它们作为json文档放到Couchbase中，尽管我不知道如何将消费者偏移量提交给Kafka，但它看起来甚至很好。如果我已经清楚地理解了隐藏在Kafka消费者偏移量后面的主要思想，在任何故障或重启发生的情况下，流从最后提交的偏移量读取所有消息，并且由于我们没有提交任何偏移量，它可能会再次重

浏览 21提问于2019-10-18得票数 0

1回答

当您重新启动火花作业时，如果它在输入给kafka的数据中遇到了意外的格式，会发生什么？

apache-spark、apache-kafka、spark-structured-streaming

如何解决卡夫卡主题中的数据问题，以恢复中止的火花作业？

浏览 0提问于2020-05-12得票数 3

回答已采纳

3回答

不清楚卡夫卡中auto.offset.reset和enable.auto.commit的含义

apache-kafka、kafka-consumer-api

我刚接触Kafka，我并不真正理解Kafka配置的含义，有没有人能给我解释一下更容易理解！

浏览 0提问于2017-07-05得票数 21

1回答

从mysql中提取每日数据更改并将其部署到timeseries DB

mysql、apache-kafka、apache-kafka-connect、debezium、druid

但我只需要在那个时间/一天内所做的最新更改，并需要将这些数据输入到timeseries DB (计划使用druid) 在读取binlog时，有没有什么机制可以避免重复并保留最新的更改？

浏览 4提问于2021-10-28得票数 0

1回答

如何从FlinkKafkaConsumer获取最新的消息偏移量？

apache-flink、kafka-consumer-api

我正在使用FlinkKafkaConsumer来消费来自kafka的数据？但我无法从收到的数据中获得最新的偏移量。如何在flink?.Could中做到这一点?有没有人指导我？

浏览 4提问于2017-07-11得票数 0

1回答

再处理/再读卡夫卡记录/信息-消费者集团抵消重置的目的是什么？

apache-kafka、kafka-consumer-api

我的kafka主题共有10条记录/消息和2个分区，每个分区有5条消息。我的使用者组有2个消费者，每个使用者已经分别从他们分配的分区中读取了5条消息。现在，我希望从开始/开始重新处理/读取来自我的主题的消息(偏移量0)。.我的期望是，一旦我重新开始我的卡夫卡消费者，他们将开始阅读记录从偏移量0，即开始，但这没有发生，他们调查了他们的最后位置，即偏移量5。为什么

浏览 3提问于2019-06-27得票数 4

2回答

Kaka auto.offset.reset查询

apache-kafka、kafka-consumer-api

我的项目使用Kafka 0.10.2版本。在消费者中设置enable.auto.commit=false和auto.offset.reset=latest。如果在维护后重新启动使用者，则使用者将再次从第一个偏移量读取，而不是等待最新的偏移消息。为什么会发生这种事？我是否错误地理解了结构？

浏览 2提问于2020-04-14得票数 1

回答已采纳

1回答

如何使用Pykafka获取主题的最新消息？

python、apache-kafka、kafka-consumer-api、pykafka

我一直在使用pykafka生成一个主题的消息我想知道最新的消息。partitions.iteritems()]consumer.consume() 但是，我并不真正理解这里发生了什么，只有在至少有两条消息已经存在的情况下，它才能获得最新的消息

浏览 5提问于2017-06-27得票数 1

1回答

如何通过消费kafka主题消息而不是Rest API来创建命令

axon

我使用的是Axon版本(3.3)，它在SpringBoot主类中使用在我们的用例中，命令端微服务需要从kafka主题中挑选消息，而不是将其公开为Rest api。它会将事件存储在事件存储中，然后将其移动到另一个kafka主题，供查询端微服务消费。由于KafkaAutoCOnfiguration被禁用，我不能使用spring-kafka配置来编写消费</

浏览 18提问于2019-07-12得票数 0

点击加载更多