plugin fails to shutdown due to unable to finish periodic runner in codec

[kaisecheng]

file-input fails to shutdown randomly. Logstash has received SIGTERM but read process does not stop.

There is a dead lock between inputworker thread and shutdown thread. Watcher quit is not able to turn on because IdentityMapCodec fails to finish stop().

	shutdown thread	inputworker thread
t1	LogStash::Inputs::File.stop()
t2	LogStash::Codecs::IdentityMapCodec.close()
t3	cleaner.stop()
t4	@running.make_false
t5		LogStash::Inputs::FileListener.accept()
t6		LogStash::Codecs::IdentityMapCodec#record_codec_usage()
t7		cleaner.start()
t8		@running.make_true (initialize again due to shutdown thread made false)
t9		@thread = Thread.start (spawn new thread)
t10		@thread never end...
t11	while @thread.alive? (keep spinning ) blocking LogStash::Inputs::File.stop() make quit to true
t12		file input process next line as quit is false. LogStash::Inputs::FileListener.accept()

Steps to reproduce:

1. Delete all sincedb file
2. run a script to generate test files

i=1
j=1
path="/tmp/loop"
while true; do
   echo -e "${i}" >> "$path/iii-${j}.txt"
   echo -e "${i}" >> "$path/jjj-${j}.txt"
   ((i++))
   ((j++))

   if [ "$i" -ge 30 ]; then
       ((j=1))
       ((i=1))
   fi
done

3. Start logstash

- pipeline.id: pipeI
  queue.type: persisted
  pipeline.workers: 11
  pipeline.batch.size: 300
  config.string: |
    input {
      file {
        path => "/tmp/iii-*.txt"
        sincedb_path => "/tmp/sincedbI.db"
        mode => "read"
        file_completed_action => "log_and_delete"
        file_completed_log_path => "/tmp/processeI.log"
      }
    }
    
    output {
      stdout { }
    }

- pipeline.id: pipeJ
  pipeline.workers: 2
  config.string: |
    input {
       file {
             path => "/tmp/jjj-*.txt"
             sincedb_path => "/tmp/sincedbJ.db"
             mode => "read"
             file_completed_action => "log_and_delete"
             file_completed_log_path => "/tmp/processeJ.log"
       }
     }
    
     output {
      stdout { }
     }

4. Send SIGTERM to logstash jps | grep Logstash | cut -d' ' -f1 | xargs kill -15
5. If the issue didn’t happen, start over again

Provide logs (if relevant):