Plugin stopping pipeline in logstash v8.9.0

[michael-doubez]

Upgrading to logstash v8.9.0 (opensearch flavour) and using OpenJDK17, the plugin seems to work but crashes on some FIX messages.

Environment

Linux REDHAT
logstash 8.9.0
jruby 9.3.10.0 (2.6.8) 2023-02-01 107b2e6697 OpenJDK 64-Bit Server VM 17.0.4.1+1 on 17.0.4.1+1 +indy +jit [x86_64-linux]
java 17.0.4.1 (Eclipse Adoptium)
jvm OpenJDK 64-Bit Server VM / 17.0.4.1+1

Version isntalled is latest (0.3.3)

$ ./bin/logstash-plugin list --installed logstash-filter-fix_protocol
logstash-filter-fix_protocol

$ cat ./vendor/bundle/jruby/2.6.0/gems/logstash-filter-fix_protocol-0.3.3/logstash-filter-fix_protocol.gemspec
# coding: utf-8
lib = File.expand_path('../lib', __FILE__)
$LOAD_PATH.unshift(lib) unless $LOAD_PATH.include?(lib)
require 'logstash/filters/version'

Gem::Specification.new do |s|
  s.name          = "logstash-filter-fix_protocol"
  s.version       = "0.3.3"
  s.authors       = ["Connamara Systems"]
  s.email         = ["[email protected]"]

  s.summary       = "FIX Protocol Logstash Filter"
  s.description   = "Put your financial application logs to work with logstash FIX filtering"
  s.homepage      = "https://github.com/connamara/logstash-filter-fix_protocol"
  s.licenses      = ['Apache License (2.0)']

  s.files         = Dir['lib/**/*','vendor/**/*','*.gemspec','*.md','CONTRIBUTORS','Gemfile','LICENSE.txt','QUICKFIX_LICENSE.txt','NOTICE.TXT', 'spec/**/*', 'features/**/*']

  s.test_files    = s.files.grep(%r{^(spec|features)/})

  s.require_paths = ["lib"]

  s.metadata = { "logstash_plugin" => "true", "logstash_group" => "filter" }

  if Logstash::VERSION == '5.x'
    s.add_runtime_dependency 'logstash-core', '>= 5.0.0'
  elsif Logstash::VERSION == '2.x'
    s.add_runtime_dependency 'logstash-core', '>= 2.0.0.beta2', '< 3.0.0'
  else
    raise "Invalid Logstash::VERSION - should be 2x or 5x located in `/lib/logstash/filters/version`"
  end
  s.add_runtime_dependency "logstash-input-generator"
  s.add_runtime_dependency "activesupport"
  s.add_runtime_dependency "quickfix-jruby", '~> 1.6', '>= 1.6.5'

  s.add_development_dependency "logstash-devutils"
  s.add_development_dependency "bundler", "~> 1.8"
  s.add_development_dependency "rake", "~> 10.0"
  s.add_development_dependency "rspec"
  s.add_development_dependency "pry"
end

Log

