kafka服务器重启后丢失未处理的消息
Kafka是一种分布式流处理平台，它具有高吞吐量、可扩展性和持久性的特点。当Kafka服务器重启后，未处理的消息可能会丢失，这是因为Kafka的消息存储机制。
Kafka使用一种称为日志的持久化机制来存储消息。在Kafka中，消息被追加到一个或多个分区的日志中，并且每个分区都有一个唯一的偏移量来标识消息的位置。当消费者从Kafka中读取消息时，它可以指定从哪个偏移量开始读取。
当Kafka服务器重启后，它会尝试从上次关闭时的偏移量继续读取消息。然而，如果消息尚未被完全处理或提交到消费者的外部系统中，这些未处理的消息可能会丢失。这是因为Kafka只保证已提交的消息不会丢失，而未提交的消息在服务器重启后可能会丢失。
为了解决这个问题，可以采取以下措施：
- 使用Kafka的高级消费者API：Kafka提供了高级消费者API，它可以跟踪每个消费者组的偏移量，并在消费者组中的消费者发生故障时重新平衡分区。这样，当服务器重启后，消费者可以从上次提交的偏移量继续读取消息，从而避免丢失未处理的消息。
- 设置适当的消息提交策略：在消费者处理完消息后，可以选择手动提交偏移量或使用自动提交偏移量的方式。手动提交偏移量可以确保消息被完全处理后再提交，而自动提交偏移量可能会导致部分消息丢失。因此，根据业务需求，选择适当的提交策略来避免消息丢失。
- 使用Kafka的复制机制：Kafka支持分布式部署，并具有复制机制来提供高可用性和容错性。通过将消息复制到多个副本中，即使某个服务器重启，仍然可以从其他副本中读取未处理的消息，从而避免消息丢失。
总结起来，为了避免Kafka服务器重启后丢失未处理的消息，可以使用Kafka的高级消费者API、适当的消息提交策略和复制机制来确保消息的可靠性和持久性。
腾讯云提供了一系列与Kafka相关的产品和服务，例如TDMQ（消息队列服务）、CKafka（分布式消息队列服务）等。您可以访问腾讯云官方网站获取更多关于这些产品的详细信息和介绍。
- TDMQ产品介绍：https://cloud.tencent.com/product/tdmq
- CKafka产品介绍：https://cloud.tencent.com/product/ckafka
相关·内容
我们有Kafka设置，以便能够通过多个服务器并行处理消息。但每条消息必须只处理一次(并且只由一台服务器处理)。我们已经启动并运行了它，它工作得很好。
现在，我们面临的问题是，Kafka的消费者为了获得最大的效率而批量阅读消息。这会导致一个问题，如果/当处理失败时，服务器关闭或其他什么，因为我们丢失了即将处理的数据。
有没有办法让消费者一次只读消息，让Kafka保留未处理的消息？例如；Consumer拉出一条消息-> process -> commit offset完成后，重复。使用Kafka可行吗？有什么想法/想法吗？
谢谢!
浏览 0提问于2015-08-17得票数 13
1回答
我在kubernetes中部署了一个go应用程序。我的应用程序有一个go例程，它不断地监听来自kafka主题的消息，并将新消息推入go通道进行进一步处理，并提交kafka消息。如果我使用hpa缩小副本的数量/如果副本由于恐慌而下降，那么仍然处于未处理的go通道中的消息会发生什么情况？
浏览 4提问于2022-02-10得票数 -3
我已经写了一个非常简单的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
rabbitmq服务器或集群重启后，所有队列都恢复了所有消息，即使消息已经确认(从rabbitmq服务器启动时开始)，并重新处理所有消息。
队列详细信息
根据我的理解，在消息参数中将persistenet设置为false，如果代理重新启动，消息将不会继续存在。此外，我还将队列的durabele设置为false。
我是否遗漏了任何其他设置？
浏览 1提问于2018-07-10得票数 0
例如，我的kafka消费者(在consumer Group 1中)正在阅读来自Kafka主题A的消息。
现在，如果该消费者在失败之前消费了12条消息。
当消费者重新启动，并且现在它有不同的消费者组(即消费者组2)时，问题1 -？在重启时，它会从偏移量(或位置)中停止的地方继续，因为该偏移量是由Kafka和/或ZooKeeper存储的，还是会从第一条消息开始消费消息。
问题2->有没有办法确保在重启时(当消费者有不同的消费群体时)，它仍然从重启之前停止的地方开始消费？
为了给你提供上下文，我正在尝试在收到关于kafka主题的消息时更新每个节点/服务器中的内存缓存。为此，我对每个节点/服务器
浏览 1提问于2021-02-04得票数 0
在topic级别更改保留时间后，是否需要重启Kafka服务器。 我已经更改了主题级别的Kafka保留时间。那么，要反映的变化，是否需要重启kafka服务
浏览 53提问于2021-09-13得票数 1
在我们的生产环境中，我们经常看到分区在使用来自主题的消息时被复制。我们使用的是Kafka 0.11。从文档中可以理解的是
删除配置参数replica.lag.max.messages。分区领导者在决定哪些副本同步时不再考虑滞后消息的数量。
配置参数replica.lag.time.max.ms现在不仅是指上次从副本获取请求所经过的时间，也是指自副本上次捕获以来的时间。那些仍然从领导者那里获取消息但没有赶上replica.lag.time.max.ms中最新消息的副本将被认为是不同步的。
我们如何解决这个问题？复制不同步的不同原因是什么？在我们的场景中，所有的Kafka代理都位于刀片服务器的单个
浏览 0提问于2018-07-24得票数 11
回答已采纳
2回答
我正在使用hortonwork Sandbox for kafka服务器，试图用java代码连接eclipse中的kafka。使用此配置连接到生产者以发送消息
metadata.broker.list=sandbox.hortonworks.com:45000
serializer.class=kafka.serializer.DefaultEncoder
zk.connect=sandbox.hortonworks.com:2181
request.required.acks=0
producer.type=sync
其中sandbox.hortonworks.com是我连接的沙盒名称
在k
浏览 1提问于2015-03-20得票数 4
我设置的and服务器是为了开发和测试目的，严格的本地，不能从Internet访问，但需要尽可能符合生产环境，包括SSL。因此，我在服务器上设置了一个本地CA，使用SubjectAltNames为它们上运行的几个节点创建了一个服务器请求，并使用刚刚创建的CA对其进行了签名，并指示apache使用此设置。 作为服务器的两个CA中的CN是tdc.nl -这是我在本地使用的‘域’。它有站点manage.tdc.nl和www.tdc.nl，所以在服务器请求中，我定义了： [ v3-req ]
...
subjectAltNames = @alt_names
...
[ alt_names ]
DNS
浏览 55提问于2020-01-14得票数 0
1回答
我有4个卡夫卡运行与debezium。经过几天的良好运行，三台kafka机器在一段时间内脱离了网络，在connectDistributed.out日志文件中，我收到了许多消息，其中包含以下错误：
[2020-05-04 13:27:02,526] WARN [Consumer clientId=connector-consumer-sink-warehouse-6,
groupId=connect-sink-warehouse] 133 partitions have leader brokers without a matching listener,
including [SCT010
浏览 26提问于2020-05-04得票数 0
回答已采纳
我在比较Apache Kafka和ActiveMQ (以及类似的消息队列产品)时发现的信息从来都不清楚每个解决方案的完整性属性(特别是一致性)。
使用Kafka，您可以获得即使在出现故障时也不会丢失消息的保证。使用"LazyPersistence“选项会失去这种保证吗？
所谓“无损失”，我的意思是消息将对客户端可用，即使在重启后出现故障-理想情况下，所有到达客户端的消息都以正确的顺序。
ActiveMQ ( "classic“或Artemis)是否保证失败时不会丢失消息？有没有什么配置选项可以保证这一点？如果"classic“和Artemis的答案不同，那就太好了。
浏览 3提问于2020-08-23得票数 0
在kafka的spark streaming直接方法中，有一种方法可以让我知道kafka偏移级别范围。但是，如果我想定期维护偏移量，以便在需要时可以重新处理偏移量中的项。当我处理每条消息时，有没有办法在rdd中检索消息的偏移量？例如，使用偏移量范围，我有RDD的开始和结束偏移量，但如果在处理RDD系统的每个记录时遇到错误和作业结束，该怎么办？现在，如果我想从失败的记录开始处理，我如何首先保存上一次成功的偏移量，以便下次启动时可以从该偏移量开始。
浏览 1提问于2016-05-24得票数 0
我有一个包含以下内容的docker-compose.yml文件：
services:
kafka_listener:
build: .
command: bundle exec ./kafka foreground
restart: always
# other services
然后我从容器开始：docker-compose up -d
在我的亚马逊实例上，kafka-server (例如)有时无法启动，因此./kafka foregound脚本失败。当输入docker ps时，我看到一条消息：Restarting (1) 11 minutes ago。我认为
浏览 4提问于2017-03-27得票数 9
回答已采纳
1回答
场景：
Stream create [StreamName] --definition " Kafka -zkconnect=10.10.10.1:2181 --topic=<topic name> | MyCompositeModule " --deploy
我们在分布式模式下运行这个流，而redis是传输总线。据我所知，kafka source通过streamname-kafka-offsets主题维护MyCompositeModule使用的消息的偏移量(这是一个接收器，因为它是通过'module compose‘过程创建的一个模块)。这是不可读的，如果
浏览 0提问于2015-12-18得票数 1
当我启动Kafka服务器时，Zookeper已经启动并在windows env.Now中运行。
.\bin\windows\kafka-server-start.bat .\config\server.properties
未显示任何内容
G:\Apache Kafka\kafka_2.11-0.11.0.0\kafka_2.11-0.11.0.0>.\bin\windows\kafka-serv
er-start.bat .\config\server.properties
G:\Apache Kafka\kafka_2.11-0.11.0.0\kafka_2.11-0.11.0.0
浏览 0提问于2017-08-06得票数 1
2回答
卡夫卡独立消费者
、、
我是第一次接触Kafka，正在尝试验证我的设计。下面是我所拥有的。 我有一个生产者发布到一个主题，并有一堆容器(我的web应用程序部署在那里)，其中一个消费者在每个容器上运行。这些消费者不在消费者组中，并且独立地消费消息。每个使用者都应该阅读该主题中的所有消息。例如，假设有3条关于主题的消息- m0，m1，m2，那么consumer1 to consumerN应该独立地读取m0，m1，m2。每个消费者在处理读取消息后都会立即提交offset。每当重新部署我的web应用程序时，所有使用者都会停机并重新启动。它们应该从上次提交的偏移量开始读取。 对于这个设置，正如我所说的，我所有的消费者都不在任
浏览 22提问于2019-03-19得票数 3
回答已采纳
如果在zookepeer的配置中，我将服务器设置为：
clientPort=2181
server.1=kafka-test-01:2888:3888
server.2=kafka-test-02:2888:3888
server.3=kafka-test-03:2888:3888
其中kafka-test-0X是一个实例，并且可以从任何实例进行域解析，我注意到Zookeeper的端口正在侦听127.0.1.1。事实上，由于无法通过3888进行通信，因此“先导端口”2888也不会侦听。
tcp6 0 0 127.0.1.1:3888 :::*
浏览 0提问于2020-05-21得票数 1
当kafka消费者实例重启时，其消费者group.id在每次重启时都会发生变化。有这样的行为是好的吗，还是会影响kafka的性能？
我们可以为“log.retention.ms”属性提供的最小值是什么。
浏览 9提问于2018-08-27得票数 0
我使用的是一个主题，一个分区，一个消费者，Kafka客户端版本是0.10。
我得到了两个不同的结果：
如果我先进行分区，然后生成消息并调用方法。KafkaConsumer可以成功地轮询未提交的消息。
但是，如果我首先生成消息，而没有提交它的偏移量，那么在几秒钟后暂停分区，以调用恢复方法。KafkaConsumer将不会接收未提交的消息。我使用kafka-consumer-groups.sh在Kafka服务器上检查了它，它显示了LOG-END-OFFSET minus CURRENT-OFFSET = LAG = 1。
我已经想了两天了，我反复做了很多次这样的测试，结果总是这样。
浏览 0提问于2018-11-12得票数 0
回答已采纳
我们使用的是spring，kafka(spring kafka 2.4.5，spring cloud stream 3.0.1)，openshift。我们有以下配置。多个代理/主题，每个主题有8个分区，复制因子为3，多个spring boot消费者。
作为弹性测试的一部分，当我们关闭其中一个代理时，我们会得到下面的异常，甚至在我们启动服务器之后，仍然会得到相同的错误。
org.apache.kafka.common.errors.NotLeaderForPartitionException: This server is not the leader for that topic-partit
浏览 287提问于2020-05-20得票数 1
1回答
我正在使用kafka节点模块向kafka发送消息。在集群环境中，我有一个主题，其中有3个分区，复制因子为3。
主题描述是-
Topic:clusterTopic PartitionCount:3 ReplicationFactor:3 Configs:min.insync.replicas=2,segment.bytes=1073741824
Topic: clusterTopic Partition: 0 Leader: 1 Replicas: 1,2,3 Isr: 1,2,3
Topic: cluste
浏览 3提问于2019-09-23得票数 1
我刚接触kafka和logstash。我希望使用logstash作为生产者，并希望使用SSL。
下面是我的logstash conf
input {
http {
port => 5044
codec => json
}
}
output {
kafka {
#bootstrap_servers => ["localhost:9093"]
bootstrap_servers => ["kafka broker's IP:9093"]
topic_id => "te
浏览 9提问于2020-06-26得票数 2
1回答
速率限制:由于Kafka能够以远高于MQ消耗的速率生成消息，我们是否可以使用一些配置设置@ kafka consumer来启用速率限制传输，以保护MQ的稳定性？
同样-只需一次语义-理解kafka只支持一次语义，这将阻止已经被消费者消费的消息的重新传输。有人可以指导我如何设置此配置吗？
我们在组织中使用的是confluent kafka企业版。
浏览 1提问于2020-10-07得票数 0
我在172.19.91.125中运行zookeeper和大多数kafka服务器。我把这台机器上的所有东西都运行得很好。
但是，当我尝试使用消费者访问172.19.101.61上的kafka服务时：
bin/kafka-console-consumer.sh --zookeeper 172.19.91.125:2181 --from-beginning --topic my-topic
我收到以下错误消息：
[2014-12-16 01:52:42,531] ERROR Producer connection to linux:9100 unsuccessful (kafka.producer.
浏览 0提问于2014-12-16得票数 2
我已经保留了auto.offset.reset = earliest 这里我的场景是不同的； 我有1个zookeeper和3个brokers.Now，我停止了消费者并开始发布消息(假设有4条消息，后来我停止了Kafka服务器(1个zookeeper和3个代理)，现在我组成了消费者。一段时间后，我建立了Kafka服务器(1个zookeeper和3个代理)。 由于我的4条消息没有与消费者一起处理，它现在应该处理，但这并没有发生。
浏览 25提问于2019-12-06得票数 0
我正在使用kafka控制台工具，并试图设置它，如果至少有一个kafka服务器实例仍然在线，仍然可以发送和接收消息，同时使用复制。
配置:动物园管理员(本地宿主:2181)，kafka1 (本地宿主:9092)，kafka2 (本地宿主:9093)，producer1 (本地宿主:9092)，producer2(本地宿主:9093)，consumer1 (localhost:9092)，consumer2 (localhost:9093)
主题: partitions=6，replication=2
如果kafka1服务器处于联机状态，则无论kafka2状态如何，发送和接收消息都是正常的。但是，
浏览 6提问于2017-02-28得票数 0
回答已采纳
我们运行的是3节点Kafka 0.10.0.1集群。我们有一个消费者应用程序，它有一个连接到多个主题的消费者组。我们在消费者日志中看到了奇怪的行为。有了这些行
Fetch offset 1109143 is out of range for partition email-4, resetting offset
Fetch offset 952168 is out of range for partition email-7, resetting offset
Fetch offset 945796 is out of range for partition email-5, reset
浏览 0提问于2018-06-17得票数 3
1回答
嗨，我正在致力于将kafka升级到.9，并将kafka消费者升级到与.9一起发布的新java消费者。在升级时，我正在使用现有的主题，步骤只是停止.8 kafka并启动指向相同log.dirs的.9 kafka，在消费者端，我使用相同的组名和主题名称，但新的消费者正在从主题中的起始位置再次消费消息。我已经在.8中提交了它们。我添加了auto.offset.reset =最早的。
是否知道为什么会发生这种情况，或者我们需要首先与.8消费者一起使用所有消息，并让延迟变为0。谢谢，桑尼
浏览 0提问于2016-05-05得票数 0
3回答
我有一个弹簧启动应用程序与单一的kafka消费者，以获取一些主题的消息。但有时在处理消息时会发生错误。
我希望像往常一样继续接收以下消息，同时能够不丢失该消息并接收它，例如，下次服务在修复后与使用者重启时。
有可能做到这一点吗？
我知道我需要禁用自动提交并手动提交成功的消息，但是，在这种情况下，如果我没有对这个异常情况抛出任何异常，并手动提交下一个成功的消息，那么我将丢失前一个失败的消息，对吧？
浏览 0提问于2020-08-30得票数 1
看起来，Kafka只允许消费者从leader代理(运行主分区的代理)中提取消息。followers只与leader同步，使用者不能从followers读取消息。
为什么不让消费者也从followers**?** 中提取消息？
我认为，如果使用者能够从followers读取消息，那么我们就可以对从所有代理服务器读取消息进行负载平衡，整个系统的性能应该会更好。比方说，leader太忙了，leader的带宽成了瓶颈。如果我们能够从followers中提取消息，那么性能应该会更好。
浏览 0提问于2016-11-25得票数 9
我试图用这样的测试数据填充kafka主题：
public static void main(String[] args){
Properties properties = new Properties();
properties.put("bootstrap.servers", "192.168.0.2:9092");
properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
浏览 0提问于2019-03-27得票数 0
1回答
我在找一个小小的指导。我有一个Access应用程序，它连接到服务器来运行SQL查询。在运行任何查询之前，我可以测试连接(ping)，但是当连接丢失时(这似乎发生得太频繁)，我无法优雅地处理。
我已经准备好了err_Handing，但是我似乎得到了一个嵌套的错误消息集合，包括：
3146 - ODBC Error
3151 - ODBC Connection Error
3704 - Object is Closed
2046 - Quit not available
>> Requires Ctl-Break, or Task Mgr to break loose...
我确实考虑
浏览 1提问于2016-10-18得票数 0
有没有可能使用Kafka源代码，在Beam中使用KafkaIO进行一次处理？有一个可以设置的名为AUTO_COMMIT的标志，但它似乎是在使用者处理数据之后直接提交给Kafka，而不是在管道完成处理消息之后。
浏览 2提问于2018-01-18得票数 3
回答已采纳
2回答
我试图理解卡夫卡的以下信息丢失行为。简单地说，当一个代理在消息处理之后早早就死了，那么所有其他代理都会死掉。如果先死亡的经纪人启动，那么在其他经纪人出现后，它就不会赶上其他经纪人。相反，所有其他代理都报告错误并重置其偏移量以匹配第一个代理。是否期望这种行为，以及确保零消息丢失的更改/设置是什么？
Kafka版本: 2.11-0.10.2.0
可复制步骤
启动了1个动物园管理员实例和3个kafka经纪人
创建一个主题，复制因子为3，分区为3
将kafka-控制台-使用者附加到主题中
使用Kafka-控制台-生产者来产生2条消息
杀害两名经纪人(1和2)
发了两条消息
浏览 0提问于2017-06-16得票数 1
我们使用spring kafka stream producer向kafka topic生成数据。当我们做弹性测试时，我们得到了下面的错误。 `2020-08-28 16:18:35.536警告，26 - ad | producer-3 o.a.k.clients.producer.internals.Sender : Producer clientId=producer-3在分区topic1-0上收到无效的元数据错误，由于分区org.apache.kafka.common.errors.NotLeaderForPartitionException:此服务器不是该topic- partiti
浏览 107提问于2020-08-29得票数 1
2回答
我有一个Java服务，可以读取Kafka的消息。服务很简单。我有个听众：
@KafkaListener(topics = "${kafka.topic.name}", groupId = "${kafka.topic.group}", containerFactory = "mesagueKafkaListenerContainerFactory")
然后，我就有了这样的感觉：
@Autowired
private KafkaProperty kafkaProperty;
private ConsumerFactory<Str
浏览 5提问于2020-02-04得票数 0
回答已采纳
我搞不懂为什么Flume数据在使用文件通道时是可恢复的，而在使用内存通道时是不可恢复的。
我尝试了Kafka接收器，当我在阅读时放下Flume时，Kafka正确地读取了通道中的数据(文件)。重新启动Flume后，管道将继续以可靠的方式传送数据。那么为什么内存通道是不可恢复的呢？在这种情况下，我需要恢复通道中的数据？，特别是当Flume开始从保存的偏移量读取文件时。
浏览 0提问于2017-08-03得票数 0
我有一个五个节点的kafka集群(合流5.5社区版)，在不同的aws实例上有3个动物园管理员节点。在进行故障转移测试时，注意到kafka集群运行良好，即使所有动物园管理员节点都关闭了。我能够生产，消费，也创造了新的消费者。
为什么卡夫卡集群如果不能连接到任何动物园管理员节点，它就不会停止?如果我们在生产中不知道这样一个失败的场景，并且kafka集群继续运行而没有动物园管理员连接，那么它会有什么可能的问题？，我们如何处理这种情况?
浏览 0提问于2021-02-17得票数 0
回答已采纳
1回答
我使用spring框架和kafka，它有3个代理集群。我发现消费者没有消费一些消息(假设所有发送的消息之间有0.01%)，所以在生产者代码中，我记录了api返回的消息偏移量： ListenableFuture<SendResult<String, Message>> future = messageTemplate.sendDefault(id, message);
SendResult<String, Message> sendResult = future.get();
String offset = sendResult.getRecordMetad
浏览 15提问于2019-05-16得票数 2
回答已采纳
2回答
将同步副本配置为Acks=all和min.insync.replicas = N，
想了解如何为未处理的生产者记录的消息/记录配置重试
当Kafka无法处理ISR时，在线记录在处理过程中是N-1，最小配置的ISR是N个副本。
浏览 8提问于2021-01-17得票数 0
让我们假设我们有一个运行在apache we服务器上的PHP api。我想问一下，如何以结构化格式将日志从PHP放到kafka中(让我们说avro或protobuf，因为它将传输特定的业务合同，而不仅仅是纯apache日志)。Php不在我的领域或专业范围内，所以它可能很愚蠢，但解决方案可能只是从php本身向kafka发送消息。但考虑到它是在网络服务器下运行的，它很可能必须为每次调用创建到kafka的连接。另一种可能的解决方案是将日志记录到文件中，然后使用filebeat之类的东西将日志放到kafka中。但我不确定如果我们使用结构化格式，如我已经提到的那样，它将如何工作。我怀疑filebeat
浏览 16提问于2021-10-22得票数 0
有一个应用程序(不是我的)从Kafka读取消息，对它们进行一些处理，并将记录存储在数据库中。我用Java编写了一个程序，它以给定的速率将消息写入队列。现在，它通过在测试运行结束时查询数据库来进行简单的性能测量，以确保记录in = records out。但是，我想将其扩展为定期检查队列，以查看应用程序尚未处理的挂起消息的数量，以及是否正在备份它。
我想我可以在Zookeeper中检查应用程序组ID的偏移量。我查看了，但它只给出了基本的消费者示例，并且API文档充其量也是稀疏的，所以我不确定如何查找这些信息。
我需要调用哪些API才能找出应用程序当前在队列中的位置，以及该位置后面的队列中有多少消
浏览 2提问于2015-07-23得票数 3
有时，当我试图运行Kafka broker时，我会执行以下错误：
ERROR Fatal error during KafkaServer startup. Prepare to shutdown (kafka.server.KafkaServer)
kafka.common.InconsistentClusterIdException: The Cluster ID m1Ze6AjGRwqarkcxJscgyQ doesn't match stored clusterId Some(1TGYcbFuRXa4Lqojs4B9Hw) in meta.properties. The bro
浏览 3提问于2020-11-24得票数 2
1回答
突然有几个主题在日志中出现了重复的错误，我该怎么办？
May 04 14:17:23 2kafka01.rtty.in kafka-server-start.sh[149443]: kafka.common.NotAssignedReplicaException: Leader 1 failed to record follower 2's position -1 since the replica is not recognized to be one of the assigned replicas 1,3,4 for partition [impression,3].
May 0
浏览 1提问于2017-05-04得票数 1
1回答
我是卡夫卡的新手，还在学习同样的基础知识。我想要创建一个持久的主题，即使在动物园管理员和/或kafka服务器关机之后也要保留。我注意到了这一点-我有一个动物园管理员和卡夫卡服务器运行在我的本地macbook。当我关闭动物园管理员服务器并再次快速打开它时，我可以看到先前创建的主题。但是如果我重新启动系统，然后重新启动动物园管理员服务器--我看不到我之前创建的主题。我在本地系统上运行kafka_2.9.2-0.8.1.1。
浏览 2提问于2015-09-12得票数 1
回答已采纳
重启属于集群的所有kafka服务器的建议方法是什么？
这个问题的本质是因为我们通过ansible管理它，并且我们有一个关于例如server.properties的更改的notify，所以重启将并发发生。
通常，当一个代理消失时，会重新平衡数据，所以我想知道如果所有代理同时重新启动会发生什么。
浏览 1提问于2020-01-09得票数 0
我们正在使用Kafka流媒体库来构建关于Kafka主题的传入消息的实时通知类系统，因此当流媒体应用程序运行时，它会实时处理主题中的所有传入消息，并在遇到某种预定义的传入消息时发送通知。
如果流应用关闭并重新启动，我们只需要处理流应用初始化后最近到达的消息。这是为了避免处理在流媒体应用程序未运行或停机时未处理的旧记录。默认情况下，流式应用程序开始处理自上次提交偏移量以来的旧消息。Kafka Streaming App有没有什么设置，可以只处理最近的消息？
浏览 2提问于2018-05-10得票数 2
2回答
在使用Kafka的星火流中，我面临数据丢失的问题，我的用例如下：
火花流(DirectStream)应用程序从Kafka主题读取消息并进行处理。
在处理消息的基础上，应用程序将处理后的消息写入不同的Kafka主题，例如，如果消息是协调的，那么将消息写到协调的主题或不协调的主题上。
现在，问题是，在流过程中，不知何故，我丢失了一些消息，即所有传入的消息都没有写入到协调或不协调的主题中。例如，如果应用程序在一批中接收了30条消息，那么有时它会将所有消息写入输出主题(这是预期的行为)，但有时它只写了27条消息(3条消息丢失了，这个数字可能会改变)。
以下是我使用的版本：
浏览 4提问于2017-05-30得票数 8
我对Kafka很陌生，我有一个使用java微服务的kafka流，它消耗了由生产者和过程产生的kafka主题的信息。使用auto.commit.interval.ms设置了kafka提交间隔。我的问题是，在提交之前，如果微服务崩溃，那么处理但未提交的消息会发生什么情况？会有复制的记录吗？如果发生这种重复，该如何解决？
浏览 6提问于2022-02-25得票数 1
1回答
我有一个与Apache集成的简单java代码，它还使用camel-kafka组件来记录kafka主题中的消息。我已经创建了一个处理单个请求的类。使用线程，我可以创建各种线程来调用上述类方法来记录消息。
目前，我需要使用工具来加载这个JAR。我想知道一个学习曲线很低的工具。
负载测试：
增加用户，允许并发记录多条消息
更改消息以增加/减小消息大小
特定用户记录特定消息所需的时间。
我经历过
JMeter (学习曲线很大)
JProfiler (它不加载测试，但如果我没有错，它会监视应用程序)
Netbeans负载生成器(再次使用JMeter)
浏览 0提问于2015-09-15得票数 0
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
实时音视频活动推荐
腾讯云开发者
