Logstash持久化队列配置无效：Kafka数据未存盘致丢失求助

一、先确认持久化队列基础配置是否生效 #

• 必须重启Logstash：logstash.yml的配置修改后，只有重启Logstash进程才会加载新配置，这是新手最容易忽略的点。
• 检查队列目录权限：
  执行命令ls -ld /var/lib/logstash/queue，确保Logstash运行用户（通常是logstash）对该目录有读写权限。如果权限不足，执行chown -R logstash:logstash /var/lib/logstash/queue修正。
• 验证队列是否启动：
  查看Logstash日志（默认路径/var/log/logstash/logstash-plain.log），搜索Persistent queue is enabled关键字确认队列启用；同时检查/var/lib/logstash/queue目录下是否生成.checkpoint、.page等文件，没有则说明配置未生效（可能是YAML语法错误，比如缩进不一致，需确保每个配置参数前的空格数统一）。

二、配置PostgreSQL输出的重试策略 #

仅靠持久化队列还不够，必须让输出插件在数据库故障时持续重试，触发队列缓存逻辑。在你的JDBC输出配置中添加以下参数：

output {
  jdbc {
    connection_string => "jdbc:postgresql://你的数据库地址:5432/库名"
    username => "用户名"
    password => "密码"
    statement => "INSERT INTO 表名 (字段1, 字段2) VALUES (?, ?)"
    parameters => { "字段1" => "%{字段1}" , "字段2" => "%{字段2}" }
    # 核心重试配置
    retry_on_failure => true       # 开启失败重试
    retry_max_interval => 300      # 重试间隔上限（秒）
    retry_attempts => -1           # 无限重试（直到数据库恢复）
  }
}

三、调整Kafka输入的Offset提交策略 #

如果输入插件自动提交Kafka Offset，即使输出失败，Offset也会被提交，导致数据丢失。修改Kafka输入配置：

input {
  kafka {
    bootstrap_servers => "Kafka地址:9092"
    topics => ["你的主题"]
    group_id => "logstash消费组ID"
    auto_commit_enable => false    # 关闭自动提交，由Logstash在输出成功后提交
    consumer_threads => 1
  }
}

四、验证故障场景 #

1. 手动停止PostgreSQL服务，向Kafka发送测试数据；
2. 观察Logstash日志是否出现重试提示，同时查看/var/lib/logstash/queue目录下的文件大小是否增长；
3. 恢复PostgreSQL服务，确认数据被写入数据库，且队列目录文件逐步减少。

内容的提问来源于stack exchange，提问作者P N Jain