DLQ throw NPE when the pipeline shutdown

[kaisecheng]

Pipeline running dlq-input throws NullPointerException when received SIGTERM with the following conditions

• dead letter queue directory has never had a available segment
• commit_offsets => true
• start_timestamp is not specified
• sincedb_path => "/path/to/sincedb"

When DeadLetterQueueReader finds no available segment in the path

• currentReader (segment) is unset (null) here
• readerHasState is set to true.

Logstash received SIGTERM, close() try to update SinceDB before quit by getting the path of segment

Steps

# dlq.conf
input {
    dead_letter_queue {
        path => "/tmp/logstash/data/dead_letter_queue"  # /tmp/logstash/data/dead_letter_queue/pipe1 should be empty folder
        sincedb_path => "/path/to/sincedb" # a valid non empty sincedb
        commit_offsets => true
        pipeline_id => "pipe1"
    }
}
output {
    stdout {  }
}

• Run bin/logstash -f dlq.conf
• After pipeline started, shutdown logstash
• You can see the error log

Logs

[2024-08-21T08:11:24,812][ERROR][logstash.agent           ] Failed to execute action {:action=>LogStash::PipelineAction::StopAndDelete/pipeline_id:deadletterqueue-standalone, :exception=>"Java::JavaLang::NullPointerException", :message=>"Cannot invoke \"org.logstash.common.io.RecordIOReader.getPath()\" because \"this.currentReader\" is null", :backtrace=>["org.logstash.common.io.DeadLetterQueueReader.getCurrentSegment(DeadLetterQueueReader.java:428)", "org.logstash.input.SinceDB.getUpdated(SinceDB.java:76)", "org.logstash.input.DeadLetterQueueInputPlugin.close(DeadLetterQueueInputPlugin.java:159)", "usr.share.logstash.vendor.bundle.jruby.$3_dot_1_dot_0.gems.logstash_minus_input_minus_dead_letter_queue_minus_2_dot_0_dot_0.lib.logstash.inputs.dead_letter_queue.RUBY$method$stop$0(/usr/share/logstash/vendor/bundle/jruby/3.1.0/gems/logstash-input-dead_letter_queue-2.0.0/lib/logstash/inputs/dead_letter_queue.rb:107)", "org.jruby.internal.runtime.methods.CompiledIRMethod.call(CompiledIRMethod.java:152)", "org.jruby.internal.runtime.methods.MixedModeIRMethod.call(MixedModeIRMethod.java:148)", "org.jruby.internal.runtime.methods.DynamicMethod.call(DynamicMethod.java:212)", "org.jruby.runtime.callsite.CachingCallSite.cacheAndCall(CachingCallSite.java:456)", "org.jruby.runtime.callsite.CachingCallSite.call(CachingCallSite.java:195)", "org.jruby.ir.interpreter.InterpreterEngine.processCall(InterpreterEngine.java:350)", "org.jruby.ir.interpreter.StartupInterpreterEngine.interpret(StartupInterpreterEngine.java:66)", "org.jruby.internal.runtime.methods.MixedModeIRMethod.INTERPRET_METHOD(MixedModeIRMethod.java:128)", "org.jruby.internal.runtime.methods.MixedModeIRMethod.call(MixedModeIRMethod.java:115)", "org.jruby.runtime.callsite.CachingCallSite.cacheAndCall(CachingCallSite.java:446)", "org.jruby.runtime.callsite.CachingCallSite.call(CachingCallSite.java:92)", "org.jruby.RubySymbol$SymbolProcBody.yieldInner(RubySymbol.java:1508)", "org.jruby.RubySymbol$SymbolProcBody.doYield(RubySymbol.java:1523)", "org.jruby.runtime.BlockBody.yield(BlockBody.java:108)", "org.jruby.runtime.Block.yield(Block.java:189)", "org.jruby.RubyArray.each(RubyArray.java:1989)", "org.jruby.RubyArray$INVOKER$i$0$0$each.call(RubyArray$INVOKER$i$0$0$each.gen)", "org.jruby.internal.runtime.methods.JavaMethod$JavaMethodZeroBlock.call(JavaMethod.java:561)", "org.jruby.runtime.callsite.CachingCallSite.call(CachingCallSite.java:90)", "org.jruby.ir.instructions.CallBase.interpret(CallBase.java:548)", "org.jruby.ir.interpreter.InterpreterEngine.processCall(InterpreterEngine.java:367)", "org.jruby.ir.interpreter.StartupInterpreterEngine.interpret(StartupInterpreterEngine.java:66)", "org.jruby.ir.interpreter.InterpreterEngine.interpret(InterpreterEngine.java:76)", "org.jruby.internal.runtime.methods.MixedModeIRMethod.INTERPRET_METHOD(MixedModeIRMethod.java:164)", "org.jruby.internal.runtime.methods.MixedModeIRMethod.call(MixedModeIRMethod.java:151)", "org.jruby.internal.runtime.methods.DynamicMethod.call(DynamicMethod.java:212)", "org.jruby.runtime.callsite.CachingCallSite.cacheAndCall(CachingCallSite.java:456)", "org.jruby.runtime.callsite.CachingCallSite.call(CachingCallSite.java:195)", "org.jruby.ir.interpreter.InterpreterEngine.processCall(InterpreterEngine.java:350)", "org.jruby.ir.interpreter.StartupInterpreterEngine.interpret(StartupInterpreterEngine.java:66)", "org.jruby.internal.runtime.methods.MixedModeIRMethod.INTERPRET_METHOD(MixedModeIRMethod.java:128)", "org.jruby.internal.runtime.methods.MixedModeIRMethod.call(MixedModeIRMethod.java:115)", "org.jruby.ir.targets.indy.InvokeSite.fail(InvokeSite.java:777)", "org.jruby.ir.targets.indy.InvokeSite.fail(InvokeSite.java:809)", "usr.share.logstash.logstash_minus_core.lib.logstash.pipeline_action.stop_and_delete.RUBY$block$execute$1(/usr/share/logstash/logstash-core/lib/logstash/pipeline_action/stop_and_delete.rb:30)", "usr.share.logstash.logstash_minus_core.lib.logstash.pipelines_registry.RUBY$method$terminate_pipeline$0(/usr/share/logstash/logstash-core/lib/logstash/pipelines_registry.rb:192)", "usr.share.logstash.logstash_minus_core.lib.logstash.pipelines_registry.RUBY$method$terminate_pipeline$0$__VARARGS__(/usr/share/logstash/logstash-core/lib/logstash/pipelines_registry.rb:184)", "org.jruby.internal.runtime.methods.CompiledIRMethod.call(CompiledIRMethod.java:139)", "org.jruby.internal.runtime.methods.MixedModeIRMethod.call(MixedModeIRMethod.java:112)", "org.jruby.ir.targets.indy.InvokeSite.performIndirectCall(InvokeSite.java:725)", "org.jruby.ir.targets.indy.InvokeSite.invoke(InvokeSite.java:657)", "usr.share.logstash.logstash_minus_core.lib.logstash.pipeline_action.stop_and_delete.RUBY$method$execute$0(/usr/share/logstash/logstash-core/lib/logstash/pipeline_action/stop_and_delete.rb:29)", "usr.share.logstash.logstash_minus_core.lib.logstash.pipeline_action.stop_and_delete.RUBY$method$execute$0$__VARARGS__(/usr/share/logstash/logstash-core/lib/logstash/pipeline_action/stop_and_delete.rb:28)", "org.jruby.internal.runtime.methods.CompiledIRMethod.call(CompiledIRMethod.java:139)", "org.jruby.internal.runtime.methods.MixedModeIRMethod.call(MixedModeIRMethod.java:112)", "org.jruby.ir.targets.indy.InvokeSite.performIndirectCall(InvokeSite.java:735)", "org.jruby.ir.targets.indy.InvokeSite.invoke(InvokeSite.java:657)", "usr.share.logstash.logstash_minus_core.lib.logstash.agent.RUBY$block$converge_state$2(/usr/share/logstash/logstash-core/lib/logstash/agent.rb:386)", "org.jruby.runtime.CompiledIRBlockBody.callDirect(CompiledIRBlockBody.java:141)", "org.jruby.runtime.IRBlockBody.call(IRBlockBody.java:64)", "org.jruby.runtime.IRBlockBody.call(IRBlockBody.java:58)", "org.jruby.runtime.Block.call(Block.java:144)", "org.jruby.RubyProc.call(RubyProc.java:352)", "org.jruby.internal.runtime.RubyRunnable.run(RubyRunnable.java:111)", "java.base/java.lang.Thread.run(Thread.java:840)"]}