[2023-08-25T16:22:22,799][ERROR][logstash.javapipeline    ][fix-decoding] Pipeline worker error, the pipeline will be stopped
  {
    :pipeline_id=>"fix-decoding",
    :error=>"Cannot invoke \"quickfix.Group.setField(quickfix.StringField)\" because \"group\" is null",
    :exception=>Java::JavaLang::NullPointerException,
    :backtrace=>[
        "quickfix.Message.checkFieldValidation(Message.java:662)",
        "quickfix.Message.parseGroup(Message.java:633)",
         "quickfix.Message.parseBody(Message.java:559)",
         "quickfix.Message.parse(Message.java:470)",
         "quickfix.Message.fromString(Message.java:453)",
         "quickfix.Message.<init>(Message.java:102)", "rubyobj.LogStash.Filters.FixMessage.<init>(/.../logstash/vendor/bundle/jruby/2.6.0/gems/logstash-filter-fix_protocol-0.3.3/lib/logstash/filters/fix_message.rb:8)",
         "rubyobj.LogStash.Filters.FixMessage.<init>(/..../logstash/vendor/bundle/jruby/2.6.0/gems/logstash-filter-fix_protocol-0.3.3/lib/logstash/filters/fix_message.rb:8)",
         "jdk.internal.reflect.GeneratedConstructorAccessor89.newInstance(Unknown Source)",
         "java.base/jdk.internal.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)",
         "java.base/java.lang.reflect.Constructor.newInstanceWithCaller(Constructor.java:499)",
         "java.base/java.lang.reflect.Constructor.newInstance(Constructor.java:480)",
         "org.jruby.java.proxies.ConcreteJavaProxy$NewMethodReified.call(ConcreteJavaProxy.java:303)",
         "org.jruby.internal.runtime.methods.DynamicMethod.call(DynamicMethod.java:238)",
         "org.jruby.internal.runtime.methods.DynamicMethod.call(DynamicMethod.java:234)",
         "...logstash.vendor.bundle.jruby.$2_dot_6_dot_0.gems.logstash_minus_filter_minus_fix_protocol_minus_0_dot_3_dot_3.lib.logstash.filters.fix_protocol.RUBY$method$filter$0(/.../logstash/vendor/bundle/jruby/2.6.0/gems/logstash-filter-fix_protocol-0.3.3/lib/logstash/filters/fix_protocol.rb:46)",
         "org.jruby.internal.runtime.methods.CompiledIRMethod.call(CompiledIRMethod.java:165)",
         "org.jruby.internal.runtime.methods.MixedModeIRMethod.call(MixedModeIRMethod.java:185)",
         "org.jruby.ir.targets.indy.InvokeSite.fail(InvokeSite.java:278)",
         "...logstash.filters.base.RUBY$method$do_filter$0(/.../logstash/logstash-core/lib/logstash/filters/base.rb:159)",
         "org.jruby.internal.runtime.methods.CompiledIRMethod.call(CompiledIRMethod.java:165)",
         "org.jruby.internal.runtime.methods.MixedModeIRMethod.call(MixedModeIRMethod.java:185)",
         "org.jruby.ir.targets.indy.InvokeSite.fail(InvokeSite.java:278)",
         "...logstash.filters.base.RUBY$block$multi_filter$1(/.../logstash/logstash-core/lib/logstash/filters/base.rb:178)",
         "org.jruby.runtime.CompiledIRBlockBody.yieldDirect(CompiledIRBlockBody.java:151)",
         "org.jruby.runtime.BlockBody.yield(BlockBody.java:106)", "org.jruby.runtime.Block.yield(Block.java:188)",
         "org.jruby.RubyArray.each(RubyArray.java:1865)",
         "...logstash.filters.base.RUBY$method$multi_filter$0(/.../logstash/filters/base.rb:175)",
         "org.jruby.internal.runtime.methods.CompiledIRMethod.call(CompiledIRMethod.java:165)",
         "org.jruby.internal.runtime.methods.MixedModeIRMethod.call(MixedModeIRMethod.java:185)",
         "org.jruby.internal.runtime.methods.DynamicMethod.call(DynamicMethod.java:218)",
         "org.logstash.config.ir.compiler.FilterDelegatorExt.doMultiFilter(FilterDelegatorExt.java:128)",
         "org.logstash.config.ir.compiler.AbstractFilterDelegatorExt.lambda$multiFilter$0(AbstractFilterDelegatorExt.java:133)",
         "org.logstash.instrument.metrics.timer.ConcurrentLiveTimerMetric.time(ConcurrentLiveTimerMetric.java:47)",
         "org.logstash.config.ir.compiler.AbstractFilterDelegatorExt.multiFilter(AbstractFilterDelegatorExt.java:133)",
         "org.logstash.generated.CompiledDataset3.compute(Unknown Source)",
         "org.logstash.generated.CompiledDataset11.compute(Unknown Source)",
         "org.logstash.generated.CompiledDataset3.compute(Unknown Source)",
         "org.logstash.config.ir.CompiledPipeline$CompiledUnorderedExecution.compute(CompiledPipeline.java:347)",
         "org.logstash.config.ir.CompiledPipeline$CompiledUnorderedExecution.compute(CompiledPipeline.java:341)",
         "org.logstash.execution.ObservedExecution.lambda$compute$0(ObservedExecution.java:17)",
          "org.logstash.execution.WorkerObserver.lambda$observeExecutionComputation$0(WorkerObserver.java:39)",
         "org.logstash.instrument.metrics.timer.ConcurrentLiveTimerMetric.time(ConcurrentLiveTimerMetric.java:47)",
         "org.logstash.execution.WorkerObserver.lambda$executeWithTimers$1(WorkerObserver.java:50)",
         "org.logstash.instrument.metrics.timer.ConcurrentLiveTimerMetric.time(ConcurrentLiveTimerMetric.java:47)",
         "org.logstash.execution.WorkerObserver.executeWithTimers(WorkerObserver.java:50)",
         "org.logstash.execution.WorkerObserver.observeExecutionComputation(WorkerObserver.java:38)",
         "org.logstash.execution.ObservedExecution.compute(ObservedExecution.java:17)",
         "org.logstash.execution.WorkerLoop.abortableCompute(WorkerLoop.java:113)",
         "org.logstash.execution.WorkerLoop.run(WorkerLoop.java:86)",
         "jdk.internal.reflect.GeneratedMethodAccessor33.invoke(Unknown Source)",
         "java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)",
         "java.base/java.lang.reflect.Method.invoke(Method.java:568)",
         "org.jruby.javasupport.JavaMethod.invokeDirectWithExceptionHandling(JavaMethod.java:442)",
         "org.jruby.javasupport.JavaMethod.invokeDirect(JavaMethod.java:306)",
         "org.jruby.java.invokers.InstanceMethodInvoker.call(InstanceMethodInvoker.java:32)",
       "...logstash.java_pipeline.RUBY$block$start_workers$1(/.../logstash/logstash-core/lib/logstash/java_pipeline.rb:304)",
         "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:143)", "org.jruby.RubyProc.call(RubyProc.java:309)",
         "org.jruby.internal.runtime.RubyRunnable.run(RubyRunnable.java:107)",
         "java.base/java.lang.Thread.run(Thread.java:833)"],
    :thread=>"#<Thread:0x139a6016@/.../logstash/logstash-core/lib/logstash/java_pipeline.rb:134 sleep>"
}