未清除Kafka消息
是指在使用Kafka消息队列时，消费者没有正常消费和确认消息，导致消息在Kafka中一直保留未被删除的状态。这种情况可能会导致消息堆积、占用大量存储空间和资源，并对系统性能产生负面影响。
为了解决未清除Kafka消息的问题，可以采取以下措施：
- 检查消费者逻辑：首先要确认消费者逻辑是否正确。消费者应该正确订阅并消费消息，并在处理完毕后发送确认信息给Kafka，以便告知Kafka可以删除该消息。确保消费者正确处理消息并发送确认。
- 调整消费者组的配置：检查消费者组的配置，确认消费者组中的消费者数量、分区分配等是否合理。适当调整消费者组的配置，以确保消息能够被合理地分发给消费者。
- 增加消费者数量：如果消息消费速度较慢，可以考虑增加消费者数量，以提高消息处理能力。通过增加消费者数量来分摊消息处理的负载，从而减少消息堆积的可能性。
- 定期提交消费位移：及时提交消费位移，可以确保在消费者出现故障或重启后，能够继续从上次消费的位置开始消费，避免消息的重复消费。
总之，未清除Kafka消息可能会导致消息堆积和性能下降，为了解决这个问题，需要保证消费者逻辑正确、消费者组配置合理、消费者数量足够，并且定期提交消费位移。腾讯云提供了Kafka相关的产品TDMQ（消息队列 TDMQ），可用于消息的高性能传输和分发，适用于解决实时计算、日志处理、流式数据处理等场景。
参考链接：腾讯云TDMQ产品介绍
相关·内容
我完全意识到卡夫卡是一根木头，就像哑巴。但是在我们的应用程序中，我们有一些“遵从性主题”，我们需要确保读取其中的所有消息。
由于Kafka内部处理所有消费者的抵消(因此可以监视消费者延迟)，我想知道是否可以设置log.retention=25GB，但不要覆盖没有被任何人读取的消息，而是抛出一个生产者异常(就好像我们设置了log.retention=1week和HDD已经满了一样)。
如果无法在Kafka级别上配置这种逻辑，那么在kafka-生产者或kafka流上如何实现这一逻辑，从而在“完全未读日志”情况下失败吗？
浏览 0提问于2018-04-04得票数 1
回答已采纳
我使用的是Kafka 0.8.1。
我正在使用Kafka在我们的处理集群中分发我们的事件。
我正在使用一个带有多个分区的主题。
我担心的是卡夫卡的再平衡过程。目前我们的处理逻辑发生在运行Kafka消费者的机器上，所以我想知道在重新平衡的情况下会发生什么？
避免处理分区的新节点和旧节点之间的竞争的最佳实践是什么？
处理结果最终被持久存储在一些存储中，我想知道如何避免旧的消费者覆盖新的消费者更新。
如果我使用auto.commit false，当有未提交的消息时，Kafka是否有一些保证不会重新平衡。
谢谢。
浏览 2提问于2015-11-24得票数 1
我有一个flink作业，它从一个Kafka主题读取数据，丰富数据，并将其发送到输出Kafka主题。
从我的来源kafka，我得到了一致的3MB和200条消息每秒。但有时我可以得到峰值70毫巴和9k信息每秒钟。
当出现这样的高峰时，我的输出主题配额无法处理发送的消息数量。所以我会百分之一百的忙在我的水槽上，这会造成背压，导致检查点的失效，这就导致了工作的失败。在那里，作业不会成功地重新启动。
flink是否可以在不导致作业重新启动的情况下处理此类数据的峰值。(比如放慢对源主题的读取速度)
浏览 9提问于2022-09-01得票数 1
回答已采纳
1回答
我正在执行卡夫卡的ConsumerGroupExample消费者，它挂着等待消息
它似乎正在等待消息(例如，在，在"for ( KafkaStream流: streams)“处等待。
配置详细信息：(a)在VMPlayer Ubuntu操作系统中运行；(b)运行kafka版本: 0.8.0；(c)运行相同VM上的动物园管理员、Kafka、使用者和生产者(ubuntu/localhost)；(d)使用与此版本的kafka捆绑在一起的动物园管理员；(e)运行生产者示例(未更改，可在cwiki.apache.org/confluence/display/kafka/)获得；(f)运行消费群示
浏览 1提问于2014-01-19得票数 4
回答已采纳
2回答
我们想知道对于给定的Kafka主题，还没有被Kafka Consumer消费和确认的消息数量。
有没有办法在Java中获得给定主题(主题有10个分区)中未消费的Kafka消息的计数?？
浏览 3提问于2019-10-24得票数 3
我们有一个需求，当我们使用来自一个主题的消息时，会有一些充实发生，然后我们将消息发布到另一个主题。下面是一些事件
消费者-消费messageEnrichment -丰富了已消费的messageProducer -发布了其他主题的丰富信息。
我正在使用活页夹，一切都很好。最近，我们引入了幂等生成器并包含了transactionIdPrefix属性，并且我们观察到出站通道在主题中开始发送2条消息，因为它应该只发送一条消息。一条带有实际json的消息将另一条消息的值计算为‘b’\x00\x06&‘字符。下面是代码和配置。如果删除transactionIdPrefix，那么我可以看到只发送给出
浏览 3提问于2021-10-01得票数 1
1回答
我正在使用Kafka流，我注意到它使我的kafka日志记录了很多日志消息，例如：
[2019-04-17 09:06:16,541] INFO [Log partition=my-application-KSTREAM-AGGREGATE-STATE-STORE-0000000076-repartition-0, dir=/opt/kafka/data/logs] Incrementing log start offset to 316423 (kafka.log.Log) [2019-04-17 09:06:16,545] INFO [Log partition=my-application
浏览 1提问于2019-04-17得票数 3
我有一个简单的Spark结构化流作业，它使用Kafka0.10API从Kafka读取数据并写入我们的S3存储。从日志中我可以看到，对于触发的每一批，流应用程序正在取得进展，并且正在消耗来自源的数据，因为endOffset大于startOffset，而且每一批都在增加。但是numInputRows始终为零，并且没有写入S3的行。
为什么抵消量逐渐增加，但火花批次却没有消耗任何数据？
19/09/10 15:55:01 INFO MicroBatchExecution: Streaming query made progress: {
"id" : "90f21e5f-
浏览 0提问于2019-09-10得票数 0
回答已采纳
我对Kafka和Spark都是新手，正在尝试写一个作业(流或批处理)。我想从Kafka读取预定义数量的消息(比如x)，通过workers处理收集，然后只开始处理下一组x消息。基本上，在Kafka中的每个消息都是10KB，我想将价值2 GB的消息放在一个S3文件中。那么，有没有办法指定接收方获取的消息数量呢？我读到我可以在创建DStream时指定'from offset‘，但是这个用例有点不同。我需要能够指定‘从偏移量’和‘到偏移量’。
浏览 1提问于2015-11-25得票数 0
1回答
当恢复所有主题的分区时，我不断地收到调试消息。如下所示。这条消息在我的服务器上每毫秒连续打印一次。
08:44:34.850 [default-akka.kafka.default-dispatcher-10] DEBUG o.a.k.clients.consumer.KafkaConsumer - Resuming partition test222-7
08:44:34.850 [default-akka.kafka.default-dispatcher-10] DEBUG o.a.k.clients.consumer.KafkaConsumer - Resuming partition
浏览 0提问于2016-12-07得票数 5
这可能是数据丢失的情况吗？如果由于不可纠正硬件问题，kafka的请求队列排队，如果这一次，我们关闭/退回kafka代理，那么追随者将会发生什么？队列中的消息会发生什么？ kafka.network:type=RequestChannel,name=RequestQueueSize
Size of the request queue. A congested request queue will not be able to process incoming or outgoing requests 根据我从kafka那里学到的，这应该在networklayer中，这是否意味着队列中的消息将被
浏览 45提问于2021-08-25得票数 0
使用从只有少数消息的主题中读取消息。卡夫卡一遍又一遍地重复队列中的信息。
它接收来自我的主题的一条消息(返回到每个消息内容)，然后抛出ERROR - Heartbeat session expired - marking coordinator dead并继续遍历其余的消息，并继续重播它们。更多的日志：
kafka.coordinator - ERROR - Heartbeat session expired - marking coordinator dead
kafka.coordinator - WARNING - Marking the coordinator dead (node 1
浏览 15提问于2017-09-29得票数 2
1回答
我对kafka，docker和kubernetes都是新手。我尝试在minikube中运行单节点或多节点kafka集群，并使用.net控制台应用程序向/从kafka broker产生/消费消息。
有没有人能给我指点一下.net应用程序的解决方案？使用它，我可以创建一个简单的POC，它可以向kafka经纪人发送消息/从kafka经纪人接收消息？
我花了太多的时间来使用和融合的简单生产者/消费者代码来实现这一点，但我只能使用minikube集群中的主题来访问，在集群之外，kafka代理是不可见的。我为同一个找到了另一个解决方案，但是因为我不熟悉kubernetes/networking和yaml，
浏览 0提问于2018-03-29得票数 2
我们有一个spring引导应用程序，它使用来自IBM的消息，进行一些转换，并将结果发布到Kafka主题。为此，我们使用。我知道Kafka不支持XA；但是，在文档中，我发现了一些关于使用ChainedKafkaTransactionManager链接多个事务管理器和同步事务的输入。同一文档还提供了一个示例，说明如何同步Kafka和数据库，同时读取Kafka的消息并将它们存储在数据库中。
在我的示例中，我遵循同样的示例，并将JmsTransactionManager与KafkaTransactionManager链接到ChainedKafkaTransactionManager的伞下。bean定义
浏览 3提问于2020-05-21得票数 0
回答已采纳
我正在尝试使用Kerberos从ActiveMQ桥向Kafka (Cloudera)生成消息。
ActiveMQ v5.15.4
Camel：2.21.1
Kafka客户端：1.1.0
服务器版本：Apache2.4.6 (CentOS)
camel.xml代码片段是：
<log message="Started The Producer Route" />
<to uri="kafka://10.100.70.00:9092?topic=MyEvents.s1.v1&brokers=10.100.70.00:9
浏览 0提问于2018-06-22得票数 2
1回答
卡夫卡分区和偏移消失了
、、、
我的Kafka客户端正在GCP环境中运行，启用了自动缩放功能(GCP将实例计数保持在至少两个，并且由于CPU使用率低，它大多是2个)。在这两个VM中运行的使用者组几个月来一直在使用来自20个分区中的各种主题的消息，最近我注意到旧主题中的分区缩小到只有1 (!)该消费者组的偏移量被重置为0。主题分区目录也从kafka-log目录中删除。奇怪的是，最近创建的主题分区是完整的。我有3个不同的环境(都在GCP中)，这发生在所有三个环境中。我们没有看到任何丢失的消息或数据问题，但我们想了解发生了什么，以避免再次发生这种情况。
kafka代理和动物园管理员在同一个GCP计算引擎实例中运行(我知道这不是最佳
浏览 2提问于2017-12-15得票数 1
我需要记录到达Kafka使用者的消息，因为对于Kafka消费者，没有Avro模式兼容性。
我怎样才能做到这一点？
我想，默认情况下，Avro不兼容的消息永远不会到达我的Kafka消费者代码。我使用Spring的@KafkaListener。
浏览 11提问于2021-12-07得票数 0
1回答
当我检查使用者延迟时，它会显示某个特定的consumer-id是从某个主题的特定主机消费中运行的。但是当我去那个主机时，没有这样的消费者在运行。
如何杀死此consumer-id，以便可以为其所属的组重置使用者偏移量。
Kafka server version: 0.11.0.1
Kafka client version(stuck): 0.10.0.2
这个consumer-id被卡在了第一位，因为它无法使用消息，因为有些消息在Kafka中有标题。
我已经尝试过以下几种方法：
从不同的主机和不同的卡夫卡版本消费，它消费的信息，但消费者id，主机不会改变。
重新启动kafka代理，它
浏览 0提问于2019-02-18得票数 3
2回答
我想知道关于Kafka如何维护消息序列的信息/解释，当消息被写入多个分区的topic时。例如，我有多个消息生成器，每个消息生成器都顺序地生成消息，并使用超过1个分区在Kafka主题上进行编写。在这种情况下，消费者组将如何工作来消费消息。
浏览 0提问于2016-12-01得票数 2
我使用下面的命令来清除主题中的消息。但这并不能立即删除所有消息，我仍然可以看到消息。
kafka-configs --zookeeper localhost:2181 --alter --entity-type topics --entity-name topic1 --add-config retention.ms=1000,cleanup.policy=delete
在运行此cmd并等待几分钟后，我删除了配置。
kafka-configs --zookeeper localhost:2181 --alter --entity-type topics --entity-name topic1
浏览 0提问于2019-07-31得票数 0
D:\Softwares\kafka\kafka_2.10-0.10.0.1\bin\windows>kafka-console-consumer.bat --zookeeper本地主机:2181--主题样本
我使用上面的命令来消费消息，有什么我遗漏的吗？帮我一下：
这
它们是生产者和消费者……
浏览 0提问于2016-09-29得票数 0
1回答
我正在尝试将asyncio功能集成到我的kafka主题侦听器中，并且遇到了一些问题(在python中异步编程非常新)。
我创建了一个confluent-kafka consumer，用于收听一个主题。主题经常有消息，性能是最重要的(因此引入了异步io)。
main()函数如下所示：
def main(self):
while True:
try:
msg = consumer.poll(timeout=5.0)
if msg is None:
continue
as
浏览 1提问于2021-03-30得票数 0
1回答
在带有Ubuntu18.04的 (WSL)中安装 kafka /ZooKeepv2.4.1(二进制kafka_2.4.1.tgz)的集群/实例中，卡夫卡代理在清理日志文件时意外地关闭了以下错误消息：
ERROR Failed to clean up log for __consumer_offsets-11 in dir /tmp/kafka-logs due to IOException (kafka.server.LogDirFailureChannel)
java.io.IOException: Invalid argument
at java.io.RandomAccessFi
浏览 0提问于2020-04-10得票数 2
Spring Cloud Contract是否支持Spring Kafka消息开箱即用的contract验证？
浏览 1提问于2017-02-16得票数 1
我想将主题mytopic的偏移量设置为1分区，并将组ID testgroup1设置为0。然而，这并不总是可能的。如果要将偏移量设置为0，则会收到以下消息：
bash-4.4# kafka-consumer-groups.sh --bootstrap-server localhost:9092 --topic mytopic --group testgroup1 --reset-offsets --to-offset 0 --execute
[2021-06-04 09:23:30,854] WARN New offset (0) is lower than earliest offset fo
浏览 2提问于2021-06-04得票数 2
我想要删除我的GlobalKTable<Integer, Long>存储的状态。
我试图通过以下方式移除这个州：
通过调用kafka-topics.sh --zookeeper localhost:8080 --delete --topic my-topic删除整个Kafka主题
在应用程序启动时运行KafkaStreams.cleanUp()
向我的流生成'Test: null‘消息，因为应该将空值视为存储中的DELETE语句，正如所描述的。但是，我的流应用程序失败了，因为空值不能反序列化为LONG。
请参阅下列例外情况：
org.apache.kaf
浏览 0提问于2018-03-09得票数 2
25回答
我把一条太大的信息推到了本地机器上的kafka消息主题中，现在我收到了一个错误：
kafka.common.InvalidMessageSizeException: invalid message size
在这里增加fetch.size并不理想，因为我实际上不想接受那么大的消息。
浏览 25提问于2013-04-29得票数 250
回答已采纳
2回答
我试着让我的卡夫卡制作人做交易。我发送10条信息，.If，任何错误发生，不应该发送给卡夫卡，即没有或全部。
我正在使用Spring。
@Configuration
@EnableKafka
public class KakfaConfiguration {
@Bean
public ProducerFactory<String, String> producerFactory() {
Map<String, Object> config = new HashMap<>();
// props.put(Com
浏览 31提问于2020-04-30得票数 1
回答已采纳
1回答
我们如何在Kafka主题中使用来自多个分区或单个分区的自定义消息？
当我在卡夫卡主题中使用来自多个分区的消息时，如何只读取1条消息？
浏览 1提问于2018-07-06得票数 0
嗨，我们一直在使用旧的spring版本和kafka 1.1，有以下的依赖项
+--- org.springframework.boot:spring-boot-starter-web: -> 1.5.4.RELEASE (*)
+--- org.springframework.cloud:spring-cloud-starter-config: -> 1.3.1.RELEASE (*)
+--- org.springframework.cloud:spring-cloud-starter-bus-kafka: -> 1.3.1.RELEASE (*)
+--- org.spr
浏览 2提问于2020-05-20得票数 0
1回答
在阅读了Kafka幂等操作和事务之后，我想看看Kafka源连接器是否使用了它们中的任何一个，以确保只有一次语义。 经过一点阅读，我发现源连接器使用了自己的某种“偏移量”，这让它存储了它所做的最后一次操作，然后我看到 用于MongoDB的Debezium源连接器 只能实现至少一次语义。 所以我的问题是，为什么MongoDB源连接器不能保证只有一次语义？如果连接器可以将其偏移量与事务中的每条消息一起提交，它就可以确保消息和偏移量都在Kafka中，从而保证只有一次语义。
浏览 11提问于2020-01-17得票数 2
回答已采纳
我正试着阅读卡夫卡主题的信息。邮件格式如下(示例格式)：
{"schema":{"type":"struct","name":"emp_table","fields":[{"field":"emp_id","type":"string"},{"field":"emp_name","type":"String"},{"field":"city"
浏览 1提问于2019-01-04得票数 2
回答已采纳
我使用Debezium MySql连接器将数据库更改事件从MySql数据库流式传输到Kafka。我需要将特定的转换应用于某些指定表(而不是其他表)中的记录。有没有一种只使用Kafka Connect单消息转换(没有Kafka Streams或KSQL)的方法？我甚至不介意写一个自定义的SMT。但我找不到如何为消息指定要应用SMT的表/主题。谢谢。 在简短的调查之后，我决定使用KSQL。感谢大家的回答
浏览 10提问于2019-04-11得票数 1
Java: OpenJdk 11 Kafka: 2.2.0Kafka流库: 2.3.0
我正在尝试将我的Kafka应用程序部署到停靠器容器中，但是它在尝试使用TopicAuthorizationException创建内部状态存储时失败了。在当地效果很好。本地和服务器之间的主要区别是，它连接到部署了Kafka的服务器，并使用通常的Kerberos auth进行身份验证。我无法理解身份验证与本地存储之间的联系。
我的小溪看起来是这样的：
StreamsBuilder builder = new StreamsBuilder();
//We stream from the sour
浏览 2提问于2019-10-09得票数 8
回答已采纳
我用Kafka写了一些集成测试。我用@DirtiesContext标记了它们，但当我一起运行所有测试时，其中一些测试失败了，导致kafka消息在并行运行时最终通过测试，然后它们失败了。
有没有办法在一个完全独立的上下文中运行每个测试，或者只是一个接一个地按顺序运行它们？
我正在尝试运行: mvn测试
浏览 16提问于2021-05-19得票数 0
我在Ubuntu上安装了Kafka，并尝试使用shell文件测试一个简单的场景：
启动了默认配置，启动了一个节点或代理，使用默认配置创建了一个主题，为其命名，单个分区和复制因子为1，并将其与动物园管理员默认地址关联，打开了一个生产者和一个使用者：
bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic mytopicname
其次是：
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic mytopicname --f
浏览 0提问于2020-08-18得票数 1
回答已采纳
火花流中有spark.streaming.kafka.maxRatePerPartition属性，它限制每秒从Apache读取消息的数量。斯托姆也有类似的财产吗？
浏览 2提问于2016-06-10得票数 0
回答已采纳
我正在使用星火结构流(2.3.0)与Kafka (1.0.0)。
val event_stream: DataStreamReader = spark
.readStream
.format(_source)
.option("kafka.bootstrap.servers", _brokers)
.option("subscribe", topic)
.option("startingOffsets", "earliest")
.option("failOnDataLoss", "
浏览 1提问于2018-08-22得票数 1
回答已采纳
我使用Kafka Streams 2.3.1 suppress()操作符来限制发送到底层KTable的更新数量。
这里的用例是，在我的处理逻辑中，我想要进行一个HTTP调用，然而，为了限制调用的数量，我对流进行了窗口操作，并将落入同一时间窗口的源主题消息聚合在一起，以进行单个API调用。
代码大致如下所示
KTable<Windowed<String>, List<Event>> windowedEventKTable = inputKStream
.groupByKey()
.windowedBy(TimeWindows.of(Durati
浏览 0提问于2019-12-04得票数 0
我们的Apache拓扑使用KafkaSpout侦听来自Kafka的消息，在进行了大量映射/还原/浓缩/聚合等操作之后，最后将数据插入Cassandra。还有一个kafka输入，我们接收用户查询数据，如果拓扑找到一个响应，然后它发送到第三个kafka主题。现在我们想要使用Junit编写E2E测试，在这个测试中我们可以直接编程地将数据插入到拓扑中，然后通过插入用户查询消息，我们可以在第三点断言在我们的查询上收到的响应是正确的。
为了实现这一点，我们考虑启动EmbeddedKafka和CassandraUnit，然后用它们替换实际的Kafka和Cassandra，然后我们可以在这个单独的Junit测
浏览 2提问于2019-07-12得票数 0
回答已采纳
我正在尝试在Spark Structured和Hazelcast Jet之间映射一些概念，并理解其他主题。
Q1 -在Spark中，每个Kafka分区将成为spark中的一个分区，然后它们将由单个任务并行处理。我想我在某处读到过Hazelcast Jet将合并来自kafka的所有消息，而不考虑group.id和主题分区，对吗？
Q2 -我们如何在Jet程序中增加“消费者”的数量，以增加来自kafka的吞吐量消耗？在Spark中，我想我们只需要增加主题分区的数量，以便为新分区分配新的spark任务。
Q3 -如果上面的Q1是真的，有可能避免合并和分发要并行处理的kafka分区吗？一旦消息已经在k
浏览 0提问于2018-01-23得票数 2
1回答
如何处理卡夫卡中的大消息，比如超过20 to的消息等等。
2009-03- 13 08:59:10,923错误当发送消息到主题测试时，键:13字节，值: 11947696字节与错误：(org.apache.kafka.clients.producer.internals.ErrorLoggingCallback) org.apache.kafka.common.errors.RecordTooLargeException:请求包含一条大于服务器将接受的最大消息大小的消息。
2019-03-1303:59:14,478在用键向主题测试发送消息时出错: 13字节，值: 11947696字节带有错误
浏览 0提问于2019-03-13得票数 1
回答已采纳
我使用的是一个主题，一个分区，一个消费者，Kafka客户端版本是0.10。
我得到了两个不同的结果：
如果我先进行分区，然后生成消息并调用方法。KafkaConsumer可以成功地轮询未提交的消息。
但是，如果我首先生成消息，而没有提交它的偏移量，那么在几秒钟后暂停分区，以调用恢复方法。KafkaConsumer将不会接收未提交的消息。我使用kafka-consumer-groups.sh在Kafka服务器上检查了它，它显示了LOG-END-OFFSET minus CURRENT-OFFSET = LAG = 1。
我已经想了两天了，我反复做了很多次这样的测试，结果总是这样。
浏览 0提问于2018-11-12得票数 0
回答已采纳
下载合流开源3.3之后，在命令行进行一个简单的实验：
confluent start -它将显示所有服务而不会出现任何明显的错误：
confluent-3.3.0% confluent start
Starting zookeeper
zookeeper is [UP]
Starting kafka
kafka is [UP]
Starting schema-registry
schema-registry is [UP]
Starting kafka-rest
kafka-rest is [UP]
Starting connect
connect is [UP]
向avro控制台生成器发
浏览 0提问于2018-06-28得票数 0
回答已采纳
1回答
我的问题是如何使用Spring Cloud Stream Kafka管理多实例。 让我解释一下，在Spring Cloud Stream微服务上下文(eureka，configserver，kafka)中，我希望拥有相同微服务的两个实例。当我在GIT Repository中更改配置时，configserver (通过webhook)会将一条消息推送到Kafka主题中。 如果我在我的微服务中使用相同的group-id，那么两个实例中只有一个会收到通知，并重新加载他的spring上下文。但我需要刷新所有实例... 因此，为了做到这一点，我配置了一个唯一的组id：${spring.applicat
浏览 44提问于2020-07-17得票数 2
回答已采纳
当我试图连接到我的Kafka代理时，我看到了以下错误消息：
2020-08-10 15:59:54.974 ERROR 31447 --- [ scheduling-1] c.m.f.adapter.KafkaMessageListener : Failed to consume message
java.lang.IllegalStateException: There are no in-flight requests for node -1
at org.apache.kafka.clients.InFlightRequests.requestQueue(InF
浏览 4提问于2020-08-10得票数 2
回答已采纳
1回答
在没有任何自定义配置的情况下在DCOS上安装Kafka会给出以下日志
集群配置是：
8个专用节点，2个公共节点
24核总计320 GB Ram总计
所有资源都是免费的。
StdErr输出：
log4j:WARN No appenders could be found for logger (org.jboss.logging).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for mor
浏览 0提问于2016-10-13得票数 4
回答已采纳
1回答
我对停止ConcurrentMessageListenerContainer后(Spring) Kafka的民意测验()行为感到有点困惑。
我想要达到的目标是:在引发异常后停止使用者(例如，无法将消息保存到数据库)，不要提交偏移量，在给定时间后重新启动它，然后从先前失败的消息中再次开始处理。
我阅读了这篇文章，其中说容器将使用来自轮询的其余记录()调用侦听器，这意味着无法保证在失败消息之后，成功处理的另一条消息将提交偏移量。这可能会以丢失/跳过的消息结束。
真的是这样吗?如果是的话，在不升级更新版本的情况下，有没有解决这个问题的解决方案？(DLQ不是我的案子的解决方案)
我已经做了什么:设置s
浏览 0提问于2018-02-20得票数 6
回答已采纳
我试图设置使用者组属性，使其只能每分钟使用一条消息。
consumer.properties
group.id=test_group
max.poll.interval.ms=60000
session.timeout.ms=60000
max.poll.records=1
生产者
kafka-console-producer.sh --broker-list localhost:6667 --security-protocol SASL_PLAINTEXT --topic cdc_poc
Consumer
kafka-console-consumer.sh --bootstrap-serv
浏览 2提问于2019-03-01得票数 1
回答已采纳
我有一个卡夫卡Avro主题生成使用KafkaAvroSerializer。
我的独立属性如下。
我使用Confluent 4.0.0来运行Kafka connect。
key.converter=io.confluent.connect.avro.AvroConverter
value.converter=io.confluent.connect.avro.AvroConverter
key.converter.schema.registry.url=<schema_registry_hostname>:8081
value.converter.schema.registry.ur
浏览 0提问于2018-06-28得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
