将消费者偏移量从Kafka流重置到开头
是指将消费者在Kafka消息队列中的偏移量重新设置为最早的位置，以便重新消费之前的消息。这个操作通常在以下情况下使用：
- 数据重处理：当需要重新处理之前的消息时，可以将消费者偏移量重置到开头，以便重新消费所有消息并进行处理。
- 错误修复：如果消费者在处理消息时发生错误，可以将偏移量重置到开头，然后重新消费消息以修复错误。
- 数据回滚：当需要回滚到之前的某个时间点或特定偏移量时，可以将偏移量重置到开头，然后重新消费消息直到目标时间点或偏移量。
为了实现将消费者偏移量重置到开头，可以使用Kafka提供的工具或API。以下是一些常用的方法：
- 使用Kafka命令行工具：可以使用kafka-consumer-groups.sh脚本来重置消费者组的偏移量。具体命令如下：kafka-consumer-groups.sh --bootstrap-server <kafka服务器地址> --group <消费者组ID> --topic <主题名称> --reset-offsets --to-earliest --execute这将把消费者组的偏移量重置为最早的位置。
- 使用Kafka客户端API：可以使用Kafka提供的Java或其他编程语言的客户端API来编写代码，将消费者偏移量重置到开头。具体方法取决于所使用的编程语言和Kafka客户端库。
推荐的腾讯云相关产品是腾讯云消息队列 CMQ（Cloud Message Queue），它是一种高可靠、高可用、分布式的消息队列服务，适用于大规模分布式系统中的消息通信。CMQ提供了消息队列和主题订阅两种模式，可以满足不同场景下的需求。CMQ支持消息的持久化存储和可靠投递，具备高吞吐量和低延迟的特点。
腾讯云消息队列 CMQ产品介绍链接地址：https://cloud.tencent.com/product/cmq
相关·内容
我正在学习卡夫卡。我在Spring boot中创建了一个非常基本的Kafka消费者。以下是相关的类： KafkaConfig.java： @EnableKafka
@Configuration
public class KafkaConfig {
@Bean
public ConsumerFactory<String, String> consumerFactory() {
Map<String, Object> config = new HashMap<>();
config.
浏览 45提问于2021-10-01得票数 0
我的项目使用Kafka 0.10.2版本。在消费者中设置enable.auto.commit=false和auto.offset.reset=latest。如果在维护后重新启动使用者，则使用者将再次从第一个偏移量读取，而不是等待最新的偏移消息。为什么会发生这种事？我是否错误地理解了结构？
我的要求是，使用者不应该自动提交，应该只读取新的消息放入主题时，它是活动的。
浏览 2提问于2020-04-14得票数 1
回答已采纳
2回答
我是Apache Nifi的新手，但我在尝试连接kafka微服务(与生产者)和Apache nifi消费者时遇到了一些问题。 基本上，我有一个这样的docker-compose： zookeeper:
container_name: zookeeper_test
image: wurstmeister/zookeeper #zookeeper:3.5.7
ports:
- 2181:2181
kafka:
container_name: kafka_test
image: wurstmeister/kafka #:2.13-2.6.0
ports:
- 90
浏览 37提问于2021-07-14得票数 0
我试图在每次调用消费者时重置消费者offset，这样当我多次调用消费者时，它仍然可以读取生产者发送的记录。我设置了props.put("auto.offset.reset","earliest");并调用了consumer.seekToBeginning(consumer.assignment());，但是当我第二次调用消费者时，它将不会收到任何记录。我该如何解决这个问题呢？
public ConsumerRecords<String, byte[]> consumer(){
Properties props = new Propert
浏览 0提问于2018-05-08得票数 2
1回答
我使用Apache Nifi、Spark和Kafka在它们之间发送消息。首先，我使用Nifi获取数据，并将其发送到Spark进行处理。然后，我再次将数据从Spark发送到Nifi，以将其插入到DB中。
我的问题是，每次我运行Spark，我都会得到同样的3.142条记录。我停止了Nifi的第一部分，第二部分正在运行，每次我运行Spark时，我都有相同的3.142条记录，现在我无法理解这些数据。
它是从哪里来的？
我尝试查看Kafka-Queue-I (从Nifi到Spark)或Kafka-Queue-II (从Spark到NiFi)上是否有数据，但在这两种情况下，答案都是否定的。只有当我运行Sp
浏览 15提问于2020-02-14得票数 0
我已经开始我的生产者发送数据到Kafka，也开始我的消费者拉相同的data.When，我正在使用消费卡夫卡处理器(kafka版本1.0)在Apache Nifi，我在我的脑海中有几个问题，与Kafka消费者相关。
Q.1)当我第一次启动ConsumeKafka处理器时，如何读取开始消息和当前消息？
Q.2)还有在Kafka中消费者关机的情况下，如何在最后一条消费消息之后阅读消息？
我们如何在使用Apache Nifi的同时实现上述两个功能？
浏览 20提问于2018-07-30得票数 2
回答已采纳
例如，我的kafka消费者(在consumer Group 1中)正在阅读来自Kafka主题A的消息。
现在，如果该消费者在失败之前消费了12条消息。
当消费者重新启动，并且现在它有不同的消费者组(即消费者组2)时，问题1 -？在重启时，它会从偏移量(或位置)中停止的地方继续，因为该偏移量是由Kafka和/或ZooKeeper存储的，还是会从第一条消息开始消费消息。
问题2->有没有办法确保在重启时(当消费者有不同的消费群体时)，它仍然从重启之前停止的地方开始消费？
为了给你提供上下文，我正在尝试在收到关于kafka主题的消息时更新每个节点/服务器中的内存缓存。为此，我对每个节点/服务器
浏览 1提问于2021-02-04得票数 0
所以基本上我看到的是，我正在使用@KafkaListener消费消息，我故意不想承认它们，所以消费者忽略它们，并在主题中侦听新的消息。因此，如果应用程序重新启动，使用者只会从最后提交的消息开始。因此，我希望避免这种情况，我希望消费者在不重新启动的情况下考虑未提交的消息。下面是我正在使用的配置和@kafkalistener。
ReceiverConfig.java
package com.config;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serializ
浏览 3提问于2019-12-05得票数 1
我是卡夫卡的新手。我想了解当kafka指定的偏移不存在时，kafka消费者的行为。可能由于保留策略或使用者指定了无效的偏移量值，在给定偏移量处的消息被删除。
浏览 0提问于2019-11-20得票数 6
注意:使用kafka_2.11-0.9.0.1
我用3 Partitions创建了一个名为：consumer-tutorials的kafka主题，如下所示：
C:\kafka_2.11-0.9.0.1>.\bin\windows\kafka-topics.bat --describe --topic consumer-tutorial
--zookeeper localhost:2181
Topic:consumer-tutorial PartitionCount:3 ReplicationFactor:1 Configs:
Topic: consu
浏览 2提问于2016-06-21得票数 0
我想删除一个Kakfa消费者组，这样当应用程序创建一个消费者并订阅一个主题时，它可以从主题数据的开头开始。 这是使用当前最新的Confluent平台3.1.2的单节点vm开发，该平台使用Kafka 0.10.1.1。 我尝试正常的语法： sudo /usr/bin/kafka-consumer-groups --new-consumer --bootstrap-server localhost:9092 --delete --group my_consumer_group 我得到了错误： Option [delete] is only valid with [zookeeper]. Note
浏览 253提问于2017-03-03得票数 20
回答已采纳
1回答
我已经使用kafka源连接器将文档从Couchbase获取到kafka。然后将这些文档复制到Mongo DB。 Couchbase --> Source Connector --> Kafka --> Sink Connector -> Mongo 如果源连接器关闭，那么如何再次将所有文档同步到Kafka？ 有没有get和touch功能可以将kafka主题在停机期间所做的所有更改通知给用户？
浏览 20提问于2019-10-17得票数 0
我正在尝试使用最新的Kafka CLI工具来重置消费者偏移量。
kafka-consumer-groups.bat --bootstrap-server kafka-host:9092 --group my-group --reset-offsets --to-earliest --all-topics
因此，我看到了以下输出：
TOPIC PARTITION NEW-OFFSET
FirstTopic 0 0
SecondTopic
浏览 2提问于2017-08-14得票数 47
回答已采纳
我希望我没有犯错，但我记得在Kafka文档中提到，使用高级API你不能从特定的偏移量开始读取消息，但它提到它会改变。
现在是否可以使用高级API从特定分区和特定偏移量读取消息？你能给我举一个怎么做的例子吗？
我使用的是kafka 0.8.1.1。
提前谢谢。
浏览 0提问于2015-12-16得票数 7
回答已采纳
我想使用Kafka高级消费者API，同时我想禁用偏移量的自动提交。我试图通过以下步骤来实现这一点。
1) auto.commit.enable = false
2) offsets.storage = kafka
3) dual.commit.enabled = false
我创建了一个偏移管理器，它定期创建对kafka的offsetcommit请求并提交偏移。
我仍然有以下问题
1)高级消费者API是否会自动从kafka存储中获取偏移量，并使用该偏移量进行初始化？或者我应该使用简单的消费者API来实现这一点？
2)是否所有代理都支持基于kafka的偏移量存储？或者只在一个代理上维护它？
浏览 1提问于2015-04-28得票数 5
我的kafka主题共有10条记录/消息和2个分区，每个分区有5条消息。我的使用者组有2个消费者，每个使用者已经分别从他们分配的分区中读取了5条消息。现在，我希望从开始/开始重新处理/读取来自我的主题的消息(偏移量0)。
我停止了kafka使用者，并按照命令将使用者组偏移量重置为0。
./kafka-consumer-groups.sh --group cg1 --reset-offsets --to-offset 0 --topic t1 --execute --bootstrap-server "..."
我的期望是，一旦我重新开始我的卡夫卡消费者，他们将开始阅读记录从偏移量
浏览 3提问于2019-06-27得票数 4
2回答
我正在使用python运行confluent_kafka客户端。目前，我在尝试生产和消费消息时没有收到任何错误，但问题是生产者说它成功了，但是消费者找不到任何消息。
我已经创建了一个主题，这是我正在使用的构建类：
from confluent_kafka import Producer, Consumer
from config import config
import json
class Kafka:
"""
Kafka Handler.
"""
def __init__(self, kafka_broke
浏览 8提问于2018-10-10得票数 2
这是我的输出。我认为一切都很好：
[2018-07-06 22:27:51,458] INFO [Consumer clientId=consumer-4, groupId=connect-elk-sink] Setting newly assigned partitions [elk-test-0] (org.apache.kafka.clients.consumer.internals.ConsumerCoordinator)
[2018-07-06 22:27:51,471] INFO [Consumer clientId=consumer-4, groupId=connect-elk-
浏览 0提问于2018-07-07得票数 1
我目前正在Dataproc上运行星星之火作业，并且在尝试重新加入一个组并从kafka主题读取数据时出错。我已经做了一些调查，不知道问题是什么。我已经将auto.offset.reset设置为earliest，因此应该读取最早可用的未提交偏移量，最初我的星火日志如下所示：
19/04/29 16:30:30 INFO
org.apache.kafka.clients.consumer.internals.Fetcher: [Consumer
clientId=consumer-1, groupId=demo-group] Resetting offset for
partition
浏览 0提问于2019-04-29得票数 5
我处理活动应用程序的时序数据。因此，旧数据没有任何意义。我只想处理流应用程序启动后收到的数据，而不是之前提交的偏移量。重启后忽略kafka流app旧记录的正确方式是什么？
对于kafka consumer API，我通常使用seekToEnd()方法来跳转到最新的记录。是否有用于流的等效机制？我希望避免过滤自上次提交以来的所有消息，以忽略旧消息。
浏览 3提问于2017-11-15得票数 2
2回答
我已经启动了我的zookeeper和Kafka服务器。我启动了我的Kafka制作器，它发送了10条主题为'xxx‘的消息。然后叫住了我的卡夫卡制作人。现在我启动了我的Kafka用户，并订阅了主题'xxx‘。我的消费者使用我的Kafka生成器发送的10条消息，该生成器现在没有运行。我需要我的Kafka消费者应该只消费运行Kafka服务器的消息。有什么方法可以做到这一点吗？以下是我的消费者属性中的内容。
props.put("bootstrap.servers", "localhost:9092");
String consumeGroup
浏览 1提问于2016-09-30得票数 3
我刚接触Kafka，我并不真正理解Kafka配置的含义，有没有人能给我解释一下更容易理解！
下面是我的代码：
val kafkaParams = Map[String, Object](
"bootstrap.servers" -> "master:9092,slave1:9092",
"key.deserializer" -> classOf[StringDeserializer],
"value.deserializer" -> classOf[StringDeserializer],
&#
浏览 0提问于2017-07-05得票数 21
1回答
我正在尝试使用来使用消息。能够使用kafka-控制台- able .able消费消息。但是，无法使用来自java的消息。没有收到任何错误或消息。帮我做错了什么。
import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import java.util.HashMap
浏览 2提问于2015-12-28得票数 2
我运行一个系统，包括一个InfluxDB、一个卡夫卡代理和产生时间序列数据的数据源(传感器)。代理的目的是保护数据库不受入站事件过载的影响，并作为接收数据的不确定格式平台。数据通过阿帕奇骆驼路线从卡夫卡传输到InfluxDB。我想使用Kafka，一种中间消息缓冲区，以防骆驼路径崩溃或不可用--这是系统中最常见的错误。到目前为止，我还没有实现对Kafka的配置，使入站消息仍然可供以后使用。如何正确配置它？
浏览 1提问于2020-11-20得票数 0
回答已采纳
2回答
使用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保持不变。
带有属性的Kafka生产者：
Properties properties = new Properties();
properties.put
浏览 16提问于2017-01-18得票数 0
我已经写了一个非常简单的Flink流作业，从卡夫卡使用FlinkKafkaConsumer082的数据。
protected DataStream<String> getKafkaStream(StreamExecutionEnvironment env, String topic) {
Properties result = new Properties();
result.put("bootstrap.servers", getBrokerUrl());
result.put("zookeeper.connect", ge
浏览 0提问于2015-11-03得票数 2
1回答
我有一个来自订阅主题的消费者投票。它会消耗每条消息，并进行一些处理(在几秒钟内)，推送到不同的主题并提交偏移量。
总共有5000条信息，
在重新启动之前使用2900条消息和提交的偏移量。
重新启动后-从偏移量0开始消耗。
即使使用相同的使用者组创建使用者，它也开始处理来自偏移量0的消息。
kafka版本(strimzi) > 2.0.0 kafka-python == 2.0.1
浏览 1提问于2020-10-02得票数 1
由于某些原因，Java应用程序中的Kafka使用者重复使用了主题中的消息。
生产者向有四个分区的主题发送消息。我们有一个消费者消费来自这个主题的消息。应用程序总是在工作日运行，周末除外:它不会在周末调用poll方法。
消费者配置:自动提交，自动提交时间为5s (默认)。
应用程序运行得很好，直到一个星期天，它重新开始调用poll方法。我们看到来自这个主题的数百万条消息被轮询。使用者基本上是在轮询来自该主题的所有消息。将新的偏移量与周末停止之前的偏移量进行比较。新的偏移量要小得多，这就像是将所有四个分区的偏移量重置为非常低的数字。
我们不知道消费者端发生了什么，因为它没有调用轮询方法，所以没有打
浏览 3提问于2018-02-10得票数 0
场景： 当ConsumeKafka_2_0在卡夫卡中消费消息时，在卡夫卡中的偏移量会变成100.When Nifi被破坏，使得Nifi恢复到过去的状态，在卡夫卡中的偏移量是50.However，实际上在卡夫卡中的偏移量是"100“，因为卡夫卡没有恢复。 当Nifi重启时，我猜是从偏移量"100“开始消费消息，而不是从偏移量"50”开始，因为在time.So中，偏移量在Kafka中是"100“，有些偏移量的消息是"50”-- "100“会丢失，如何恢复？
浏览 11提问于2020-04-28得票数 0
1回答
我正在使用sparkstreaming从Kafka消费，想想如果在消费者端发生了一些崩溃或类似的事情。我不担心复制记录，但我不能错过任何一条记录。如果我重新启动我的消费者，我将不会从我离开的地方继续。我想我可以通过检查点来做到这一点？？我不知道。但在最坏的情况下，我想检查一下从kafka主题中最古老的记录运行我的消费者的可能性。
浏览 2提问于2018-03-02得票数 0
在我们的项目中，有主动的Kafka服务器(PR)和被动的Kafka服务器(DR)，这两个Kafka brokers在我们的项目中都配置了相同的组名、主题名称和分区。从PR切换到DR时，在DR上手动设置_consumer_offsets。
我这里的问题是，Kafka消费者是否能够无缝地从上次读取的地方消费消息？
浏览 7提问于2020-03-31得票数 2
我们试图从kafka 0.8升级到0.9，我们希望使用作为0.9的一部分发布的kafka consumer offset management。到目前为止，我们将偏移量存储在数据库中，并编写了我们的自定义消费者，它将检查数据库中的偏移量，并从kafka获取消息。
我们如何才能将这些偏移量迁移到kafka。我读过关于kafka offset从zookeeper迁移到kafka的文章，但没有发现任何关于这方面的东西。
浏览 0提问于2016-04-08得票数 0
1回答
我们是否可以在不汇集所有消息的情况下，从偏移量或持续时间内重新运行消息？如果是，请让我知道。我试过使用Kafka API，但它首先消耗所有消息，然后根据记录进行过滤。我的需求是发送一个偏移量作为输入以获取消息详细信息，或者发送自和结束日期/时间间隔以获取记录
浏览 4提问于2018-01-30得票数 0
我正在使用poll()从一个主题中获取一堆消息(比如100条)。在我的配置中，我将自动提交设置为false，并将max.poll.records设置为100。
我将使用收到的100条消息中的10条消息进行处理。因此，我希望收到的其余90条消息是下一个轮询响应的一部分。
因此，我提取了第10个消息偏移量，并将其作为参数传递给commitSync() API，如下所示。
consumer.commitSync(Collections.singletonMap(
new TopicPartition(message.topic(), messa
浏览 20提问于2020-05-18得票数 1
回答已采纳
5回答
我有一个问题，假设有一个主题T1，有两个消费者C1，C2属于两个不同的组，当前偏移量是0。我们知道Kafka维护consumer.So的偏移量，如果C1消费消息，偏移量变成1，那么如果C2消费消息，它是从1偏移量开始还是从0偏移量开始消费消息？意味着如何维护两个不同消费者组的偏移量？
谢谢
浏览 1提问于2020-02-20得票数 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
1回答
我计划在Informatica中使用Kafka数据对象。我想每5分钟通过Informatica从Kafka读取数据。我很少有疑问，卡夫卡如何认识到当前的偏移，从它必须加载。我们可以设置一个主题的偏移量，Informatica需要从它加载Kafka数据对象中的增量或更改的数据吗？我找到了以下答案，请您纠正我的确切过程: 1)动物园管理员是否将当前偏移量存储到加载数据的位置。2)若要读取上次提交的偏移量后的消息，需要将使用者配置文件中的auto.offset.reset参数设置为最大。
请给我一个如何从卡夫卡获取数据的方法。
致以敬意，
斯利纳特
浏览 2提问于2017-06-30得票数 0
我有一个卡夫卡的主题，我想读从最早的事件。
我想要做的是从一个主题获取所有数据(从绝对最早的事件到某一特定日期的事件)。
每个事件的结构都有一个名为dateCliente的字段，我使用该字段作为筛选事件的阈值。到目前为止，我已经完成了读和写。我正在写一个临时的拼花文件，我用它作为一个Hive表的分区。但是，尽管我已经在auto.offset.reset参数中指定了最早的参数，但它从一开始就没有读取数据。
每当我运行我的代码，我就会得到从这个日期开始的所有事件。每次我再次执行代码时，它都会继续读取Kafka事件，这是我在前一次代码执行中看到的最后一个事件。
我用来配置Kafka消费者和订阅主题的
浏览 0提问于2018-04-20得票数 7
回答已采纳
1回答
我使用的是Spotify docker版本的Kafka。我现在只有一个生产者、一个经纪人和一个消费者。
问题是，使用这种配置，我无法从Kafka获取最新消息。如果我默认使用auto_offset_reset='earliest'，那么我会得到所有的消息。
这是我为消费者编写的代码，它不会打印来自Kafka的任何消息，即使那里有新消息。
def consumer():
consumer = KafkaConsumer('stories',
enable_auto_commit=False,
浏览 21提问于2018-08-16得票数 1
1回答
我用scala编写了kafka consumer。当我运行consumer时，控制台上显示为空白。我使用了以下代码：
val topicProducer = "testOutput"
val props = new Properties()
props.put("bootstrap.servers","host:9092,host:9092")
props.put("key.deserializer","org.apache.kafka.common.serialization.StringDeseria
浏览 0提问于2018-03-27得票数 0
我们有主题，保留为7天(168小时)。当生产者发送消息时，就会实时地使用消息.一切都如期而至。然而，最近在一台生产服务器上，Devops意外地将时区从PST更改为EST，作为OS修补程序的一部分。
在Kafka服务器重新启动后，我们看到很少(不是全部，而是随机的)旧消息正在被消费者消耗。我们要求Devops将其更改为PST并重新启动。同样，这些旧信息也在本周末再次出现。
在较低的环境(开发、质量保证、阶段等)中，我们还没有看到这个问题。
卡夫卡版本: kafka_2.12-0.11.0.2
任何帮助都是非常感谢的。
添加更多信息..。最近，我们的CentOS进行了补丁更新，不知何故，管理员从P
浏览 0提问于2018-06-24得票数 3
当我在对一个老话题做一些测试时，我发现了一些奇怪的行为。读了一下卡夫卡的日志，我注意到这条“删除了8个过期的偏移量”的信息：
[GroupCoordinator 1001]: Stabilized group GROUP_NAME generation 37 (kafka.coordinator.GroupCoordinator)
[GroupCoordinator 1001]: Assignment received from leader for group GROUP_NAME for generation 37 (kafka.coordinator.GroupCoordinator)
浏览 3提问于2016-08-24得票数 36
2回答
我正在使用Kafka streams，并希望将一些从Java到开头的消费者偏移量重置。KafkaConsumer.seekToBeginning(...)听起来是正确的选择，但我使用的是Kafka Streams：
KafkaStreams streams = new KafkaStreams(builder, props);
...
streams.start();
我猜，根据我定义的具体流管道，这将在引擎盖下创建几个消费者。我能接触到这些吗？或者，是否有其他方法可以通过编程方式重置偏移量？
浏览 23提问于2017-07-09得票数 2
回答已采纳
这是一个基于Armando Ballaci提供的答案的"Where do zookeeper store Kafka cluster and related information?“的后续问题。 现在很明显，消费者偏移量存储在Kafka集群中的一个名为__consumer_offsets的特殊主题中。很好，我只是想知道这些偏移量的检索是如何工作的。 主题不像RDBS，我们可以在RDBS上根据某个谓词查询任意数据。例如，如果数据存储在RDBMS中，下面这样的查询可能会获得某个消费者组的特定消费者的特定主题分区的消费者偏移量。 select consumer_offset__read,
浏览 40提问于2020-07-05得票数 1
我想测试一个kafka示例:生产者：
object ProducerApp extends App {
val topic = "topicTest"
val props = new Properties()
props.put("bootstrap.servers", "localhost:9092")
props.put(ConsumerConfig.GROUP_ID_CONFIG, "consumer")
props.put("key.serializer", "org.apache.kafka.
浏览 0提问于2017-11-03得票数 0
我试图整合星火和卡夫卡，因为它消耗了卡夫卡的信息。我也有生产者代码来发送关于“临时”主题的信息。另外，我正在使用卡夫卡的控制台制作人来制作关于“临时”主题的信息。
我已经创建了下面的代码来使用来自同一个“临时”主题的消息，但是它也不会接收单个消息。
计划：
import java.util.Arrays;
import java.util.Map;
import java.util.HashMap;
import static org.apache.commons.lang3.StringUtils.SPACE;
import org.apache.spark.SparkConf;
impo
浏览 3提问于2017-08-31得票数 1
我们正在使用Kafka流媒体库来构建关于Kafka主题的传入消息的实时通知类系统，因此当流媒体应用程序运行时，它会实时处理主题中的所有传入消息，并在遇到某种预定义的传入消息时发送通知。
如果流应用关闭并重新启动，我们只需要处理流应用初始化后最近到达的消息。这是为了避免处理在流媒体应用程序未运行或停机时未处理的旧记录。默认情况下，流式应用程序开始处理自上次提交偏移量以来的旧消息。Kafka Streaming App有没有什么设置，可以只处理最近的消息？
浏览 2提问于2018-05-10得票数 2
我刚接触卡夫卡，当我们使用卡夫卡时，我们可以设置retention.bytes。假设我们设置为1G，如果消息量达到1G，kafka会删除消息，我想问一下偏移量是否会重置为零？第二，消费者将auto.offset.reset设置为最大，在kafka删除消息后，消费者会开始什么偏移量？
浏览 15提问于2017-01-19得票数 0
回答已采纳
我使用了下面的Kafka消费者示例代码：
KafkaConsumer kconsumer = new KafkaConsumer(props);
ConsumerRecords<String, byte[]> records = kconsumer.poll(timeoutInMS);
kconsumer.commitSync();
我有两个问题：
(1)偏移量提交的地方是Kafka还是Zookeer？(我知道旧的kafka版本是提交给zookeeper的，但是新的kafka版本仍然是ZK，我甚至没有在配置文件中指明zookeeper的IP )。
(2)如果需要重复使用数据，如何检
浏览 4提问于2018-01-12得票数 0
我是Kafka的新手，我已经尝试过Kafka-Python包。
我设法设置了一个简单的生产者和消费者，它可以发送和接收消息。在这种情况下，消费者没有使用消费组，如下所示：
consumer = KafkaConsumer(queue_name, bootstrap_servers='kafka:9092')
但是，当我开始使用group_id时，如下所示，它停止接收任何消息：
consumer = KafkaConsumer(bootstrap_servers='kafka:9092', auto_offset_reset='earliest',
浏览 1提问于2019-03-19得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
