未删除过期的Apache Kafka消息
Apache Kafka是一个分布式流处理平台，最初由LinkedIn开发并开源。它被设计用于高吞吐量、低延迟的数据传输，以及实时数据流的处理。Kafka基于发布-订阅模式，通过将消息分区存储在多个服务器上，实现了高可用性和容错性。
未删除过期的Apache Kafka消息是指在Kafka中，消息被写入主题（topic）后，并不会立即被删除。相反，Kafka会根据配置的保留策略来决定消息的保留时间。如果消息超过了保留时间，它将被自动删除。
保留策略可以根据时间或者大小来配置。时间保留策略指定了消息在主题中保留的时间段，而大小保留策略则指定了主题中消息的总大小。一旦消息超过了保留时间或者主题的总大小达到了限制，Kafka将开始删除最旧的消息。
未删除过期的消息在某些场景下可能会导致存储空间的浪费，因此在配置Kafka时需要根据实际需求来设置保留策略。例如，在需要长时间保留消息的场景下，可以设置较长的保留时间或者较大的主题大小限制。而对于实时数据处理的场景，可以设置较短的保留时间或者较小的主题大小限制，以避免存储空间的浪费。
腾讯云提供了一系列与Kafka相关的产品和服务，包括TDMQ（Tencent Distributed Message Queue）和CKafka（Cloud Kafka）。TDMQ是腾讯云自研的分布式消息队列产品，提供高可用、高性能的消息传输和处理能力。CKafka是腾讯云基于开源Kafka打造的托管式Kafka服务，提供了简单易用的管理界面和弹性扩展能力。
关于TDMQ的详细介绍和产品链接地址：TDMQ产品介绍
关于CKafka的详细介绍和产品链接地址：CKafka产品介绍
相关·内容
我正在使用Logstash输入插件来读取来自某个主题的消息。更早的时候，我能够启动新的消费者--属于新的消费者群体，通过设置auto_offset_reset=earliest可以从主题一开始就使用消息。
插件配置：
input {
kafka {
bootstrap_servers => "localhost:9092"
topics => ["test_topic"]
group_id => "new_consumer"
client
浏览 6提问于2016-12-23得票数 0
回答已采纳
我有一个有3个节点的融合平台Kafka集群，一切似乎工作正常，但当产品与控制台的一些消息，得到错误！
命令：
/opt/kafka/confluent-4.0.0/bin/kafka-console-producer --broker-list 109.169.xxx.xxx:9092 --topic sh
错误：
ERROR Error when sending message to topic sh with key: null, value: 1 bytes with error: (org.apache.kafka.clients.producer.internals.ErrorLo
浏览 4提问于2018-04-17得票数 3
使用从只有少数消息的主题中读取消息。卡夫卡一遍又一遍地重复队列中的信息。
它接收来自我的主题的一条消息(返回到每个消息内容)，然后抛出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回答
错误
Expiring 2 record(s) for catering-0:120012 ms has passed since batch creation
首先我使用了配置
zookeeper:
image: wurstmeister/zookeeper
container_name: zookeeper
ports:
- "2181:2181"
restart: always
kafka:
image: wurstmeister/kafka
container_name: kafka
浏览 9提问于2022-07-11得票数 0
我有一个流应用程序与下面的驱动程序代码的实时消息转换.
String topicName = ...
KStreamBuilder builder = new KStreamBuilder();
KStream<String, String> source = builder.stream(topicName);
source.transform(() -> new MyTransformer()).to(...);
KafkaStreams streams = new KafkaStreams(builder, appConfig);
streams.setUncaug
浏览 17提问于2017-02-22得票数 7
回答已采纳
apache kafka的新版本在内部存储消费者组数据，而不是使用zookeeper
这意味着命令行实用程序kafka-consumer-groups没有用处，因为相关信息现在没有存储在zookeeper中
你能告诉我：
如何在使用新的消费者API时删除属于主题T的消费者组C
浏览 6提问于2017-11-10得票数 0
1回答
我们正在开发并尝试删除某个主题的模式，因为该更改与旧模式不兼容。
我们删除了模式/主题，并尝试以相同的主题名称创建新的模式，并成功创建了模式。
但是，当我们运行应用程序时，它仍然指向相同的模式ID。
较旧的架构ID (用于主题‘topic1’)：51
新架构ID (用于主题'topic1')：52
应用程序失败，并在反序列化消息时出错
at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:736)
Caused by: org.apache.kafka.com
浏览 0提问于2019-09-25得票数 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
2回答
描述
I有一个卡夫卡连接分布式模式的pubSubSource连接器，它只是简单地从PubSub订阅中读取并写入一个Kafka主题。问题是，即使我要向GCP PubSub发布一条消息，我也要在我的Kafka主题中写两次这个消息。。
如何繁殖
部署Kafka和Kafka连接
创建具有以下pubSubSource配置的连接器：
内容类型: application/json“-d”{“名称”："pubSubSource"，"config"：{ "org.apache.kafka.connect.json.JsonConverter"，"key.c
浏览 1提问于2021-02-24得票数 1
回答已采纳
更新
为了提供一些上下文，使用LZ4压缩启用了Kafka主题。
当识别压缩类型时，尝试将LZ4与运行时由于某种原因可以找到的KafkaLZ4BlockInputStream相关联。
@Override
public InputStream wrapForInput(ByteBuffer inputBuffer, byte messageVersion, BufferSupplier decompressionBufferSupplier) {
try {
return new KafkaLZ4BlockInputStream(inputBuffer, decompres
浏览 18提问于2021-12-09得票数 1
1回答
我是Apache的新手，我正在尝试使用来自某个主题的消息。我从控制台运行了以下命令来查看消息：
bin\windows\kafka-console-consumer.bat --zookeeper localhost:2181 --topic test --from-beginning
但在那之后，我可以看到一个名为“console - consumer 4158”的消费者被创建，我意外地关闭了控制台。
如何终止正在运行的使用者？
浏览 4提问于2017-02-28得票数 0
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
回答已采纳
1回答
我在centos 7(汇合)上安装了Apache，我试图在分布式模式下运行filestream连接，但是我得到了以下错误：
[2017-08-10 05:26:27,355] INFO Added alias 'ValueToKey' to plugin 'org.apache.kafka.connect.transforms.ValueToKey' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader:290)
Exception in thread "main" o
浏览 6提问于2017-08-10得票数 1
回答已采纳
我是Kafka的初学者，最近开始在我的工作项目中使用。我想知道的一件重要的事情是，当消息在kafka中过期时，是否有可能捕获事件。这样做的目的是捕获这些过期邮件，并将它们备份到备份存储中。
浏览 1提问于2021-03-17得票数 0
我正在研究一个用例，在这个用例中，我必须将外部Kafka主题元数据导入apache地图。以下是我想澄清的几个问题：
是否可以从不用于地图集通知目的的外部Kafka导入主题元数据？如果可能的话，怎么做？
如何使Kafka元数据更新与Hive或Hbase类似，而不是每次手动运行导入脚本？
没有导入主题的沿袭数据。在什么情况下，会为主题捕获谱系数据？
既然只有一个卡夫卡相关的实体"kafka_topic"，难道根本就没有关系数据吗？
在什么情况下，审计被捕获为主题？
浏览 1提问于2018-12-11得票数 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回答
如何处理卡夫卡中的大消息，比如超过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
回答已采纳
我有一个卡夫卡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
我想要删除我的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
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
回答已采纳
使用以下配置创建接收器连接器时
connector.class=io.confluent.connect.s3.S3SinkConnector
s3.region=us-west-2
topics.dir=topics
flush.size=3
schema.compatibility=NONE
topics=my_topic
tasks.max=1
s3.part.size=5242880
format.class=io.confluent.connect.s3.format.avro.AvroFormat
# added after comment
key.converter=io.co
浏览 5提问于2018-07-20得票数 0
回答已采纳
我已经安装了Kafka并做了一些基本的测试。我可以使用Kafka-broker/bin文件夹下提供的脚本创建主题。
但是，当我每次尝试生成低于WARNing的消息时，我都会运行此命令。并且没有生成任何消息。敬请指教。
[root@node2 bin]# ./kafka-console-producer.sh --broker-list localhost:9092 --topic test_master
>testmsg1
[2019-05-15 06:25:19,092] WARN [Producer clientId=console-producer] Connection
浏览 19提问于2019-05-16得票数 3
2回答
我是卡夫卡的新手。我们正在尝试将数据从csv文件导入到Kafka。我们每天都需要导入数据，平均来说，前一天的数据是被降级的。在python中，如何删除Kafka主题下的所有消息？或者我如何删除python中的Kafka主题？或者我看到有人建议等待数据过期，如果有可能，我如何设置数据过期时间？如有任何建议，将不胜感激！
谢谢
浏览 5提问于2015-09-21得票数 4
回答已采纳
每当我运行任何kafka CLI工具时，我都会收到以下警告消息 kafka-topics.sh --bootstrap-server kafka-dev:9092 --command-config client.properties --list
[2021-07-14 07:02:31,595] WARN The configuration 'ssl.truststore.location' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig
浏览 104提问于2021-07-14得票数 0
回答已采纳
可能重复的，但没有接受的答案，也没有一个明确的解决方案。
我有一个简单的java程序来生成给Kafka的消息：
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("acks", "all");
props.put("retries", 1);
props.put("batch.size", 16384);
props.put("linger.ms&#
浏览 0提问于2018-10-25得票数 4
假设我们正在KStream和KTable之间进行内部连接，如下所示：
StreamsBuilder sb = new StreamsBuilder();
JsonSerde<SensorMetaData> sensorMetaDataJsonSerde = new JsonSerde<>(SensorMetaData.class);
KTable<String, String> kTable = sb.stream("sensorMetadata",
Consum
浏览 2提问于2020-06-06得票数 0
回答已采纳
我正在为Apache使用Spring，并创建了一个服务，该服务通过Spring的org.apache.kafka.clients.producer(通过Spring的KafkaTemplate )使用卡夫卡生产者( Kafka Producer，KafkaTemplate)向主题发送消息。在目标Kafka集群上，我禁用了自动主题创建。使用这里列出的生产者配置的组合( )，我成功地控制了重试请求的次数、重试之间的时间等等。
如果我提供了一个不存在的主题，那么当我期望它达到max.block.ms).值时，请求就会超时。但是，超时之后，我将继续在retry.backoff.ms设置的时间间隔内获
浏览 4提问于2020-02-17得票数 2
回答已采纳
我有一个Spring Kafka Consumer应用程序，它位于K8中。有时应用程序会被回收/重新启动。当消费者重新启动时，我希望它使用回收过程中产生的所有消息。我用auto.offset.rest=earliest进行了实验，结果和预期的一样，但是我注意到kafka的默认值是最新的。 如果我使用最早的，会有什么风险？在什么情况下我最早使用最新的v.s？我试图在这里找到一篇文章，通过一个场景来解释它，但大多数都是从一些文档中复制粘贴过来的，而不是现实生活中的例子。
浏览 45提问于2021-04-28得票数 2
火花流中有spark.streaming.kafka.maxRatePerPartition属性，它限制每秒从Apache读取消息的数量。斯托姆也有类似的财产吗？
浏览 2提问于2016-06-10得票数 0
回答已采纳
1回答
我已经成功地获得了一个用于消费、转换和生成数据的流应用程序，但是我注意到，定期地，流处理器将转换到ERROR状态，并且进程将处于不退出状态。
给我看原木如下：
All stream threads have died. The instance will be in error state and should be closed.
有什么方法可以告诉流应用程序一旦到达ERROR状态就退出吗？也许是某种监控线程？
我在Kafka代码的注释中看到了，当应用程序达到这种状态时，用户需要关闭它，但是，我在文档中还没有找到对这个任务的提及。
有简单的方法来完成这个关机步骤吗？
可能不正确的方式可能关
浏览 0提问于2018-03-06得票数 10
回答已采纳
我间歇性地(一个月一到两次)看到错误org.apache.kafka.common.errors.TimeoutException: Expiring 1 record(s) for cart-topic-0: 5109 ms has passed since batch creation plus linger time in my logs due to which the corresponding message was not processed by Kafka Producer。
尽管所有代理都已启动并可用，但我不确定为什么会观察到此错误。在此期间，即使是负载也不是很多。
我已
浏览 0提问于2019-07-17得票数 1
用例
我为卡夫卡消费集团补偿()写了一个prometheus出口商，它通过使用内部__consumer_offsets主题来工作。本主题中所有消息的键和值都是二进制的，我能够对它们进行解码。当任何组偏移到期时，Kafka应该为该键创建一个墓碑(由组id、主题名称和分区id组成)。
问题
到目前为止，这种方法非常有效，但是Kafka显然没有为一些过期的偏移/组提供墓碑。我使用以下命令讨论了偏移主题：
kafka-console-consumer.sh --from-beginning --topic __consumer_offsets zookeeper.connect=${ZOO_HOST}
浏览 3提问于2019-12-27得票数 0
Flink-kafka-consumer有两种类型的消费者，例如： 这两个消费者层次结构都扩展了相同的FlinkKafkaConsumerBase类。我想知道维护编号类背后的设计决策是什么？我们应该在什么时候使用其中一个呢？ 我注意到带有编号后缀的类有一些更多的功能(例如，分级限制)。 https://github.com/apache/flink/blob/master/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaCons
浏览 6提问于2020-09-14得票数 0
我正在使用具有以下配置的KafkaConnect - MongoSource： curl -X PUT http://localhost:8083/connectors/mongo-source2/config -H "Content-Type: application/json" -d '{
"name":"mongo-source2",
"tasks.max":1,
"connector.class":"com.mongodb.kafka.connect.MongoSourceCo
浏览 13提问于2019-12-21得票数 2
我正在尝试使用Kafka控制台生成器向运行在同一台机器上的Kafka代理发送消息。当我跑的时候
echo "Hello world" | ./kafka-console-producer.sh --broker-list localhost:9092 --topic test
我收到以下错误消息：
[2016-06-29 15:00:44,069] ERROR Error when sending message to topic test with key: null, value: 11 bytes with error: (org.apache.kafka.client
浏览 2提问于2016-06-29得票数 6
我们已经实现了一个应用程序来生成消息和标题，并将其发送到kafka主题。消费者端应用程序正在聆听卡夫卡的主题。消费者端应用程序能够接收有效载荷，但是头不是received.Earlier，我们的头是在有效载荷属性中接收的。为了禁用将报头(包括contentType)嵌入到有效负载中，我们使用了
spring.cloud.stream.bindings.output.producer.headerMode=raw
spring.cloud.stream.bindings.input.consumer.headerMode=raw
但是，我们无法在使用者端应用程序中接收报头。我们如何接收这些标头？
浏览 2提问于2019-05-08得票数 0
在执行kstreams以获取聚合计数时，我将得到此错误。
Exception in thread "KStreamThread-1" org.apache.kafka.streams.errors.ProcessorStateException: task [1_2] Failed to flush state store KSTREAM-AGGREGATE-STATE-STORE-0000000002
at org.apache.kafka.streams.processor.internals.ProcessorStateManager.flush(ProcessorSta
浏览 0提问于2019-01-29得票数 3
我的生产者用键0,1,2,3,4,5,6,7,8,9写了10条消息
我的消费者(自动提交= false)在第一次循环期间读取消息，但当读取消息"2“时，我强制执行异常，因此不提交。(异常仅在第一次循环时抛出-有标志forceError)
在第二个循环中，消费者应该读取所有消息0,1,2,3,4,5,6,7,8,9并提交，但事实并非如此，为什么？
日志是这样的：
write message with key: 0
write message with key: 1
write message with key: 2
write message with key: 3
write mess
浏览 9提问于2018-02-15得票数 4
我正在尝试使用kafka- avro -console-consumer以log4j格式打印关于kafka主题的avro消息。
为此，我使用以下kafka-avro-console-consumer命令：
bin/kafka-avro-console-consumer --bootstrap-server localhost:9092 --topic avro-test -property print.key=true --formatter kafka.tools.LoggingMessageFormatter
我已经通过以下命令导出了KAFKA_OPTS：
export $KAFKA
浏览 0提问于2018-07-18得票数 1
我刚刚安装了Cloudera QuickStart VM，并在其中添加了Kafka服务。在添加了Kafka服务之后，我可以轻松地创建一个生产者/消费者，一切都如预期的那样工作。几天后，虚拟机重新启动，我得到了以下异常：
线程“主”org.apache.kafka.common.errors.TimeoutException:中的异常: 60000 ms后，java.util.concurrent.ExecutionException未能更新元数据。在org.apache.kafka.clients.producer.KafkaProducer$FutureFailure.(KafkaProd
浏览 0提问于2018-04-06得票数 0
我的用例：使用Postman，我调用了Spring端点。端点创建一个KafkaProducer并向特定主题发送消息。我还有一个TaskScheduler来使用这个主题。
问题：调用soap将消息推送到主题时，我得到以下错误：
2017年-11-14 21:29:31.463错误6389 -广告
下面是我用来推介主题的方法：
public DomainEntity push(DomainEntity pDomainEntity) throws Exception {
logger.log(Level.INFO, "streaming...");
wKafk
浏览 5提问于2017-11-15得票数 6
我刚接触Kafka，似乎对Apache Kafka的“快速入门”指南有几个问题，可以在这里找到：
最终，我正在尝试学习如何加载带有许多kafka消息的kafka队列，因此本快速入门指南的Step 7部分似乎是相关的。
我在这里安装了二进制下载包(scala2.11- kafka_ 2.11 -1.1.0.tgz )：
我最初试图直接跳到第7步，但在找到这个问题()后，我意识到我必须执行之前的几个步骤
因此，我成功地完成了第一步：
tar -xzf kafka_2.11-1.1.0.tgz
cd kafka_2.11-1.1.0
然后，我执行了第二步：
bin/zookeeper-server-
浏览 7提问于2018-04-09得票数 0
我正在尝试构建一个pub/sub应用程序，我正在探索最好的工具。我现在看到的卡夫卡，并有一个小的演示应用程序已经运行。然而，我遇到了一个概念性的问题。
我有一个生产者(Java代码)：
String topicName = "MyTopic;
String key = "MyKey";
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092,localhost:9093");
props
浏览 3提问于2017-03-10得票数 3
回答已采纳
我需要删除Apache Kafka 0.8.1.1中的主题test。
正如在中所表达的，我已经执行了：
bin/kafka-topics.sh --zookeeper localhost:2181 --delete --topic test
但是，这会导致以下消息：
Command must include exactly one action: --list, --describe, --create or --alter
如何删除此主题？
浏览 98提问于2014-06-18得票数 108
回答已采纳
我正在尝试创建一个spring boot应用程序(Java)，它必须能够通过给出偏移量和分区来删除kafka主题中的消息。我一直在研究可以做到这一点的java或spring boot包类，但我只发现了这样的东西：Delete Messages from a Topic in Apache Kafka有一个java kafka客户端，它有一个方法可以删除偏移前的所有消息，但我只有一个方法可以删除一个。这有可能吗？ 提前感谢
浏览 36提问于2020-09-08得票数 0
回答已采纳
Context：我试图在使用ESP8266设备和SoftAP创建的本地WiFi网络上连接卡夫卡。
我正在努力实现的：我希望设备(与ESP8266 SoftAP连接的笔记本电脑)使用卡夫卡在没有连接到互联网的本地WiFi网络上相互发送和接收消息。
Problem：网络中的设备之一是Raspberry Pi，我正在运行Kafka。我所犯的错误是
[2022-05-09 14:54:43,622] WARN [Controller id=0, targetBrokerId=0] Error connecting to node :192.168.4.2:9092 (id: 0 rack: null)
浏览 4提问于2022-05-09得票数 0
我正在尝试向JdbcSink推送一条消息，消息如下 {
"schema": {
"type": "struct",
"fields": [{
"field": "ID",
"type": {
"type": "bytes",
"scale": 0,
"precision": 64,
&
浏览 27提问于2019-12-31得票数 1
回答已采纳
2回答
我用卡夫卡制片发送价格到主题。当我发送第一条消息时，它会打印出厂配置，然后为此发送消息，这需要更多的时间来发送第一条消息。
在第一条消息之后，发送一条消息几乎不需要1/2毫秒。
我的问题是，我们能不能做一些事情，这样配置部分就可以跳过，或者我们可以在发送第一条消息之前启动？
我正在使用春季卡夫卡在我的项目。我也读过其他的问题，但没有多大帮助。
Application.yml
server:
port: 8081
spring:
kafka:
bootstrap-servers: ***.***.*.***:9092
producer:
浏览 2提问于2020-10-06得票数 1
我完全意识到卡夫卡是一根木头，就像哑巴。但是在我们的应用程序中，我们有一些“遵从性主题”，我们需要确保读取其中的所有消息。
由于Kafka内部处理所有消费者的抵消(因此可以监视消费者延迟)，我想知道是否可以设置log.retention=25GB，但不要覆盖没有被任何人读取的消息，而是抛出一个生产者异常(就好像我们设置了log.retention=1week和HDD已经满了一样)。
如果无法在Kafka级别上配置这种逻辑，那么在kafka-生产者或kafka流上如何实现这一逻辑，从而在“完全未读日志”情况下失败吗？
浏览 0提问于2018-04-04得票数 1
回答已采纳
我在远程服务器上使用Kafka Debezium MySQL连接器。我一直收到两条错误消息，它们随机出现(我没有编辑任何内容)。
第一个是
[2021-02-26 14:08:00,523] INFO Kafka commitId: 9c1fbb3db1e0d69d (org.apache.kafka.common.utils.AppInfoParser:118)
[2021-02-26 14:08:00,523] INFO Kafka startTimeMs: 1614370080522 (org.apache.kafka.common.utils.AppInfoParser:119)
[2
浏览 6提问于2021-02-27得票数 1
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
实时音视频
即时通信 IM活动推荐
腾讯云开发者
