in_tail: Care about concurrent key deletion

Signed-off-by: Daijiro Fukuda <fukuda@clear-code.com>

lib/fluent/plugin/in_tail.rb

@@ -618,6 +618,8 @@ def detach_watcher_after_rotate_wait(tw, ino)
 
       return if @tails_rotate_wait[tw]
 
+      @tails_rotate_wait[tw] = { ino: ino, timer: nil }
+
       if throttling_is_enabled?(tw)
         # When the throttling feature is enabled, it might not reach EOF yet.
         # Should ensure to read all contents before closing it, with keeping throttling.
@@ -630,15 +632,17 @@ def detach_watcher_after_rotate_wait(tw, ino)
             detach_watcher(tw, ino)
           end
         end
-        @tails_rotate_wait[tw] = { ino: ino, timer: timer }
       else
         # when the throttling feature isn't enabled, just wait @rotate_wait
         timer = timer_execute(:in_tail_close_watcher, @rotate_wait, repeat: false) do
           @tails_rotate_wait.delete(tw)
           detach_watcher(tw, ino)
         end
-        @tails_rotate_wait[tw] = { ino: ino, timer: timer }
       end
+
+      # To avoid conflict with timer execution (It deletes the key concurrently)
+      v = @tails_rotate_wait[tw]
+      v[:timer] = timer if v
     end
 
     def flush_buffer(tw, buf)