在kafka分区中压缩的偏移量会发生什么情况？
在 Kafka 分区中压缩的偏移量会发生以下情况：
- 压缩的偏移量是指在 Kafka 中，当消息被消费者成功消费后，消费者会提交当前消费的偏移量。为了减少存储空间和提高性能，Kafka 提供了偏移量压缩的功能。
- 当压缩的偏移量发生时，Kafka 会将已经消费的消息的偏移量进行压缩，以减少存储空间的占用。压缩后的偏移量会被存储在 Kafka 的内部主题 "__consumer_offsets" 中。
- 压缩的偏移量可以通过配置参数来控制，例如可以设置多久进行一次压缩，或者设置压缩的阈值。
- 压缩的偏移量的优势在于减少存储空间的占用，特别是在大规模的消息系统中，可以显著降低存储成本。
- 压缩的偏移量的应用场景包括大规模的实时数据处理、日志收集和分析等。
- 腾讯云提供了 Kafka 服务，可以使用腾讯云的云原生消息队列 CMQ（Cloud Message Queue）来实现消息的压缩和偏移量的管理。具体产品介绍和使用方法可以参考腾讯云的官方文档：腾讯云云原生消息队列 CMQ。
相关·内容
我想知道spring boot内部的spring kafka Producer是否保证交付。
有人知道如果某个随机监听器无法接收消息会发生什么情况吗？spring kafka是否会重试发送消息？
浏览 0提问于2019-05-29得票数 0
我有一个被压缩的主题：
/opt/kafka/bin/kafka-topics.sh --zookeeper localhost --describe --topic myTopic
Topic:myTopic PartitionCount:1 ReplicationFactor:1 Configs:cleanup.policy=compact
它上没有任何信息：
/opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic myTopic --from-beginning --p
浏览 0提问于2019-05-02得票数 0
回答已采纳
我对Kafka很陌生，我有一个使用java微服务的kafka流，它消耗了由生产者和过程产生的kafka主题的信息。使用auto.commit.interval.ms设置了kafka提交间隔。我的问题是，在提交之前，如果微服务崩溃，那么处理但未提交的消息会发生什么情况？会有复制的记录吗？如果发生这种重复，该如何解决？
浏览 6提问于2022-02-25得票数 1
我们有一个在YARN集群上运行的Spark Streaming application。
它接收来自Kafka topics的消息。
实际上，我们的处理时间超过了批处理间隔。
Batch Interval : 1 Minute
Processing Time : 5 Minutes
我想知道，如果在处理时间之间接收到一些数据会发生什么情况，在处理结束之前这些数据会在内存中可用吗？还是会在后续的数据获取中被覆盖？
我们使用Direct Streaming approach从Kafka主题中获取数据。
我应该使用Window based操作吗？例如，如果我有Window length as 5 M
浏览 10提问于2017-02-07得票数 6
回答已采纳
超出保留期的主题会发生什么情况？消息将被清除，但是主题是否仍然存在，如果存在，如果主题上只有一个分区，它是否会写入偏移量0？
浏览 0提问于2017-07-25得票数 1
我需要一种强制压缩__consumer_offsets主题的方法。在一个测试环境中，我尝试删除文件清洁器-偏移-检查点，然后kafka删除了许多段，如下所示。在生产环境中删除此文件安全吗？
删除清洁剂之前的-偏移-检查点：
# ls -la data/kafka/__consumer_offsets-*/*
-rw-r--r--. 1 root root 0 14 giu 17.20 data/kafka/__consumer_offsets-0/00000000000000000000.log
-rw-r--r--. 1 root root 35452 14 giu 17
浏览 2提问于2021-06-15得票数 0
回答已采纳
一旦我从kafka收到一条消息，我需要运行一个长时间运行的进程(最多需要20秒)，只有当这个过程完成时，我才需要将一条消息视为成功。
此外，我需要确保每条消息至少处理一次。
考虑使用具有以下属性的KafkaMessageListenerContainer：
ThreadPoolTaskExecutor用于listenerTaskExecutor
使用MessageListener类型的AcknowledgingMessageListener
将确认模式设置为MANUL_IMMEDIATE。
但是，我唯一的问题是，如果带有偏移量的特定消息(例如15 )首先被成功处理，但是带有
浏览 1提问于2016-08-14得票数 1
我正在使用Kafka运行一个结构化的流应用程序。我发现如果由于某种原因系统停机了几天...检查点变得陈旧，并且在Kafka中找不到与检查点对应的偏移量。我如何让Spark结构化流媒体应用选择最后一个可用的偏移量，并从那里开始。我尝试将偏移量重置设置为较早/最新，但系统崩溃，出现以下错误：
org.apache.kafka.clients.consumer.OffsetOutOfRangeException: Offsets out of range with no configured reset policy for partitions: {MyTopic-574=6559828}
at
浏览 3提问于2020-04-19得票数 0
这个问题基本上类似于这里提出的问题：。即，如果作业在两个检查点之间重新启动，会发生什么情况？它是否会重新处理在上一个检查点之后已经处理过的记录？
以我有两份工作为例，job1和job2。Job1使用来自Kafka的记录，对它们进行处理，并再次将它们生成第二个Kafka主题。Job2使用来自第二个主题的内容并处理记录(在我的例子中，它使用AerospikeClient更新了aerospike中的值)。
现在从这个问题的答案，我可以在某种程度上相信，如果job1重新启动，它将不会在接收器中产生重复的记录。我使用的是扩展TwoPhaseCommitSinkFunction ()的FlinkKafka
浏览 0提问于2021-03-08得票数 0
在Kafka中，如果启用了日志压缩，它将只存储最近的键值。如果我们尝试回复这些消息，它会只重放最新的消息吗？Kafka-reply究竟是如何工作的？
浏览 2提问于2018-05-12得票数 0
2回答
我正在尝试了解如何使用Kafka跟踪消息摄取。
我们现在遵循的工作流程是清除主题中的所有消息，然后我们重新摄取代码更改。我需要知道这些代码更改有多成功。在目前的状态下，我使用Kafka工具并手动刷新消息总数，并将结果保存在csv中，我知道这不是长期可持续的。
关于自动获取Kafka主题中的消息计数，您有什么建议？理想情况下，我想按分钟的频率命中该主题，并获得计数，以及时间窗口，如1天等。
*我不能使用KSQL，因为我们有稳定性问题。
浏览 16提问于2020-04-21得票数 1
2回答
我想为几个主题创建一个kafka消费者。consumer的方法构造函数允许我在订阅中传递主题列表的参数，如下所示：
private Consumer createConsumer() {
Properties props = getConsumerProps();
Consumer<String, byte[]> consumer = new KafkaConsumer<>(props);
ArrayList<String> topicMISL = new ArrayList<>();
for (String s:
浏览 0提问于2020-06-22得票数 2
我有点搞不懂这在概念上是如何运作的。
kafka流如何保证由kafka broker分配给它的分区与分配给其他主题的分区匹配？好像需要协调一下吗？而且，卡夫卡流总是从一开始就阅读压缩的主题，还是从最新的偏移量中读取？一旦它从压缩的主题中读取消息，它是否提交偏移量？
浏览 2提问于2020-07-31得票数 0
回答已采纳
4回答
在我读过的所有Kafka教程中，他们都提到"Kafka分区是不可变的“。但是，我也从这个站点上读到，Kafka会不时地删除分区中的旧消息(取决于您在log- that命令中设置的保留时间)。您可以从下面的屏幕截图中看到，在删除分区中的重复键之后，分区中的数据明显发生了变化：
所以我的问题是，说"Kafka分区是不变的“到底是什么意思？
浏览 8提问于2019-10-03得票数 1
回答已采纳
我在Scala中使用Spark在Kafka消费者应用程序中消费和处理消息。有时候，处理Kafka消息队列中的消息会比平时多花一点时间。此时，我需要消费最新的消息，忽略生产者已经发布但尚未消费的早期消息。
下面是我的消费者代码：
object KafkaSparkConsumer extends MessageProcessor {
def main(args: scala.Array[String]): Unit = {
val properties = readProperties()
val streamConf = new SparkConf().setMaster("
浏览 3提问于2017-08-31得票数 1
我有一个奇怪的问题，我不知道我该如何解决。我有一个kafka streams应用程序(2.1.0)，可以读取大约40个分区的主题。分区使用范围分区策略，因此目前其中一些分区可能是完全空的。
我的问题是，在应用程序停机期间，其中一个空分区被激活，并向其中写入了许多事件。当应用程序恢复时，它从其他分区读取所有事件，但它忽略了已经存储到先前空分区的事件(应用程序具有特定主题的最新OffsetResetPolicy )。最重要的是，当较新的消息到达特定分区时，它确实会使用它们，并以某种方式绕过以前的消息。
我的假设是__consumer_offsets在恢复时没有指定分区的任何条目，但是我如何避免这种
浏览 8提问于2019-10-31得票数 0
回答已采纳
如果Kafka (spring.cloud.stream.bindings..producer.partitionCount)大于1并且Kafka (spring.cloud.stream.bindings..consumer.partitioned)为false，会发生什么情况(使用partitionCount )
浏览 28提问于2021-05-08得票数 0
回答已采纳
我正在使用节点模块“kafka-节点”创建一个kafa消费者：
var client = new kafka.Client(ZOO_KEEPER_URL, PORTAL_CLIENT_ID);
var topics = [
{ "topic" : MY_TOPIC_NAME,
partition: 0
},
{ "topic" : MY_TOPIC_NAME,
partition: 1 }
];
var options = {
autoCommit: false,
浏览 1提问于2016-06-14得票数 0
回答已采纳
我有一个用例，其中我有两个消费者在不同的消费者组(cg1和cg2)中，订阅了具有4个分区的相同主题(主题A)。如果两个使用者从同一个分区读取数据，其中一个失败，而另一个提交了偏移量，会发生什么情况？
浏览 1提问于2020-03-06得票数 1
在设置Kafka Broker集群并创建几个主题之后，我们发现以下两个主题是由Kafka自动创建的：
__consumer_offsets
_schema
这些主题的重要性和用途是什么？
浏览 4提问于2016-09-16得票数 48
回答已采纳
同一个应用的多个实例拥有相同的消费者组id，监听Kafka topic会发生什么情况？ 两个实例读取的消息是相同的还是不同的？ 在消费者群体中有多个消费者...但是，当我们有多个具有相同I'd的消费者组时，会发生什么情况。
浏览 16提问于2019-05-25得票数 1
回答已采纳
我使用Kafka Streams和一个Transformer的实现来获取元数据，比如每个消息的分区和偏移量。我将此信息添加到一个列表中，并见证在某个时刻，偏移量会“倒退”(请注意，我groupByKey了，所以我假设偏移量总是在增加，并且我验证了每个消息的分区是29 )：
"offset": 441511,
"offset": 441512,
"offset": 441513,
"offset": 441514,
"offset": 441515,
"offs
浏览 2提问于2019-07-23得票数 1
1回答
我使用@StreamListener来使用Kafka消息。
我把autoCommitOffset设置为false，autoCommitOnError设置为false。
我将发送所有失败的消息给DLQ主题以及maxAttempt的失败。我在测试变更时有个问题。
如果我没有确认所使用的消息，也没有抛出任何错误，会发生什么情况？卡夫卡会在某个时候自动发送信息吗？
当我抛出错误时，重放就会重新开始，直到我的maxAttempt配置和失败的消息转到DLQ主题为止。
让我知道卡夫卡是否支持重试，如果消费者没有抛出任何错误和不承认的信息。
浏览 6提问于2020-02-01得票数 0
回答已采纳
我正在尝试设置一个Kafka Mirror机制，但是似乎Kafka MirrorMaker的消费者在Kafka集群中只在镜像生成器进程启动时从新传入的数据中读取主题，即它不读取以前保存在主题中的历史数据。
我使用Kafka的MirrorMaker类来实现：
/bin/kafka-run-class.sh kafka.tools.MirrorMaker --consumer.config consumer.config --num.streams 2 --producer.config producer.config --whitelist=".*"
要从Kafka源集群读取的c
浏览 0提问于2015-10-09得票数 4
我已经写了一个非常简单的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
我目前正在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读取数据但未能写入sink时会发生什么情况。比方说，我从kafka读取数据，并对数据进行了一些转换，最后将最终结果存储到数据库中。如果一切顺利，我的最终结果将存储在数据库中。但让我们说，由于某种原因，我的数据库是不可用的。我从kafka中读取的数据会发生什么？当我重新启动我的应用程序时，我能否再次读取相同的数据，因为我未能将它存储在接收器中？或者kafka会将这些数据标记为read，而不会允许我读取这些数据吗？
您还能告诉我这个属性用于什么- enable.auto.commit=true吗？
浏览 3提问于2020-05-08得票数 1
回答已采纳
我有一个Spring Kafka Consumer应用程序，它位于K8中。有时应用程序会被回收/重新启动。当消费者重新启动时，我希望它使用回收过程中产生的所有消息。我用auto.offset.rest=earliest进行了实验，结果和预期的一样，但是我注意到kafka的默认值是最新的。 如果我使用最早的，会有什么风险？在什么情况下我最早使用最新的v.s？我试图在这里找到一篇文章，通过一个场景来解释它，但大多数都是从一些文档中复制粘贴过来的，而不是现实生活中的例子。
浏览 45提问于2021-04-28得票数 2
1回答
分布式模式下Kafka Connect集群的Offset Management的行为是什么，即运行多个Connector并监听同一组主题(或一个主题)？ 因此，在分布式模式下，Kafka Connect将偏移量信息存储在Kafka中，该偏移量将被集群中的workers读取并提交。如果在该Kafka Connect集群中有多个Connector在监听同一主题，会发生什么情况？分区的偏移量是所有连接器的偏移量都相同，还是每个连接器在分区上的偏移量不同？
浏览 15提问于2019-07-01得票数 2
1回答
我是Kafka的新手，我正在使用Kafka 1.0。
我使用pull模式阅读Kafka消息，也就是说，我会定期poll() kafka主题以获取新消息，但我没有将偏移量写回Kafka。
我会问kafka如何知道我消耗了哪些偏移量，或者Kafka记住进度的机制是什么(Kafka偏移量)
浏览 9提问于2020-05-20得票数 1
回答已采纳
2回答
如果一些消息在kafka中没有被确认会发生什么？假设我正在并发地使用消息。还有一个消费者无法使用offset=20处理消息，也没有发回确认。但是使用offset=21的其他消息已被使用并返回。我怎么能只重放20次？
我是否需要将消息放到DLQ中并再次消费？如果那里也发生了故障怎么办？
我对保证交付有点迷惑。
浏览 0提问于2017-04-04得票数 3
我目前正在spark上运行一个流媒体应用程序，它使用新的直接方法(没有接收器)消费一个简单的kafka主题。由于我们在Kafka集群上遇到了一些问题，即使在Kafka上恢复正常后，流媒体应用程序也遇到了一些问题:一些作业正在处理几乎空的rdd，而另一些作业正在处理两倍大小的rdd。平均输入速率仍然与以前相同，但处理时间因此不规律。
我试着重启应用程序，但没有什么不同。我在Kafka和Spark两边都没有发现任何堆栈或线索。
有人能告诉我如何解决这类问题吗？或者至少到哪里去看一看，以了解发生了什么？
浏览 0提问于2016-05-18得票数 0
我想用NodeJS和Kafka构建一个API，它可以将一个偏移量和一个主题作为输入，并输出从偏移量开始的前10条消息。我在和中尝试了这种方法。
它们提供的使用者API允许使用来自特定偏移量的消息。我想停止消费的信息，一旦我已经阅读了大约10条消息。但是两个API调用都将继续获取消息，直到最后一条消息。我怎么能停止那样做呢？
这是我的编辑的完整代码
var Kafka = require('no-kafka');
var express = require("express");
var app = express();
var producer = new K
浏览 7提问于2016-04-28得票数 1
卡夫卡中的demo.incoming话题堆积如山。
我正在使用。每秒大约有70个请求通过Scrapy Cluster REST api (Producer)提交给Kafka。蜘蛛可以非常快地完成爬行，因为redis中的队列保持在一个非常低的数字，大多数时间少于10个。但demo.incoming中的消息数量每秒都在堆积。这是我用来查看Kafka中demo.incoming主题消息数的命令
kafka-run-class.sh kafka.tools.GetOffsetShell \
--broker-list localhost:9092,kafka-statefulset-2:9092,ka
浏览 1提问于2019-07-07得票数 1
我的问题是，关于卡夫卡：
我有两个程序与卡夫卡交换数据(一个是生产，另一个是阅读)。假设顾客崩溃了。生产者将继续发送消息，一段时间后，我们将重新启动客户。
对于我们当前的代理设置，下面是应该发生的事情：
崩溃发生在不到1天前(因为offsets.retention.minutes是1440)，将检索偏移量，并将处理等待的消息。
事故发生在1天多以前，客户的新冲销将被重新调整为最早的( auto.offset.reset是最早的)。问题是:如果一些消息已经被处理(在1天到7天之间)，它们将再次被处理，因为Kafka保存了7天的消息( (log.retention.hours是168)。
解决方案
浏览 4提问于2017-05-04得票数 0
1回答
我对kafka API有以下问题。我用以下命令设置我的客户：
<code>A0</code>
然后
<code>A1</code>
我想从Kafka中读取数据，从这些数据中更新数据库。但如果更新失败，我想重试，直到它工作。所以我想将DB事务应用到kafka，i.a。如果我的数据库事务是OK的，那么移动kafka指针，但如果它失败了，那么从相同的位置重试。
在我的代码中，
<code>A2</code>
没有像预期的那样工作，这意味着“如果kafka崩溃，那么从提交的位置重新启动”。但是，当我的DB事务失败时，即使我不使用c
浏览 31提问于2018-12-21得票数 1
回答已采纳
我有一个结构化的流媒体程序，它从Kafka主题A中读取数据，并进行一些处理，最后将数据放入目标Kafka主题。
注意:处理是在函数- convertToDictForEachBatch()中完成的，该函数称为the foreachBatch(convertToDictForEachBatch)
作为处理的一部分，它读取另一个Kafka主题(events_topic)，如果在上次读取之后有一个新记录，它会执行一些额外的处理--从BigQuery表重新加载数据，并将其保存。
df_stream = spark.readStream.format('kafka') \
浏览 1提问于2022-03-12得票数 0
回答已采纳
当我运行这个命令时，我得到了2个主题。我知道我创建了测试主题，但我看到了一个名为"__consumer_offsets“的附加主题。从名称上看，它意味着它与消费者偏移量有关，但它是如何使用的呢？
$ bin/kafka-topics.sh --list --zookeeper localhost:2181 __consumer_offsets test
$ bin/kafka-topics.sh --describe --zookeeper localhost:2181
Topic:__consumer_offsets PartitionCount:50 Re
浏览 5提问于2018-12-13得票数 3
1回答
我有一个spring引导kafka消费者，它使用来自某个主题的数据，并将其存储在数据库中，并在存储后确认它。它运行良好，但如果应用程序在使用记录后未能获得DB连接，则会发生问题，在本例中，我们没有发送确认，但直到或除非我们更改组id并重新启动使用者，否则消息从未被消耗过。
我的消费者如下所示
@KafkaListener(id = "${group.id}", topics = {"${kafka.edi.topic}"})
public void onMessage(ConsumerRecord record, Acknowledgment ackno
浏览 0提问于2019-03-25得票数 4
我们有主题，保留为7天(168小时)。当生产者发送消息时，就会实时地使用消息.一切都如期而至。然而，最近在一台生产服务器上，Devops意外地将时区从PST更改为EST，作为OS修补程序的一部分。
在Kafka服务器重新启动后，我们看到很少(不是全部，而是随机的)旧消息正在被消费者消耗。我们要求Devops将其更改为PST并重新启动。同样，这些旧信息也在本周末再次出现。
在较低的环境(开发、质量保证、阶段等)中，我们还没有看到这个问题。
卡夫卡版本: kafka_2.12-0.11.0.2
任何帮助都是非常感谢的。
添加更多信息..。最近，我们的CentOS进行了补丁更新，不知何故，管理员从P
浏览 0提问于2018-06-24得票数 3
1回答
我正在开发一个使用Kafka流和数据库的应用程序。
在我的应用程序中，我是手动管理Kafka偏移量，并且只在消息处理成功的情况下提交偏移量(即在处理完并更新到DB之后成功)。
但是，如果在更新DB之后，我的应用程序在提交之前就会下降，那么当它返回时，会由于未提交的偏移量而导致对DB的重复写入。
我想避免这些重复，同时仍然确保我正在处理每一条消息。做这件事的正确方法是什么？
编辑:我对DB的更新基本上使记录的计数器增加了一些值。因此合并语句不是一个选项。
浏览 3提问于2020-11-24得票数 1
回答已采纳
我在Kafka中设置了一个主题的时间间隔为7天，我正在从Kafka获取数据并将其存储在数据库中，但从过去5天我的数据库服务器宕机了，现在我必须从Kafka获取最近5天的消息并将其存储在数据库中。注:从过去5天开始，Kafka没有问题。
浏览 0提问于2017-09-05得票数 1
1回答
我正在使用快捷和卡夫卡节点构建一个简单的node.js API，在收到请求时返回来自请求的主题和使用者组的未读消息，然后关闭连接。我不需要也不希望消费者继续等待新的信息。
在kafka节点中，检查主题是否已经结束，如果是，关闭与代理的连接并退出应用程序，以防止新消息被读取的正确方法是什么？
这是我的consumer.js。它与kafka节点文档中给出的示例基本相同。
"use strict";
const kafka = require("kafka-node");
let topicName = "testTopic-01",
grou
浏览 1提问于2019-05-29得票数 0
回答已采纳
我正在考虑使用Kafka连接器和创建我自己的Kafka消费者/生产者来将一些数据从Kafka转移到Kafka，我看到了Kafka连接器在可扩展性和容错方面所提供的价值。但是，如果“任务”由于某种原因而失败，我无法找到连接器的确切行为。以下是几种场景：
对于接收器连接器(S3-Sink)，如果它(任务)成功地将数据发送到目标(例如由于网络问题)失败(在所有重试之后)，那么工作人员会发生什么情况？它会坠毁吗？它能够再消费同样的数据从Kafak以后吗？
对于源连接器(JDBC )，如果它不能发送给Kafka，它以后会重新处理相同的数据吗？这取决于来源是什么？
以上问题的答案是否取决于
浏览 0提问于2018-08-22得票数 1
我正在使用AWS Kafka MSK，我有一个有两个分区的主题。我也有两个消费者，他们是同一个消费者群体的一部分。
我想知道在以下情况下会发生这种情况：
消费者A-接受信息1-100
消费者B-接受信息101 - 200
消费者A失败
消费者B成功
信息1-100会发生什么变化？
卡夫卡汽车会让消费者B阅读1-100条信息吗？或者将启动而不是消费者A的新消费者将读取消息？
提前谢谢。
浏览 4提问于2022-02-13得票数 0
回答已采纳
我有一个小组，有两个主题。我试图将其中一个偏移量重置为1，但得到的消息是偏移量(1)低于主题分区的最早偏移量。
➜ local-kafka_2.12-2.0.0 bin/kafka-consumer-groups.sh --bootstrap-server myserver:1025 --group mygroup --topic mytopic --reset-offsets --to-offset 1 --dry-run
[2019-01-31 21:41:35,664] WARN New offset (1) is lower than earliest offset for topi
浏览 0提问于2019-02-01得票数 1
我正试图解析卡夫卡中__consumer_offsets主题中的日志。这个想法是为了找到在我的卡夫卡集群中创造负载的群体id、主题和消费者。
我正在执行的命令如下
bin/kafka-console-consumer.sh --topic __consumer_offsets --bootstrap-server brokers --formatter "kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter" --new-consumer --consumer.config consumer.
浏览 0提问于2018-09-28得票数 0
回答已采纳
我正在实现spring批处理侦听器，它读取kafka主题的消息列表并将数据发布到REST服务。我想了解在REST服务下降时的偏移管理，批处理的偏移量不应该提交，下一次轮询的消息应该被处理。我已经阅读了spring文档，但在理解侦听器错误处理程序和在批处理中寻求当前容器错误处理程序之间的区别方面存在混淆。我使用的是spring 2.0.0.M7及以下版本是我的代码。
Listener Config:
@Bean
KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>
浏览 2提问于2017-12-13得票数 0
我试图用最新的kafka_2.10-0.8.2.1使用低级的Consumer手动管理偏移量。为了验证我从Kafka提交/读取的偏移量是否正确，我使用了kafka.tools.ConsumerOffsetChecker工具。
下面是我的主题/消费者组的输出示例：
./bin/kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --group elastic_search_group --zookeeper localhost:2181 --topic my_log_topic
Group Topic
浏览 3提问于2015-05-20得票数 10
回答已采纳
我们从Confluent 4.0.0升级到5.0.0，升级后我们无法列出kafka connect hdfs接收器连接器消费者lag。
$ /opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server <hostname>:9092 --list | grep scribe_log_backend
Note: This will not show information about old Zookeeper-based consumers.
connect-kconn_scribe_log_backend
$ /opt/ka
浏览 0提问于2018-09-17得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
实时音视频
云直播活动推荐
腾讯云开发者
